async_nsq.py

from twisted.internet import reactor, defer                                                                            
from tornado.platform.twisted import TwistedIOLoop                                                                     
import nsq                                                                                                             
from nsq import Writer, Error                                                                                          
from nsq.async import AsyncConn                                                                                        
                                                                                                                       
# This doesn't work - the client errors with a failed send, and nsqd (when                                             
# invoked with nsqd --verbose, outputs:                                                                                
#    2014/06/01 21:26:09 TCP: new client(127.0.0.1:53443)                                                              
#    2014/06/01 21:26:09 ERROR: failed to read protocol version - EOF                                                  
# the messing around with the deferred is so this function still returns an                                            
# object which represents the lifecycle of the work it's trying to do                                                  
def main():                                                                                                            
    nsq = Writer(['127.0.0.1:4150'])                                                                                   
    deferred = defer.Deferred()                                                                                        
    def done(conn, data):                                                                                              
        f = deferred.callback                                                                                          
        if isinstance(data, Error):                                                                                    
            f = deferred.errback                                                                                       
        f((conn, data))                                                                                                
    nsq.pub('topic', 'message', callback=done)                                                                         
    return deferred                                                                                                    
                                                                                                                       
# This works, but it's sending commands directly, with no opportunity to figure                                        
# out whether the command succeeded or failed.  I don't really know how this                                           
# could be made to work on the subscribe side.                                                                         
#def main():                                                                                                           
#    c = AsyncConn('127.0.0.1', 4150)                                                                                  
#    c.connect()                                                                                                       
#                                                                                                                      
#    deferred = defer.Deferred()                                                                                       
#    def cb(*args, **kwargs):                                                                                          
#        c.send(nsq.pub('topic', 'message'))                                                                           
#        deferred.callback((args, kwargs))                                                                             
#                                                                                                                      
#    c.on('ready', cb)                                                                                                 
#    return deferred                                                                                                   
                                                                                                                       
if __name__ == '__main__':                                                                                             
    from twisted.python import log                                                                                     
    import sys                                                                                                         
                                                                                                                       
    log.startLogging(sys.stdout)                                                                                       
                                                                                                                       
    TwistedIOLoop().install()                                                                                          
    reactor.callWhenRunning(                                                                                           
        lambda: main().addCallbacks(log.msg, log.err).addCallback(lambda _: reactor.stop()))                           
    reactor.run()