Commit d83aadd0 authored by Roman Haefeli's avatar Roman Haefeli
Browse files

One2ManyMo: add logger and catch all

parent 15d82fcd
......@@ -5,6 +5,7 @@
udp_one2manymo: for 1-to-N connections
"""
import logging
import socket
import sys
import threading
......@@ -17,7 +18,7 @@ class One2ManyMoProxy(threading.Thread):
Different ports are used for source and sink clients.
"""
def __init__(self, listen_port=None, send_port=None, listen_address='0.0.0.0', timeout=10):
def __init__(self, listen_port=None, send_port=None, listen_address='0.0.0.0', timeout=10, logger=None):
super(One2ManyMoProxy, self).__init__()
for port in [listen_port, send_port]:
if not isinstance(port, int) or not 1024 <= port <= 65535:
......@@ -36,44 +37,51 @@ class One2ManyMoProxy(threading.Thread):
except socket.error as msg:
raise
self.kill_signal = False
self.logger = logger
# key of dict is sink_client's (address, port) tuple
self.sink_clients = {}
self.timeout = timeout
def run(self):
while not self.kill_signal:
# handle incoming packets from sink clients
while True:
try:
_trash, sink_addr = self.sink.recvfrom(65536)
except BlockingIOError:
break
self.sink_clients[sink_addr] = time.time()
# handle incoming packets from source client
try:
data, addr = self.source.recvfrom(65536)
except socket.timeout:
continue
# handle incoming packets from sink clients
while True:
try:
_trash, sink_addr = self.sink.recvfrom(65536)
except BlockingIOError:
break
self.sink_clients[sink_addr] = time.time()
# handle incoming packets from source client
try:
data, addr = self.source.recvfrom(65536)
except socket.timeout:
continue
# remove expired clients from sink_clients
for client, prev_ts in list(self.sink_clients.items()):
if (prev_ts + self.timeout) < time.time():
del self.sink_clients[client]
# remove expired clients from sink_clients
for client, prev_ts in list(self.sink_clients.items()):
if (prev_ts + self.timeout) < time.time():
del self.sink_clients[client]
# send data to remaining sink_clients
for client in self.sink_clients.keys():
self.sink.sendto(data, client)
# send data to remaining sink_clients
for client in self.sink_clients.keys():
self.sink.sendto(data, client)
except:
self.logger.exception('Oops, something went wrong!', extra={'stack': True})
def stop(self):
self.kill_signal = True
self.join()
def main():
logger = logging.getLogger()
handler = logging.StreamHandler(sys.stderr)
logger.addHandler(handler)
try:
source_port = int(sys.argv[1])
sink_port = source_port + 1
proxy = One2ManyMoProxy(listen_port=source_port, send_port=sink_port)
proxy = One2ManyMoProxy(listen_port=source_port, send_port=sink_port, logger=logger)
proxy.start()
proxy.join()
except KeyboardInterrupt:
......
Supports Markdown
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