hiplab/data-transport
hiplab
/
data-transport
mirror of https://github.com/lnyemba/data-transport
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

bug fix & aliasing

This commit is contained in:
Steve Nyemba 2020-02-05 18:14:30 -06:00
parent c268a117c2
commit a16b969b69
1 changed files with 20 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
transport/queue.py
View File

@ -33,6 +33,9 @@ class MessageQueue:
self.connection = None self.connection = None
self.channel = None self.channel = None
self.name = self.__class__.__name__.lower() if 'name' not in params else 'wtf'
self.credentials= pika.PlainCredentials('guest','guest') self.credentials= pika.PlainCredentials('guest','guest')
if 'username' in params : if 'username' in params :
self.credentials = pika.PlainCredentials( self.credentials = pika.PlainCredentials(
@ -57,6 +60,8 @@ class MessageQueue:
resp = self.connection is not None and self.connection.is_open resp = self.connection is not None and self.connection.is_open
self.close() self.close()
return resp return resp
def finalize(self):
pass
def close(self): def close(self):
if self.connection.is_closed == False : if self.connection.is_closed == False :
self.channel.close() self.channel.close()
@ -81,12 +86,13 @@ class QueueWriter(MessageQueue,Writer):
"""
def write(self,data,_type='text/plain'):
"""
This function writes a stream of data to the a given queue This function writes a stream of data to the a given queue
@param object object to be written (will be converted to JSON) @param object object to be written (will be converted to JSON)
@TODO: make this less chatty @TODO: make this less chatty
""" """
def write(self,data,_type='text/plain'):
# xchar = None # xchar = None
# if 'xchar' in params: # if 'xchar' in params:
# xchar = params['xchar'] # xchar = params['xchar']
@ -207,11 +213,14 @@ class QueueReader(MessageQueue,Reader):
# r[qid].append( self.data) # r[qid].append( self.data)
return self.data return self.data
class QueueListener(QueueReader): class QueueListener(MessageQueue):
""" """
This class is designed to have an active listener (worker) against a specified Exchange/Queue This class is designed to have an active listener (worker) against a specified Exchange/Queue
It is initialized as would any other object and will require a callback function to address the objects returned. It is initialized as would any other object and will require a callback function to address the objects returned.
""" """
def __init__(self,**args):
MessageQueue.__init__(self,**args)
self.listen = self.read
# def init(self,qid): # def init(self,qid):
# properties = pika.ConnectionParameters(host=self.host) # properties = pika.ConnectionParameters(host=self.host)
# self.connection = pika.BlockingConnection(properties) # self.connection = pika.BlockingConnection(properties)
@ -222,11 +231,18 @@ class QueueListener(QueueReader):
# self.channel.queue_bind(exchange=self.exchange,queue=self.info.method.queue,routing_key=qid) # self.channel.queue_bind(exchange=self.exchange,queue=self.info.method.queue,routing_key=qid)
#self.callback = callback #self.callback = callback
def finalize(self,channel,ExceptionReason):
pass
def callback(self,channel,method,header,stream) : def callback(self,channel,method,header,stream) :
raise Exception("....") raise Exception("....")
def read(self): def read(self):
self.init(self.queue) self.init(self.queue)
self.channel.basic_consume(self.queue,self.callback,auto_ack=True); self.channel.basic_consume(self.queue,self.callback,auto_ack=True);
self.channel.start_consuming() self.channel.start_consuming()