<s> from zope . interface import Interface , Attribute <EOL> from txamqp . protocol import AMQClient <EOL> from txamqp . contrib . thrift . transport import TwistedAMQPTransport <EOL> from txamqp . queue import TimeoutDeferredQueue , Closed <EOL> from twisted . internet import defer <EOL> from twisted . python import log <EOL> from thrift . transport import TTransport <EOL> class ThriftAMQClient ( AMQClient ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> AMQClient . __init__ ( self , * args , ** kwargs ) <EOL> if self . check_0_8 ( ) : <EOL> self . replyToField = "<STR_LIT>" <EOL> else : <EOL> self . replyToField = "<STR_LIT>" <EOL> self . thriftBasicReturnQueueLock = defer . DeferredLock ( ) <EOL> self . thriftBasicReturnQueues = { } <EOL> @ defer . inlineCallbacks <EOL> def thriftBasicReturnQueue ( self , key ) : <EOL> yield self . thriftBasicReturnQueueLock . acquire ( ) <EOL> try : <EOL> try : <EOL> q = self . thriftBasicReturnQueues [ key ] <EOL> except KeyError : <EOL> q = TimeoutDeferredQueue ( ) <EOL> self . thriftBasicReturnQueues [ key ] = q <EOL> finally : <EOL> self . thriftBasicReturnQueueLock . release ( ) <EOL> defer . returnValue ( q ) <EOL> @ defer . inlineCallbacks <EOL> def createThriftClient ( self , responsesExchange , serviceExchange , <EOL> routingKey , clientClass , channel = <NUM_LIT:1> , responseQueue = None , iprot_factory = None , <EOL> oprot_factory = None ) : <EOL> channel = yield self . channel ( channel ) <EOL> if responseQueue is None : <EOL> reply = yield channel . queue_declare ( exclusive = True , <EOL> auto_delete = True ) <EOL> responseQueue = reply . queue <EOL> yield channel . queue_bind ( queue = responseQueue , <EOL> exchange = responsesExchange , routing_key = responseQueue ) <EOL> reply = yield channel . basic_consume ( queue = responseQueue ) <EOL> log . msg ( "<STR_LIT>" % responseQueue ) <EOL> thriftClientName = clientClass . __name__ + routingKey <EOL> amqpTransport = TwistedAMQPTransport ( <EOL> channel , serviceExchange , routingKey , clientName = thriftClientName , <EOL> replyTo = responseQueue , replyToField = self . replyToField ) <EOL> if iprot_factory is None : <EOL> iprot_factory = self . factory . iprot_factory <EOL> if oprot_factory is None : <EOL> oprot_factory = self . factory . oprot_factory <EOL> thriftClient = clientClass ( amqpTransport , oprot_factory ) <EOL> queue = yield self . queue ( reply . consumer_tag ) <EOL> d = queue . get ( ) <EOL> d . addCallback ( self . parseClientMessage , channel , queue , thriftClient , <EOL> iprot_factory = iprot_factory ) <EOL> d . addErrback ( self . catchClosedClientQueue ) <EOL> d . addErrback ( self . handleClientQueueError ) <EOL> basicReturnQueue = yield self . thriftBasicReturnQueue ( thriftClientName ) <EOL> d = basicReturnQueue . get ( ) <EOL> d . addCallback ( self . parseClientUnrouteableMessage , channel , <EOL> basicReturnQueue , thriftClient , iprot_factory = iprot_factory ) <EOL> d . addErrback ( self . catchClosedClientQueue ) <EOL> d . addErrback ( self . handleClientQueueError ) <EOL> defer . returnValue ( thriftClient ) <EOL> def parseClientMessage ( self , msg , channel , queue , thriftClient , <EOL> iprot_factory = None ) : <EOL> deliveryTag = msg . delivery_tag <EOL> tr = TTransport . TMemoryBuffer ( msg . content . body ) <EOL> if iprot_factory is None : <EOL> iprot = self . factory . iprot_factory . getProtocol ( tr ) <EOL> else : <EOL> iprot = iprot_factory . getProtocol ( tr ) <EOL> ( fname , mtype , rseqid ) = iprot . readMessageBegin ( ) <EOL> if rseqid in thriftClient . _reqs : <EOL> pass <EOL> else : <EOL> log . msg ( '<STR_LIT>' % ( fname , rseqid , mtype , msg . routing_key , thriftClient , msg . content . body ) ) <EOL> method = getattr ( thriftClient , '<STR_LIT>' + fname ) <EOL> method ( iprot , mtype , rseqid ) <EOL> channel . basic_ack ( deliveryTag , True ) <EOL> d = queue . get ( ) <EOL> d . addCallback ( self . parseClientMessage , channel , queue , thriftClient , <EOL> iprot_factory = iprot_factory ) <EOL> d . addErrback ( self . catchClosedClientQueue ) <EOL> d . addErrback ( self . handleClientQueueError ) <EOL> def parseClientUnrouteableMessage ( self , msg , channel , queue , thriftClient , <EOL> iprot_factory = None ) : <EOL> tr = TTransport . TMemoryBuffer ( msg . content . body ) <EOL> if iprot_factory is None : <EOL> iprot = self . factory . iprot_factory . getProtocol ( tr ) <EOL> else : <EOL> iprot = iprot_factory . getProtocol ( tr ) <EOL> ( fname , mtype , rseqid ) = iprot . readMessageBegin ( ) <EOL> try : <EOL> d = thriftClient . _reqs . pop ( rseqid ) <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> d . errback ( TTransport . TTransportException ( <EOL> type = TTransport . TTransportException . NOT_OPEN , <EOL> message = '<STR_LIT>' <EOL> % ( msg . routing_key , fname ) ) ) <EOL> d = queue . get ( ) <EOL> d . addCallback ( self . parseClientUnrouteableMessage , channel , queue , <EOL> thriftClient , iprot_factory = iprot_factory ) <EOL> d . addErrback ( self . catchClosedClientQueue ) <EOL> d . addErrback ( self . handleClientQueueError ) <EOL> def catchClosedClientQueue ( self , failure ) : <EOL> failure . trap ( Closed ) <EOL> self . handleClosedClientQueue ( failure ) <EOL> def handleClientQueueError ( self , failure ) : <EOL> pass <EOL> def handleClosedClientQueue ( self , failure ) : <EOL> pass <EOL> @ defer . inlineCallbacks <EOL> def createThriftServer ( self , responsesExchange , serviceExchange , <EOL> routingKey , processor , serviceQueue , channel = <NUM_LIT:1> , iprot_factory = None , <EOL> oprot_factory = None ) : <EOL> channel = yield self . channel ( channel ) <EOL> yield channel . channel_open ( ) <EOL> yield channel . exchange_declare ( exchange = serviceExchange , type = "<STR_LIT>" ) <EOL> yield channel . queue_declare ( queue = serviceQueue , auto_delete = True ) <EOL> yield channel . queue_bind ( queue = serviceQueue , exchange = serviceExchange , <EOL> routing_key = routingKey ) <EOL> reply = yield channel . basic_consume ( queue = serviceQueue ) <EOL> queue = yield self . queue ( reply . consumer_tag ) <EOL> d = queue . get ( ) <EOL> d . addCallback ( self . parseServerMessage , channel , responsesExchange , <EOL> queue , processor , iprot_factory = iprot_factory , oprot_factory = oprot_factory ) <EOL> d . addErrback ( self . catchClosedServerQueue ) <EOL> d . addErrback ( self . handleServerQueueError ) <EOL> def parseServerMessage ( self , msg , channel , exchange , queue , processor , <EOL> iprot_factory = None , oprot_factory = None ) : <EOL> deliveryTag = msg . delivery_tag <EOL> try : <EOL> replyTo = msg . content [ self . replyToField ] <EOL> except KeyError : <EOL> replyTo = None <EOL> tmi = TTransport . TMemoryBuffer ( msg . content . body ) <EOL> tr = TwistedAMQPTransport ( channel , exchange , replyTo ) <EOL> if iprot_factory is None : <EOL> iprot = self . factory . iprot_factory . getProtocol ( tmi ) <EOL> else : <EOL> iprot = iprot_factory . getProtocol ( tmi ) <EOL> if oprot_factory is None : <EOL> oprot = self . factory . oprot_factory . getProtocol ( tr ) <EOL> else : <EOL> oprot = oprot_factory . getProtocol ( tr ) <EOL> d = processor . process ( iprot , oprot ) <EOL> channel . basic_ack ( deliveryTag , True ) <EOL> d = queue . get ( ) <EOL> d . addCallback ( self . parseServerMessage , channel , exchange , queue , <EOL> processor , iprot_factory , oprot_factory ) <EOL> d . addErrback ( self . catchClosedServerQueue ) <EOL> d . addErrback ( self . handleServerQueueError ) <EOL> def catchClosedServerQueue ( self , failure ) : <EOL> failure . trap ( Closed ) <EOL> self . handleClosedServerQueue ( failure ) <EOL> def handleServerQueueError ( self , failure ) : <EOL> pass <EOL> def handleClosedServerQueue ( self , failure ) : <EOL> pass <EOL> class IThriftAMQClientFactory ( Interface ) : <EOL> iprot_factory = Attribute ( "<STR_LIT>" ) <EOL> oprot_factory = Attribute ( "<STR_LIT>" ) <EOL> processor = Attribute ( "<STR_LIT>" ) </s>
