nytlabs/streamtools — pull request #290 by nikhan (2014-03-11 17:59:06+00:00)
Title: fixes for NSQ and all blocks in general

This patch fixes #283, #282, #132.

It also works towards #265, and may help #284.

What was wrong with NSQ:

*fromNSQ spawned multiple readers and never killed them off:*
the `reader, err :=` assignment operator meant that a new reader was created every time, despite there being a reader var in the run()'s scope.
*fromNSQ didn't check to see if a reader was running*

What was happening with the two simultaneous NSQ streams locking a routine and subsequently killing the system:
Previously we had haphhazardly alleviated pressure between the Run() and the BlockRoutine by placing some small buffered chans between the two. They were badly defined however, and never actually used.
see:
https://github.com/nikhan/streamtools/compare/nsq_ultireader?expand=1#diff-63cecc398cfb5da622d56fd332190ef2L83 vs
https://github.com/nikhan/streamtools/compare/nsq_ultireader?expand=1#diff-63cecc398cfb5da622d56fd332190ef2R101

The patch in blocks.go makes sure that each inchan added to the inroutes has a buffered chan of length 1000. If we can no longer send on this chan, then we create a new go routine that sits and waits to send the message. This at least offloads overflows to the cpu/memory instead of blocking which is what we did before.

I expect in the future, we may have to deal with orphan messages in a graceful way. Right now all those msg routines will insist on being delivered, even if the Run() gets closed before they can be sent. This means that we may need to give them an expiration -- see this https://github.com/nytlabs/streamtools/blob/master/st/library/fromSQS.go#L113.