added a notify system for consumers per queue

Scratch that last part, we do handle it.