Move event loop to eddn and log out gracefully.

diff --git a/app.py b/app.py
index 494bc55..ddc685a 100644 (file)
--- a/app.py
+++ b/app.py
@@ -242,14 +242,8 @@ modules = {
   }
 }
 
-def exception_callback(loop, context):
-  if 'exception' in context:
-    log.exception('asyncio.loop')
-    raise
-
-loop = zmq.asyncio.ZMQEventLoop()
+loop = eddn.init()
 asyncio.set_event_loop(loop)
-loop.set_exception_handler(exception_callback)
 client = discord.Client()
 bot.client = client
 bot.db = db
@@ -369,8 +363,7 @@ try:
   log.info('Opening WebSocket.')
   loop.run_until_complete(client.connect())
 except KeyboardInterrupt:
-  log.info('Logging out.')
-  loop.run_until_complete(client.logout())
+  log.info('Interrupted.')
 except RuntimeError:
   log.exception('RuntimeError')
   ret = 100
@@ -378,8 +371,11 @@ except:
   log.exception('loop')
   ret = 99
 finally:
+  if loop.is_running():
+    log.info('Logging out.')
+    loop.run_until_complete(client.logout())
   try:
-    loop.close()
+    eddn.close()
   except:
     ret = 101
 sys.exit(ret)

diff --git a/eddn.py b/eddn.py
index 6e6f7a6..f78ec28 100644 (file)
--- a/eddn.py
+++ b/eddn.py
@@ -12,13 +12,58 @@ url = 'tcp://eddn-relay.elite-markets.net:9500'
 schema_prefix = 'http://schemas.elite-markets.net/eddn'
 topics = []
 timeout = 600000
+state = { 'loop': None, 'context': None, 'subscriber': None }
+
+def exception_callback(loop, context):
+  if 'exception' in context:
+    log.exception('asyncio.loop')
+    raise
+
+def init():
+  state['loop'] = zmq.asyncio.ZMQEventLoop()
+  state['loop'].set_exception_handler(exception_callback)
+  return state['loop']
+
+def close():
+  loop = state['loop']
+  context = state['context']
+  subscriber = state['subscriber']
+
+  if loop is None:
+    log.info('Event loop is not initialised.')
+    return True
+
+  if loop.is_closed():
+    log.info('Event loop is closed.')
+    return True
+
+  if context is None:
+    log.info('Context is stopped.')
+    return True
+
+  if subscriber is not None:
+    log.info('Closing subscriber.')
+    subscriber.close()
+  else:
+    log.info('Subscriber is cloed.')
+
+  log.info('Stopping context.')
+  context.term()
+
+  if loop.is_running():
+    log.info('Stopping loop.')
+    loop.stop()
+
+  return True
 
 async def listen(plugins):
   waittime = 5
   context = zmq.asyncio.Context()
+  state['context'] = context
   while True:
     try:
       subscriber = context.socket(zmq.SUB)
+      state['subscriber'] = subscriber
       log.info('Connecting to {}'.format(url))
       subscriber.connect(url)
       if len(topics):
@@ -46,8 +91,10 @@ async def listen(plugins):
     except zmq.ZMQError:
       log.exception('ZMQSocketException')
     subscriber.disconnect(url)
+    state['subscriber'] = None
     asyncio.sleep(waittime)
   context.destroy()
+  state['context'] = None
 
 async def decode_message(compressed):
   try: