Make several modifications to executor logic

Change asyncio "join" method; construct "delayed_backgrd_task" method;
various other minor changes

Author: hballard

graphql_subscriptions/executors/asyncio.py

@@ -60,41 +60,33 @@ def ws_send(ws, msg):
     def ws_recv(ws):
         return ws.recv()
 
-    @staticmethod
-    def sleep(time):
-        return asyncio.sleep(time)
-
-    @staticmethod
-    @asyncio.coroutine
-    def set_timeout(callback, period):
-        while True:
-            callback()
-            yield from asyncio.sleep(period)
+    def sleep(self, time):
+        return self.loop.run_until_complete(asyncio.sleep(time))
 
     @staticmethod
     def kill(future):
         future.cancel()
 
-    @staticmethod
-    def join(future, timeout):
-        return asyncio.wait_for(future, timeout=timeout)
+    def join(self, future, timeout=None):
+        return self.loop.run_until_complete(asyncio.wait_for(future,
+                                                             timeout=timeout))
 
-    def join_all(self):
-        while self.futures:
-            futures = self.futures
-            self.futures = []
-            asyncio.wait(futures)
-        return futures
+    @staticmethod
+    @asyncio.coroutine
+    def delayed_backgrd_task(func, callback=None, period=0, *arg, **kwargs):
+        while True:
+            if func:
+                result = func(*arg, **kwargs)
+                if asyncio.iscoroutine(result):
+                    msg = yield from result
+                    if callback:
+                        callback(msg)
+            if callback:
+                callback(result)
+            yield from asyncio.sleep(period)
 
     def execute(self, fn, *args, **kwargs):
         coro = fn(*args, **kwargs)
         future = ensure_future(coro, loop=self.loop)
         self.futures.append(future)
         return future
-
-    def execute_and_call_callback(self, fn, callback, *args, **kwargs):
-        coro = fn(*args, **kwargs)
-        future = ensure_future(coro, loop=self.loop)
-        self.futures.append(future)
-        future.add_done_callback(callback)
-        return future

graphql_subscriptions/executors/gevent.py

@@ -40,7 +40,7 @@ def sleep(time):
         gevent.sleep(time)
 
     @staticmethod
-    def set_timeout(callback, period):
+    def delayed_backgrd_task(callback, period):
         while True:
             callback()
             gevent.sleep(period)
@@ -53,11 +53,6 @@ def kill(greenlet):
     def join(greenlet):
         greenlet.join()
 
-    def join_all(self):
-        joined_greenlets = gevent.joinall(self.greenlets)
-        self.greenlets = []
-        return joined_greenlets
-
     def execute(self, fn, *args, **kwargs):
         greenlet = gevent.spawn(fn, *args, **kwargs)
         self.greenlets.append(greenlet)

graphql_subscriptions/subscription_manager/pubsub.py

