Skip to content
Snippets Groups Projects
Commit 8c8c404f authored by Manuela Kuhn's avatar Manuela Kuhn
Browse files

finished coordinator class

parent 9a8d8d68
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
ZeroMQTunnel/receiver.py
+ 67
− 71
View file @ 8c8c404f
......@@ -23,10 +23,10 @@ class Coordinator:
zmqDataStreamPort = None
zmqLiveViewerIp = None
zmqLiveViewerPort = None
liveViewerExchangeIp = "127.0.0.1"
liveViewerExchangePort = "6072"
receiverExchangeIp = "127.0.0.1"
receiverExchangePort = "6073"
receiverExchangePort = "6072"
liveViewerExchangeIp = "127.0.0.1"
liveViewerExchangePort = "6073"
ringBuffer = []
maxRingBufferSize = 200
timeToWaitForRingBuffer = 2
......@@ -36,6 +36,12 @@ class Coordinator:
receiverThread = None
liveViewerThread = None
# sockets
receiverExchangeSocket = None
liveViewerExchangeSocket = None
zmqliveViewerSocket = None
def __init__(self, outputDir, zmqDataStreamPort, zmqDataStreamIp, zmqLiveViewerPort, zmqLiveViewerIp, context = None):
self.outputDir = outputDir
self.zmqDataStreamIp = zmqDataStreamIp
......@@ -52,28 +58,32 @@ class Coordinator:
self.zmqContext = context or zmq.Context()
# create sockets
self.receiverExchangeSocket = self.zmqContext.socket(zmq.PAIL)
self.receiverExchangeSocket = self.zmqContext.socket(zmq.PAIR)
connectionStrReceiverExchangeSocket = "tcp://" + self.receiverExchangeIp + ":%s" % self.receiverExchangePort
self.zmqSocket.bind(connectionStrReceiverExchangeSocket)
self.receiverExchangeSocket.bind(connectionStrReceiverExchangeSocket)
self.log.debug("receiverExchangeSocket started (bind) for '" + connectionStrReceiverExchangeSocket + "'")
# TODO use this to communicate with live viewer class
self.liveViewerExchangeSocket = self.zmqContext.socket(zmq.PAIR)
connectionStrLiveViewerExchangeSocket = "tcp://" + self.liveViewerExchangeIp + ":%s" % self.liveViewerExchangePort
self.liveViewerExchangeSocket.bind(connectionStrLiveViewerExchangeSocket)
self.log.debug("liveViewerExchangeSocket started (bind) for '" + connectionStrLiveViewerExchangeSocket + "'")
# create socket for live viewer
self.zmqliveViewerSocket = self.zmqContext.socket(zmq.REP)
connectionStrLiveViewerSocket = "tcp://" + self.zmqLiveViewerIp + ":%s" % self.zmqLiveViewerPort
self.zmqliveViewerSocket.bind(connectionStrLiveViewerSocket)
self.log.debug("zmqLiveViewerSocket started (bind) for '" + connectionStrLiveViewerSocket + "'")
self.poller = zmq.Poller()
self.poller.register(self.receiverExchangeSocket, zmq.POLLIN)
self.poller.register(self.liveViewerExchangeSocket, zmq.POLLIN)
# self.poller.register(self.liveViewerExchangeSocket, zmq.POLLIN)
self.poller.register(self.zmqliveViewerSocket, zmq.POLLIN)
# start file receiver
self.receiverThread = threading.Thread(target=FileReceiver, args=(self.outputDir, self.zmqDataStreamPort, self.zmqDataStreamIp, self.zmqLiveViewerPort, self.zmqLiveViewerIp))
self.receiverThread.start()
# thread to communicate with live viewer
self.liveViewerThread = threading.Thread(target=LiveViewer)
self.liveViewerThread.start()
# self.liveViewerThread = threading.Thread(target=LiveViewer)
# self.liveViewerThread.start()
# initialize ring buffer
......@@ -93,16 +103,13 @@ class Coordinator:
try:
self.log.info("Start communication")
self.communicateWithThreads()
self.communicate()
self.log.info("Stopped communication.")
except Exception, e:
trace = traceback.format_exc()
self.log.info("Unkown error state. Shutting down...")
self.log.debug("Error was: " + str(e))
self.log.debug("Closing socket")
self.receiverExchangeSocket.close()
self.liveViewerExchangeSocket.close()
self.log.info("Quitting.")
......@@ -112,7 +119,7 @@ class Coordinator:
return logger
def communicateWithThreads(self):
def communicate(self):
should_continue = True
while should_continue:
......@@ -122,18 +129,24 @@ class Coordinator:
message = self.receiverExchangeSocket.recv()
self.log.debug("Recieved control command: %s" % message )
if message == "Exit":
self.log.debug("Recieved exit command, liveViewer thread will stop recieving messages")
self.log.debug("Recieved exit command, coordinator thread will stop recieving messages")
should_continue = False
self.liveViewerSocket.send("Exit")
break
elif message.startswith("AddFile"):
self.log.debug("Received AddFile command")
# add file to ring buffer
filename, fileModTime = message[7:].split(", ")
addFileToRingBuffer(filename, fileModTime)
if self.liveViewerExchangeSocket in socks and socks[self.liveViewerExchangeSocket] == zmq.POLLIN:
message = self.liveViewerExchangeSocket.recv()
self.log.debug("Call for next file... ")
splittedMessage = message[7:].split(", ")
filename = splittedMessage[0]
fileModTime = splittedMessage[1]
self.log.debug("Send new file to ring buffer: " + str(filename) + ", " + str(fileModTime))
self.addFileToRingBuffer(filename, fileModTime)
# if self.liveViewerExchangeSocket in socks and socks[self.liveViewerExchangeSocket] == zmq.POLLIN:
# message = self.liveViewerExchangeSocket.recv()
if self.zmqliveViewerSocket in socks and socks[self.zmqliveViewerSocket] == zmq.POLLIN:
message = self.zmqliveViewer.recv()
self.log.debug("Call for next file... " + message)
# send first element in ring buffer to live viewer (the path of this file is the second entry)
if self.ringBuffer:
answer = self.ringBuffer[0][1]
......@@ -142,10 +155,22 @@ class Coordinator:
print answer
try:
self.liveViewerExchangeSocket.send(answer)
# self.liveViewerExchangeSocket.send(answer)
self.zmqliveViewerSocket.send(answer)
except zmq.error.ContextTerminated:
break
self.log.debug("sending exit signal to thread...")
# self.liveViewerExchangeSocket.send("Exit")
self.zmqliveViewerSocket.send("Exit")
# give the signal time to arrive
time.sleep(0.1)
self.log.debug("Closing socket")
self.receiverExchangeSocket.close()
self.liveViewerExchangeSocket.close()
self.zmqliveViewerSocket.close()
def addFileToRingBuffer(self, filename, fileModTime):
# prepend file to ring buffer and restore order
......@@ -163,17 +188,13 @@ class Coordinator:
class FileReceiver:
zmqContext = None
liveViewerZmqContext = None
outputDir = None
zqmDataStreamIp = None
zmqDataStreamPort = None
zmqLiveViewerIp = None
zmqLiveViewerPort = None
zmqLiveViewerExchangeIp = "127.0.0.1"
zmqLiveViewerExchangePort = "6072"
ringBuffer = []
maxRingBufferSize = 200
timeToWaitForRingBuffer = 2
exchangeIp = "127.0.0.1"
exchangePort = "6072"
log = None
......@@ -197,35 +218,20 @@ class FileReceiver:
self.log = self.getLogger()
self.log.debug("Init")
# start file receiver
self.receiverThread = threading.Thread(target=Coordinator, args=(self.outputDir, self.zmqDataStreamPort, self.zmqDataStreamIp, self.zmqLiveViewerPort, self.zmqLiveViewerIp))
self.receiverThread.start()
# create pull socket
self.zmqSocket = self.zmqContext.socket(zmq.PULL)
connectionStrZmqSocket = "tcp://" + self.zmqDataStreamIp + ":%s" % self.zmqDataStreamPort
self.zmqSocket.bind(connectionStrZmqSocket)
self.log.debug("zmqSocket started for '" + connectionStrZmqSocket + "'")
self.log.debug("zmqSocket started (bind) for '" + connectionStrZmqSocket + "'")
self.exchangeSocket = self.zmqContext.socket(zmq.PAIR)
connectionStrExchangeSocket = "tcp://" + self.zmqLiveViewerExchangeIp + ":%s" % self.zmqLiveViewerExchangePort
self.exchangeSocket.bind(connectionStrExchangeSocket)
self.log.debug("exchangeSocket started (bind) for '" + connectionStrExchangeSocket + "'")
# thread to communicate with live viewer
self.liveViewerThread = threading.Thread(target=LiveViewer)
self.liveViewerThread.start()
# initialize ring buffer
# get all entries in the directory
# TODO empty target dir -> ringBuffer = []
ringBuffer = (os.path.join(self.outputDir, fn) for fn in os.listdir(self.outputDir))
# get the corresponding stats
ringBuffer = ((os.stat(path), path) for path in ringBuffer)
# leave only regular files, insert modification date
ringBuffer = [[stat[ST_MTIME], path]
for stat, path in ringBuffer if S_ISREG(stat[ST_MODE])]
# sort the ring buffer in descending order (new to old files)
ringBuffer = sorted(ringBuffer, reverse=True)
connectionStrExchangeSocket = "tcp://" + self.exchangeIp + ":%s" % self.exchangePort
self.exchangeSocket.connect(connectionStrExchangeSocket)
self.log.debug("exchangeSocket started (connect) for '" + connectionStrExchangeSocket + "'")
try:
self.log.info("Start receiving new files")
......@@ -248,19 +254,6 @@ class FileReceiver:
return logger
def addFileToRingBuffer(self, filename, fileModTime):
# prepend file to ring buffer and restore order
self.ringBuffer[:0] = [[fileModTime, filename]]
self.ringBuffer = sorted(self.ringBuffer, reverse=True)
# if the maximal size is exceeded: remove the oldest files
if len(self.ringBuffer) > self.maxRingBufferSize:
for mod_time, path in self.ringBuffer[self.maxRingBufferSize:]:
if float(time.time()) - mod_time > self.timeToWaitForRingBuffer:
os.remove(path)
self.ringBuffer.remove([mod_time, path])
def combineMessage(self, zmqSocket):
receivingMessages = True
#save all chunks to file
......@@ -304,9 +297,10 @@ class FileReceiver:
fileModTime = payloadMetadataDict["fileModificationTime"]
self.log.info("New file with modification time " + str(fileModTime) + " received and saved: " + str(filename))
# add to ring buffer
self.log.debug("add file to ring buffer: "+ str(filename) + ", " + str(fileModTime))
self.addFileToRingBuffer(str(filename), fileModTime)
# send the file to the coordinator to add it to the ring buffer
message = "AddFile" + str(filename) + ", " + str(fileModTime)
self.log.debug("Send file to coordinator: " + message )
self.exchangeSocket.send(message)
......@@ -322,6 +316,7 @@ class FileReceiver:
loopCounter+=1
except KeyboardInterrupt:
self.log.debug("Keyboard interrupt detected. Stop receiving.")
continueReceiving = False
break
except:
self.log.error("receive message...failed.")
......@@ -435,6 +430,7 @@ class FileReceiver:
self.log.debug("sending exit signal to thread...")
self.exchangeSocket.send("Exit")
# give the signal time to arrive
time.sleep(0.1)
self.exchangeSocket.close()
self.log.debug("sending exit signal to thread...done")
......@@ -479,7 +475,7 @@ class LiveViewer():
self.liveViewerSocket = self.zmqContext.socket(zmq.REP)
connectionStrLiveViewerSocket = "tcp://" + self.liveViewerIp + ":%s" % self.liveViewerPort
self.liveViewerSocket.bind(connectionStrLiveViewerSocket)
self.log.debug("zmqLiveViewerSocket started for '" + connectionStrLiveViewerSocket + "'")
self.log.debug("zmqLiveViewerSocket started (bind) for '" + connectionStrLiveViewerSocket + "'")
# create socket for message exchange
self.exchangeSocket = self.zmqContext.socket(zmq.PAIR)
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment