added a notify system for consumers per queue

We should break this out into its own function, it's kind of a gigantic thing to be inlining.