@@ -36,7 +36,7 @@ def __init__(self,
         self.pubsub = self.redis.pubsub()
 
         self.executor = executor()
-        self.get_message_task = None
+        self.backgrd_task = None
 
         self.subscriptions = {}
         self.sub_id_counter = 1
@@ -57,8 +57,8 @@ def subscribe(self, trigger_name, on_message_handler, options):
         self.subscriptions[self.sub_id_counter] = [
             trigger_name, on_message_handler
         ]
-        if not self.get_message_task:
-            self.get_message_task = self.executor.execute(
+        if not self.backgrd_task:
+            self.backgrd_task = self.executor.execute(
                 self.wait_and_get_message)
         return Promise.resolve(self.sub_id_counter)
 
@@ -73,17 +73,30 @@ def unsubscribe(self, sub_id):
             self.executor.execute(self.pubsub.unsubscribe, trigger_name)
 
         if not self.subscriptions:
-            self.get_message_task = self.executor.kill(self.get_message_task)
+            self.backgrd_task = self.executor.kill(self.backgrd_task)
 
-    def wait_and_get_message(self):
+    def _wait_and_get_message_async(self):
+        self.executor.execute(self.executor.delayed_backgrd_task,
+                              self.pubsub.get_message,
+                              self.handle_message,
+                              .001,
+                              ignore_subscribe_messages=True)
+
+    def _wait_and_get_message_sync(self):
         while True:
-            # import pytest; pytest.set_trace()
             message = self.pubsub.get_message(ignore_subscribe_messages=True)
             if message:
                 self.handle_message(message)
             self.executor.sleep(.001)
 
+    def wait_and_get_message(self):
+        if hasattr(self.executor, 'loop'):
+            self._wait_and_get_message_async()
+        else:
+            self._wait_and_get_message_sync()
+
     def handle_message(self, message):
+
         if isinstance(message['channel'], bytes):
             channel = message['channel'].decode()
 

graphql_subscriptions/subscription_transport_ws/server.py

@@ -19,8 +19,8 @@ def __init__(self,
                  on_connect=None,
                  on_disconnect=None):
 
-        assert subscription_manager, "Must provide\
-            'subscription_manager' to websocket app constructor"
+        assert subscription_manager, ("Must provide\
+            'subscription_manager' to websocket app constructor")
 
         self.ws = websocket
         self.subscription_manager = subscription_manager
@@ -54,9 +54,11 @@ def _handle_async(self):
 
         while True:
             try:
-                self.executor.execute_and_call_callback(self.executor.ws_recv,
-                                                        self._on_message,
-                                                        self.ws)
+                self.executor.execute(self.executor.delayed_backgrd_task,
+                                      self.executor.ws_recv,
+                                      self.on_message,
+                                      0,
+                                      self.ws)
             except self.executor.error:
                 self.on_close()
                 break
@@ -86,9 +88,11 @@ def keep_alive_callback():
                 self.executor.kill(keep_alive_timer)
 
         if self.keep_alive:
-            keep_alive_timer = self.executor.execute(self.executor.set_timeout,
-                                                     keep_alive_callback,
-                                                     self.keep_alive)
+            keep_alive_timer = self.executor.execute(
+                self.executor.delayed_backgrd_task,
+                None,
+                keep_alive_callback,
+                self.keep_alive)
 
     def on_close(self):
         for sub_id in list(self.connection_subscriptions.keys()):
@@ -98,13 +102,12 @@ def on_close(self):
         if self.on_disconnect:
             self.on_disconnect(self.ws)
 
-    def _on_message(self, future):
-        self.on_message(future.result())
-
     def on_message(self, msg):
 
         if msg is None:
             return
+        elif hasattr(msg, 'result'):  # check if future from asyncio
+            msg = msg.result()
 
         non_local = {'on_init_resolve': None, 'on_init_reject': None}
 
@@ -184,8 +187,9 @@ def subscription_start_promise_handler(init_result):
 
                     def promised_params_handler(params):
                         if not isinstance(params, dict):
-                            error = 'Invalid params returned from\
-OnSubscribe!  Return value must be an dict'
+                            error = ('Invalid params returned from'
+                                     'OnSubscribe!  Return value must be an'
+                                     'dict')
 
                             self.send_subscription_fail(
                                 sub_id, {'errors': [{

tests/test_subscription_manager.py

@@ -11,10 +11,11 @@
 from graphql_subscriptions import RedisPubsub, SubscriptionManager
 from graphql_subscriptions.subscription_manager.validation import (
      SubscriptionHasSingleRootField)
-from graphql_subscriptions.executors.gevent import GeventExecutor
+# from graphql_subscriptions.executors.gevent import GeventExecutor
+from graphql_subscriptions.executors.asyncio import AsyncioExecutor
 
 
-@pytest.fixture(params=[GeventExecutor])
+@pytest.fixture(params=[AsyncioExecutor])
 def executor(request):
     return request.param
 
@@ -31,13 +32,13 @@ def test_pubsub_subscribe_and_publish(pubsub, executor, test_input, expected):
     def message_callback(message):
         try:
             assert message == expected
-            executor.kill(pubsub.get_message_task)
+            executor.kill(pubsub.backgrd_task)
         except AssertionError as e:
             sys.exit(e)
 
     def publish_callback(sub_id):
         assert pubsub.publish('a', test_input)
-        executor.join(pubsub.get_message_task)
+        executor.join(pubsub.backgrd_task)
 
     p1 = pubsub.subscribe('a', message_callback, {})
     p2 = p1.then(publish_callback)
@@ -52,7 +53,7 @@ def unsubscribe_publish_callback(sub_id):
         pubsub.unsubscribe(sub_id)
         assert pubsub.publish('a', 'test')
         try:
-            executor.join(pubsub.get_message_task)
+            executor.join(pubsub.backgrd_task)
         except AttributeError:
             return
 
@@ -190,14 +191,14 @@ def test_subscribe_with_valid_query_and_return_root_value(sub_mgr, executor):
     def callback(e, payload):
         try:
             assert payload.data.get('testSubscription') == 'good'
-            executor.kill(sub_mgr.pubsub.get_message_task)
+            executor.kill(sub_mgr.pubsub.backgrd_task)
         except AssertionError as e:
             sys.exit(e)
 
     def publish_and_unsubscribe_handler(sub_id):
         sub_mgr.publish('testSubscription', 'good')
         # pytest.set_trace()
-        executor.join(sub_mgr.pubsub.get_message_task)
+        executor.join(sub_mgr.pubsub.backgrd_task)
         sub_mgr.unsubscribe(sub_id)
 
     p1 = sub_mgr.subscribe(query, 'X', callback, {}, {}, None, None)
@@ -218,14 +219,14 @@ def callback(err, payload):
                     assert True
                 else:
                     assert payload.data.get('testFilter') == 'good_filter'
-                    executor.kill(sub_mgr.pubsub.get_message_task)
+                    executor.kill(sub_mgr.pubsub.backgrd_task)
             except AssertionError as e:
                 sys.exit(e)
 
     def publish_and_unsubscribe_handler(sub_id):
         sub_mgr.publish('filter_1', {'filterBoolean': False})
         sub_mgr.publish('filter_1', {'filterBoolean': True})
-        executor.join(sub_mgr.pubsub.get_message_task)
+        executor.join(sub_mgr.pubsub.backgrd_task)
         sub_mgr.unsubscribe(sub_id)
 
     p1 = sub_mgr.subscribe(query, 'Filter1', callback, {'filterBoolean': True},
@@ -247,15 +248,15 @@ def callback(err, payload):
                     assert True
                 else:
                     assert payload.data.get('testFilter') == 'good_filter'
-                    executor.kill(sub_mgr.pubsub.get_message_task)
+                    executor.kill(sub_mgr.pubsub.backgrd_task)
             except AssertionError as e:
                 sys.exit(e)
 
     def publish_and_unsubscribe_handler(sub_id):
         sub_mgr.publish('filter_2', {'filterBoolean': False})
         sub_mgr.publish('filter_2', {'filterBoolean': True})
         try:
-            executor.join(sub_mgr.pubsub.get_message_task)
+            executor.join(sub_mgr.pubsub.backgrd_task)
         except:
             raise
         sub_mgr.unsubscribe(sub_id)
@@ -286,13 +287,13 @@ def callback(err, payload):
             except AssertionError as e:
                 sys.exit(e)
         if non_local['trigger_count'] == 2:
-            executor.kill(sub_mgr.pubsub.get_message_task)
+            executor.kill(sub_mgr.pubsub.backgrd_task)
 
     def publish_and_unsubscribe_handler(sub_id):
         sub_mgr.publish('not_a_trigger', {'filterBoolean': False})
         sub_mgr.publish('trigger_1', {'filterBoolean': True})
         sub_mgr.publish('trigger_2', {'filterBoolean': True})
-        executor.join(sub_mgr.pubsub.get_message_task)
+        executor.join(sub_mgr.pubsub.backgrd_task)
         sub_mgr.unsubscribe(sub_id)
 
     p1 = sub_mgr.subscribe(query, 'multiTrigger', callback,
@@ -343,7 +344,7 @@ def unsubscribe_and_publish_handler(sub_id):
         sub_mgr.unsubscribe(sub_id)
         sub_mgr.publish('testSubscription', 'good')
         try:
-            executor.join(sub_mgr.pubsub.get_message_task)
+            executor.join(sub_mgr.pubsub.backgrd_task)
         except AttributeError:
             return
 
@@ -393,17 +394,17 @@ def test_calls_the_error_callback_if_there_is_an_execution_error(
     def callback(err, payload):
         try:
             assert payload is None
-            assert err.message == 'Variable "$uga" of required type\
- "Boolean!" was not provided.'
+            assert err.message == ('Variable "$uga" of required type'
+                                   '"Boolean!" was not provided.')
 
-            executor.kill(sub_mgr.pubsub.get_message_task)
+            executor.kill(sub_mgr.pubsub.backgrd_task)
         except AssertionError as e:
             sys.exit(e)
 
     def unsubscribe_and_publish_handler(sub_id):
         sub_mgr.publish('testSubscription', 'good')
         try:
-            executor.join(sub_mgr.pubsub.get_message_task)
+            executor.join(sub_mgr.pubsub.backgrd_task)
         except AttributeError:
             return
         sub_mgr.unsubscribe(sub_id)
@@ -427,14 +428,14 @@ def callback(err, payload):
         try:
             assert err is None
             assert payload.data.get('testContext') == 'trigger'
-            executor.kill(sub_mgr.pubsub.get_message_task)
+            executor.kill(sub_mgr.pubsub.backgrd_task)
         except AssertionError as e:
             sys.exit(e)
 
     def unsubscribe_and_publish_handler(sub_id):
         sub_mgr.publish('context_trigger', 'ignored')
         try:
-            executor.join(sub_mgr.pubsub.get_message_task)
+            executor.join(sub_mgr.pubsub.backgrd_task)
         except AttributeError:
             return
         sub_mgr.unsubscribe(sub_id)
@@ -459,14 +460,14 @@ def callback(err, payload):
         try:
             assert payload is None
             assert str(err) == 'context error'
-            executor.kill(sub_mgr.pubsub.get_message_task)
+            executor.kill(sub_mgr.pubsub.backgrd_task)
         except AssertionError as e:
             sys.exit(e)
 
     def unsubscribe_and_publish_handler(sub_id):
         sub_mgr.publish('context_trigger', 'ignored')
         try:
-            executor.join(sub_mgr.pubsub.get_message_task)
+            executor.join(sub_mgr.pubsub.backgrd_task)
         except AttributeError:
             return
         sub_mgr.unsubscribe(sub_id)
@@ -529,16 +530,16 @@ def test_should_not_allow_inline_fragments(validation_schema):
     errors = validate(validation_schema,
                       parse(sub), [SubscriptionHasSingleRootField])
     assert len(errors) == 1
-    assert errors[0].message == 'Apollo subscriptions do not support\
- fragments on the root field'
+    assert errors[0].message == ('Apollo subscriptions do not support'
+                                 'fragments on the root field')
 
 
 def test_should_not_allow_fragments(validation_schema):
-    sub = 'subscription S5{ ...testFragment }\
- fragment testFragment on Subscription{ test2 }'
+    sub = ('subscription S5{ ...testFragment }'
+           'fragment testFragment on Subscription{ test2 }')
 
     errors = validate(validation_schema,
                       parse(sub), [SubscriptionHasSingleRootField])
     assert len(errors) == 1
-    assert errors[0].message == 'Apollo subscriptions do not support\
- fragments on the root field'
+    assert errors[0].message == ('Apollo subscriptions do not support'
+                                 'fragments on the root field')