gistfile1.py

from tornado.platform.twisted import TwistedIOLoop
import tornado.ioloop
from twisted.internet import defer, reactor
from twisted.application import service

import nsq
from nsq import Writer, Error
from nsq.async import AsyncConn


application = service.Application('Storage Job Service')


def finish_message(msg):
    print "finish", msg
    return msg.finish()

def store_message(msg):
    print "store", msg
    return msg

def process_message(msg):
    print "process", msg
    msg.enable_async()
    d = defer.Deferred()
    d.addCallback(store_message)
    d.addCallback(finish_message)
    d.callback(msg)



if __name__ == '__main__':
    from twisted.python import log
    import sys

    log.startLogging(sys.stderr)
    TwistedIOLoop().install()
    r = nsq.Reader(message_handler=process_message,
        lookupd_http_addresses=['http://127.0.0.1:4161'],
        topic='test', channel='dork', max_in_flight=9)

    reactor.run()