From 8840d9cb444e4027ddf818a7cfa5abe7312beee4 Mon Sep 17 00:00:00 2001 From: Max Leonov Date: Tue, 23 Oct 2018 13:42:12 +0300 Subject: [PATCH 1/2] Fix register operation for gevent subscriptions --- graphql_ws/gevent.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/graphql_ws/gevent.py b/graphql_ws/gevent.py index 92a65ce..93cc594 100644 --- a/graphql_ws/gevent.py +++ b/graphql_ws/gevent.py @@ -81,13 +81,14 @@ def on_start(self, connection_context, op_id, params): connection_context.request_context, params) assert isinstance( execution_result, Observable), "A subscription must return an observable" - execution_result.subscribe(SubscriptionObserver( + disposable = execution_result.subscribe(SubscriptionObserver( connection_context, op_id, self.send_execution_result, self.send_error, self.on_close )) + connection_context.register_operation(op_id, disposable) except Exception as e: self.send_error(connection_context, op_id, str(e)) From 01d5b3c86e416b72e580f3cda8c8fe8318455cff Mon Sep 17 00:00:00 2001 From: Max Leonov Date: Wed, 24 Oct 2018 14:40:43 +0300 Subject: [PATCH 2/2] Fix on complete behavior for geven connection --- graphql_ws/gevent.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/graphql_ws/gevent.py b/graphql_ws/gevent.py index 93cc594..790d5b5 100644 --- a/graphql_ws/gevent.py +++ b/graphql_ws/gevent.py @@ -12,7 +12,8 @@ ) from .constants import ( GQL_CONNECTION_ACK, - GQL_CONNECTION_ERROR + GQL_CONNECTION_ERROR, + GQL_COMPLETE, ) @@ -86,7 +87,7 @@ def on_start(self, connection_context, op_id, params): op_id, self.send_execution_result, self.send_error, - self.on_close + self.on_complete )) connection_context.register_operation(op_id, disposable) except Exception as e: @@ -95,21 +96,24 @@ def on_start(self, connection_context, op_id, params): def on_stop(self, connection_context, op_id): self.unsubscribe(connection_context, op_id) + def on_complete(self, connection_context, op_id): + self.send_message(connection_context, op_id, GQL_COMPLETE) + class SubscriptionObserver(Observer): - def __init__(self, connection_context, op_id, send_execution_result, send_error, on_close): + def __init__(self, connection_context, op_id, send_execution_result, send_error, on_complete): self.connection_context = connection_context self.op_id = op_id self.send_execution_result = send_execution_result self.send_error = send_error - self.on_close = on_close + self.on_complete = on_complete def on_next(self, value): self.send_execution_result(self.connection_context, self.op_id, value) def on_completed(self): - self.on_close(self.connection_context) + self.on_complete(self.connection_context, self.op_id) def on_error(self, error): self.send_error(self.connection_context, self.op_id, error)