hypothesis/h — pull request #1695 by nickstenning (2014-11-18 12:26:54+00:00)
Title: Queued streamer

This PR makes it possible for multiple worker processes to receive the streamer events by passing them through an external queue.

As before, an `h.events.AnnotationEvent` subscriber is responsible for serializing events. It now publishes them to an external NSQ topic. The details of connection to NSQ are managed by the `h.queue` module.

When the streamer module is included by pyramid, it will start up a pair of greenlets which are responsible for deserializing and handling the events pulled from the NSQ topic. One of these opens a persistent TCP connection to NSQ and receives messages, pushing them onto an internal queue structure. The second reads from the internal queue structure, and broadcasts the events to active websocket sessions as appropriate.

A final point of explanation: NSQ sorts messages into "topics", which serve as a named queue, and "channels", which are associated with a specific topic. Topics and channels differ thus:

- each message sent to a topic is copied to every channel (i.e. fanout)
- if multiple clients connect to the same channel, incoming messages will be shared between those clients in a round-robin (i.e. they are worker queues)

As a result, each web worker connects to NSQ using a unique channel name, ensuring that each queue reader gets copies of every message published to the topic.

At the moment, we use ephemeral channels in order to avoid messages backing up in NSQ for web workers that have long since ceased to exist. This does mean that in the event of a network partition between web worker and NSQ, some messages may be lost and not relayed to websocket clients connected to that worker.

There are a couple of outstanding questions/issues in my head:

- [ ] Starting multiple workers doesn't actually work at the moment, because each one tries to update the elasticsearch mapping at the same time. We could solve this by setting `preload_app = True` in `gunicorn.conf.py`, but then we'd need to move the setup of the NSQ readers into a gunicorn `post_fork` rather than in the module `includeme`. Most different options here seem a bit icky.
- [ ] As it stands, this now requires you to run NSQ in development if you want realtime support. So there are two subquestions:
- [ ] Should `h.feature.streamer` default to `False` in `development.ini`?
- [ ] I could fairly easily turn `h.queue` into a package with two modules: `h.queue.nsq` and `h.queue.local` which could be swapped in `pyramid.include`, and default to a local queue implementation in development. Should I do this?
- [ ] There seem to be issues (not affecting functionality, as far as I can tell) with gunicorn's auto-reload support for paste. [See here](https://gist.github.com/nickstenning/97f68aef3643ce617a40).