Bug fix with transport and listener, actor implementation

community
Steve L. Nyemba 8 years ago
parent 357be165e2
commit e3a0e6e8b0

@ -14,9 +14,11 @@ from threading import Thread
import os import os
import subprocess import subprocess
from monitor import ProcessCounter from monitor import ProcessCounter
from utils.transport import QueueListener
class Actor(Thread): class Actor(Thread):
def __init__(self,config): def __init__(self,config):
Thread.__init__(self) Thread.__init__(self)
self.config = config
self.items = [] self.items = []
def init(self,litems): def init(self,litems):
self.items = litems self.items = litems
@ -35,13 +37,14 @@ class Actor(Thread):
def run(self): def run(self):
info = {} info = {}
info['exchange'] = self.config['organization'] host = self.config['api']
info['uid'] = self.config['id'] uid = self.config['key']
info['qid'] = ['action'] qid = self.config['id']
qlistener = QueueListener(info) qlistener = QueueListener(qid=qid,uid=uid,host=host)
qlistener.callback = self.callback
qlistener.read() qlistener.read()
r = [self.process(item) for item in self.litems] r = [self.process(item) for item in self.items]
class Kill(Actor): class Kill(Actor):
def __init__(self,config): def __init__(self,config):
Actor.__init__(self,config) Actor.__init__(self,config)
@ -62,3 +65,7 @@ class Start(Actor):
class Alert(Actor): class Alert(Actor):
pass pass
config = {"id":"demo","key":"[0v8]-247&7!v3","api":"localhost"}
actor = Kill(config)
actor.start()

@ -359,8 +359,7 @@ class QueueReader(MessageQueue,Reader):
self.channel = self.connection.channel() self.channel = self.connection.channel()
self.channel.exchange_declare(exchange=self.uid,type='direct',durable=True) self.channel.exchange_declare(exchange=self.uid,type='direct',durable=True)
self.info = self.channel.queue_ self.info = self.channel.queue_declare(queue=qid,durable=True)
declare(queue=qid,durable=True)
@ -416,15 +415,20 @@ declare(queue=qid,durable=True)
return self.data return self.data
class QueueListener(QueueReader): class QueueListener(QueueReader):
def init(self,qid,callback): 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)
self.channel = self.connection.channel() self.channel = self.connection.channel()
self.channel.exchange_declare(exchange=self.uid,type='fanout') self.channel.exchange_declare(exchange=self.uid,type='fanout')
self.info = self.channel.queue_declare(queue=qid,exclusive=True) self.info = self.channel.queue_declare(queue=qid,exclusive=True)
self.channel.queue_bind(exchange=self.uid,queue=self.info.method.queue,no_ack=True) self.channel.queue_bind(exchange=self.uid,queue=self.info.method.queue)
self.callback = callback #self.callback = callback
def read(self):
self.init(self.qid)
self.channel.basic_consume(self.callback,queue=self.qid,no_ack=True);
self.channel.start_consuming()
""" """
This class is designed to write output as sql insert statements This class is designed to write output as sql insert statements

Loading…
Cancel
Save