Изящно прекращение Autobahn ApplicationRunner (). Run () с использованием сигнала. ⇐ Python

[Anonymous]

Я ранее спрашивал, как запустить несколько экземпляров Autobahn.applicationsession из того же процесса Python, не имея их блокировки.
Проблема была решена, однако я столкнулся с новой проблемой. Я знаю код в ApplicationRunner.run () выходит на keyboardinterrupt , но я не смог получить это правильно.
Пример кода:
class PoloniexSession(ApplicationSession):

@coroutine
def onJoin(self, *args, **kwargs):
channel = self.config.extra['channel']

def onTicker(*args, **kwargs):
self.config.extra['queue'].put((channel, (args, kwargs, time.time())))

try:
yield from self.subscribe(onTicker, self.config.extra['channel'])

except Exception as e:
raise

class PlnxEndpoint(mp.Process):
def __init__(self, endpoint, q, **kwargs):
super(PlnxEndpoint, self).__init__(name='%s Endpoint Process' %
endpoint, **kwargs)
self.endpoint = endpoint
self.q = q

def run(self):
self.runner = ApplicationRunner("wss://api.poloniex.com:443", 'realm1',
extra={'channel': self.endpoint,
'queue': self.q})
self.runner.run(PoloniexSession)

def join(self, *args, **kwargs):
def sig_handler(x, y):
pass
signal.signal(signal.SIGINT, sig_handler)
super(PlnxEndpoint, self).join(*args, **kwargs)

class PoloniexWSS(WSSAPI):
def __init__(self, endpoints=None):
super(PoloniexWSS, self).__init__(None, 'Poloniex')
self.data_q = mp.Queue()
self.connections = {}
if endpoints:
self.endpoints = endpoints
else:
r = requests.get('https://poloniex.com/public?command=returnTicker')
self.endpoints = list(r.json().keys())
self.endpoints.append('ticker')

for endpoint in self.endpoints:
self.connections[endpoint] = PlnxEndpoint(endpoint, self.data_q)

def start(self):
super(PoloniexWSS, self).start()
for conn in self.connections:
self.connections[conn].start()

def stop(self):
for conn in self.connections:
self.connections[conn].join()
super(PoloniexWSS, self).stop()

Хотя это заполняет Self.q адекватно, я все еще получаю ошибку, когда мои подпроцессы остановлены:
RuntimeError: Event loop stopped before Future completed.
Traceback (most recent call last):
File "/home/nils/anaconda3/lib/python3.5/multiprocessing /process.py", line 254, in _bootstrap
self.run()
File "/home/nils/git/tools/bitexwss/bitexws//api/poloniex.py", line 46, in run
self.runner.run(PoloniexSession)
File "/home/nils/anaconda3/lib/python3.5/site-packages/autobahn-0.14.1-py3.5.egg/autobahn/asyncio/wamp.py", line 172, in run
loop.run_until_complete(protocol._session.leave())
File "/home/nils/anaconda3/lib/python3.5/asyncio/base_events.py", line 335, in run_until_complete
raise RuntimeError('Event loop stopped before Future completed.')

, который заставляет меня верить, что мой сигнал. sigint не запускается там, где я хочу. Закрытие Asyncio.event_loop также приведена в приведенной выше ошибке:
class PlnxEndpoint(mp.Process):
#...
def join(self, *args, **kwargs):
loop = get_event_loop()
loop.stop()
super(PlnxEndpoint, self).join(*args, **kwargs)
#...


Подробнее здесь: https://stackoverflow.com/questions/428 ... nal-sigint