Add requirements file and initial implementation modules

Author: hballard

requirements.txt

@@ -0,0 +1,9 @@
+appdirs==1.4.0
+gevent==1.2.1
+gevent-websocket==0.9.5
+greenlet==0.4.12
+packaging==16.8
+Promises==0.6.27
+pyparsing==2.1.10
+redis==2.10.5
+six==1.10.0

subscription_manager.py

@@ -0,0 +1,184 @@
+import redis
+import gevent
+from types import FunctionType
+from promise import Promise
+from graphql import parse, validate, specified_rules, value_from_ast, execute
+from graphql.language.ast import OperationDefinition
+
+
+class RedisPubsub(object):
+
+    def __init__(self, host='localhost', port=6379, *args, **kwargs):
+        self.redis = redis.StrictRedis(host, port, *args, **kwargs)
+        self.pubsub = self.redis.pubsub(ignore_subscribe_messages=True)
+        self.subscriptions = {}
+        self.sub_id_counter = 0
+        self.greenlet = None
+
+    def publish(self, trigger_name, message):
+        self.redis.publish(trigger_name, message)
+        return True
+
+    def subscribe(self, trigger_name, on_message_handler, options):
+        trigger = {}
+        trigger[trigger_name] = on_message_handler
+        self.pubsub.subscribe(**trigger)
+        if not self.greenlet:
+            self.greenlet = gevent.spawn(self.wait_and_get_message)
+        self.sub_id_counter += 1
+        self.subscriptions[self.sub_id_counter] = trigger_name
+        return Promise.resolve(self.sub_id_counter)
+
+    def unsubscribe(self, sub_id):
+        trigger_name = self.subscriptions[sub_id]
+        del self.subscriptions[sub_id]
+        self.pubsub.unsubscribe(trigger_name)
+        if not self.subscriptions:
+            self.greenlet = self.greenlet.kill()
+
+    def wait_and_get_message(self):
+        while True:
+            self.pubsub.get_message()
+            gevent.sleep(.001)
+
+
+class ValidationError(Exception):
+
+    def __init__(self, errors):
+        self.errors = errors
+        self.message = 'Subscription query has validation errors'
+
+
+class SubscriptionManager(object):
+
+    def __init__(self, schema, pubsub, setup_funcs={}):
+        self.schema = schema
+        self.pubsub = pubsub
+        self.setup_funcs = setup_funcs
+        self.subscriptions = {}
+        self.max_subscription_id = 0
+
+    def publish(self, trigger_name, payload):
+        self.pubsub.publish(trigger_name, payload)
+
+    def subscribe(self, query, operation_name, callback, variables, context,
+                  format_error, format_response):
+        parsed_query = parse(query)
+        errors = validate(
+            self.schema,
+            parsed_query,
+            # TODO: Need to create/add subscriptionHasSingleRootField
+            # rule
+            rules=specified_rules
+        )
+
+        if errors:
+            return Promise.reject(ValidationError(errors))
+
+        args = {}
+
+        subscription_name = ''
+
+        for definition in parsed_query.definitions:
+
+            if isinstance(definition, OperationDefinition):
+                root_field = definition.selection_set.selections[0]
+                subscription_name = root_field.name.value
+
+                fields = self.schema.get_subscription_type().get_fields()
+
+                for arg in root_field.arguments:
+
+                    arg_definition = [arg_def for arg_def in
+                                      fields[subscription_name].args if
+                                      arg_def == arg.name.value][0]
+
+                    args[arg_definition.name] = value_from_ast(
+                        arg.value,
+                        arg_definition.type,
+                        variables=variables
+                    )
+
+        if self.setup_funcs[subscription_name]:
+            trigger_map = self.setup_funcs[subscription_name](
+                query,
+                operation_name,
+                callback,
+                variables,
+                context,
+                format_error,
+                format_response,
+                args,
+                subscription_name
+            )
+        else:
+            trigger_map = {}
+            trigger_map[subscription_name] = {}
+
+        external_subscription_id = self.max_subscription_id
+        self.max_subscription_id += 1
+        self.subscriptions[external_subscription_id] = []
+        subscription_promises = []
+
+        for trigger_name in trigger_map.keys():
+            channel_options = trigger_map[trigger_name].get(
+                'channel_options',
+                {}
+            )
+            filter_func = trigger_map[trigger_name].get(
+                'filter_func',
+                True
+            )
+
+            def on_message(root_value):
+                if isinstance(context, FunctionType):
+                    context_promise = Promise(
+                        lambda resolve: resolve(context())
+                    )
+                else:
+                    context_promise = Promise.resolve(context)
+
+                def context_promise_then(context):
+                    if not filter_func(root_value, context):
+                        return
+
+                    execute(
+                        self.schema,
+                        parsed_query,
+                        root_value,
+                        context,
+                        variables,
+                        operation_name
+                    )
+
+                context_promise.then(
+                    context_promise_then
+                ).then(
+                    lambda data: callback(None, data)
+                ).catch(
+                    lambda error: callback(error)
+                )
+
+            subs_promise = Promise(
+                lambda resolve: resolve(self.pubsub.subscribe(
+                    trigger_name,
+                    on_message,
+                    channel_options
+                ))
+            )
+
+            subs_promise.then(
+                lambda id: self.subscriptions[
+                    external_subscription_id].append(id)
+            )
+
+            subscription_promises.append(subs_promise)
+
+        return Promise.all(subscription_promises).then(
+            lambda result: external_subscription_id
+        )
+
+    def unsubscribe(self, sub_id):
+        for internal_id in self.subscriptions.get(sub_id):
+            self.pubsub.unsubscribe(internal_id)
+        self.subscriptions.pop(sub_id, None)