Merge pull request #3 from rhizolab/add-extension-interface

Added an extension interface object

Dockerfile

@@ -6,7 +6,11 @@ WORKDIR /rhizo-server
 COPY requirements.txt ./
 RUN pip install -r requirements.txt
 RUN pip install psycopg2-binary
-RUN pip install rauth
+
+# uncomment to install flow-server extension dependencies
+#RUN pip install rauth
+#RUN pip install python-jose
+#RUN pip install requests
 
 # copy in the app source
 COPY . .

main/app.py

@@ -8,6 +8,7 @@
 from flask_sockets import Sockets
 from messages.socket_sender import SocketSender, clear_web_sockets
 from messages.message_queue_basic import MessageQueueBasic
+from main.extension import ExtensionInterface
 
 # create and configure the application
 app = Flask(__name__)
@@ -76,13 +77,20 @@ def force_secure():
 # fix(later): revisit this
 #clear_web_sockets()
 
+# create the extension interface
+extension_interface = ExtensionInterface()
+
 # load server extensions
 extensions = []
 auto_load_config = os.environ.get('AUTOLOAD_EXTENSIONS') in ['True', 'true']
 for extension_name in app.config.get('EXTENSIONS', []):
     print('loading extension: %s' % extension_name)
     extension_module = importlib.import_module('extensions.' + extension_name + '.ext')
-    extension = extension_module.create()
+    try:
+        extension = extension_module.create(extension_interface)
+    except TypeError:
+        # fallback for extensions that haven't updated to use the interface
+        extension = extension_module.create()
     extension.name = extension_name
     extension.path = os.path.dirname(extension_module.__file__)
     if not os.path.isabs(extension.path):

main/extension.py

@@ -1,16 +1,34 @@
 from flask import render_template_string
-
+from collections import defaultdict, OrderedDict
 
 # a super-class to be sub-classes by server extension packages
 class Extension(object):
 
     # create a new extension object
-    def __init__(self):
+    def __init__(self, extension_interface=None):
         self.name = None  # set in app.py
         self.path = None  # set in app.py
+        self.extension_interface = extension_interface
 
     # render a template from the extension's templates folder
     def render_template(self, name, **kwargs):
         template_file_name = self.path + '/templates/' + name
         template = open(template_file_name).read()
         return render_template_string(template, **kwargs)
+
+class ExtensionInterface():
+
+    def __init__(self):
+        self._listeners = defaultdict(OrderedDict)
+
+    def on(self, eventName, listener):
+        self._listeners[eventName][listener] = listener
+
+    def off(self, eventName, listener):
+        self._listeners[eventName].pop(listener)
+
+    def emit(self, eventName, *args, **kwargs):
+        for listener in list(self._listeners[eventName].values()):
+            listener(*args, **kwargs)
+
+

main/messages/socket_receiver.py

@@ -13,7 +13,7 @@
 
 
 # internal imports
-from main.app import db, socket_sender, app, message_queue
+from main.app import db, socket_sender, app, message_queue, extension_interface
 from main.users.auth import find_key, find_key_by_code
 from main.users.permissions import ACCESS_LEVEL_READ, ACCESS_LEVEL_WRITE
 from main.messages.outgoing_messages import handle_send_email, handle_send_text_message
@@ -84,6 +84,7 @@ def manage_web_socket(ws):
 
     # register this socket to receive outgoing messages
     socket_sender.register(ws_conn)
+    extension_interface.emit('socket_connected', ws_conn, socket_sender)
 
     # process incoming messages
     while not ws_conn.ws.closed:
@@ -95,6 +96,7 @@ def manage_web_socket(ws):
 
     # websocket has been closed
     ws_conn.log_disconnect()
+    extension_interface.emit('socket_disconnected', ws_conn)
     socket_sender.unregister(ws_conn)
     db.session.close()
 

main/messages/socket_sender.py

@@ -28,15 +28,19 @@ def send(self, ws_conn, message):
             except:  # WebSocketError:
                 print('unable to send to websocket (%s)' % ws_conn)
 
-    # send an error message back to a client
-    def send_error(self, ws_conn, message_text):
+    # send a message structure to a specific client
+    def send_message(self, ws_conn, type, parameters):
         message_struct = {
-            'type': 'error',
+            'type': type,
             'timestamp': datetime.datetime.utcnow().isoformat() + 'Z',
-            'parameters': {'message': message_text},
+            'parameters': parameters
         }
         self.send(ws_conn, json.dumps(message_struct))
 
+    # send an error message back to a client
+    def send_error(self, ws_conn, message_text):
+        self.send_message(ws_conn, 'error', {'message': message_text})
+
     # this function sits in a loop, waiting for messages that need to be sent out to subscribers
     def send_messages(self):
         from main.app import message_queue