Listen to EDDN.

diff --git a/app.py b/app.py
index 434d5c5..efa6048 100644 (file)
--- a/app.py
+++ b/app.py
@@ -12,11 +12,13 @@ import re
 import shlex
 import sys
 import time
+import zmq.asyncio
 
 from db import DBConnection
 from plugins import Plugins
 import bot
 import cat
+import eddn
 
 db = DBConnection()
 
@@ -240,6 +242,7 @@ modules = {
   }
 }
 
+asyncio.set_event_loop(zmq.asyncio.ZMQEventLoop())
 client = discord.Client()
 bot.client = client
 bot.db = db
@@ -307,6 +310,7 @@ async def on_ready():
     plugins.on_ready()
     asyncio.ensure_future(maybe_sleep())
     asyncio.ensure_future(request_offline_members())
+    asyncio.ensure_future(eddn.listen(plugins))
     threads_ready.value = True
 
 def mentioned_in(message, explicit = True):

diff --git a/eddn.py b/eddn.py
new file mode 100644 (file)
index 0000000..4661e31
--- /dev/null
+++ b/eddn.py
@@ -0,0 +1,53 @@
+import asyncio
+import json
+import logging
+import zlib
+import zmq
+import zmq.asyncio
+
+log = logging.getLogger('eddn')
+
+url = 'tcp://eddn-relay.elite-markets.net:9500'
+topic = b''
+timeout = 600000
+
+async def listen(plugins):
+  waittime = 5
+  context = zmq.asyncio.Context()
+  while True:
+    try:
+      subscriber = context.socket(zmq.SUB)
+      subscriber.connect(url)
+      subscriber.setsockopt(zmq.SUBSCRIBE, topic)
+      subscriber.setsockopt(zmq.RCVTIMEO, timeout)
+      poller = zmq.asyncio.Poller()
+      poller.register(subscriber, zmq.POLLOUT)
+      while True:
+        try:
+          messages = await subscriber.recv_multipart()
+        except zmq.error.Again:
+          log.exception('recv_multipart')
+          messages = False
+        if messages == False:
+          break
+        for message in messages:
+          data = await decode_message(message)
+          if data is not None:
+            await plugins.eddn_message(data['$schemaRef'], data['header'], data['message'])
+    except zmq.ZMQError:
+      log.exception('ZMQSocketException')
+    subscriber.disconnect(url)
+    asyncio.sleep(waittime)
+  context.destroy()
+
+async def decode_message(compressed):
+  try:
+    uncompressed = zlib.decompress(compressed)
+  except zlib.error:
+    log.exception('Malformed gzipped message')
+    return None
+  try:
+    return json.loads(uncompressed.decode('utf-8'))
+  except ValueError:
+    log.exception('Malformed JSON in message')
+    return None

diff --git a/plugins.py b/plugins.py
index 6da6f3c..8a8a271 100644 (file)
--- a/plugins.py
+++ b/plugins.py
@@ -19,7 +19,8 @@ class Plugins(object):
       'on_member_join',
       'on_member_update',
       'handle_command',
-      'handle_help'
+      'handle_help',
+      'eddn_message'
     ]
     self.plugins_supporting_methods = {}
     self.loaded = {}
@@ -164,3 +165,8 @@ class Plugins(object):
           continue
       return True
     return False
+
+  async def eddn_message(self, schema, data, message):
+    for name in self.plugins():
+      if self.has_method(name, 'eddn_message'):
+        asyncio.ensure_future(self.call_coroutine(name, 'eddn_message', schema, data, message))

diff --git a/requirements.txt b/requirements.txt
index dff585c..3c2a1e5 100644 (file)
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,5 +3,6 @@ discord.py[voice]==0.16.6
 feedparser
 pexpect
 pillow
+pyzmq
 pytz
 youtube_dl