Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add forgotten register operation for gevent subscription #22

Closed
wants to merge 2 commits into from
Closed

Add forgotten register operation for gevent subscription #22

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8840d9c
Fix register operation for gevent subscriptions
overlogic Oct 23, 2018
01d5b3c
Fix on complete behavior for geven connection
overlogic Oct 24, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions graphql_ws/gevent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@
)
from .constants import (
GQL_CONNECTION_ACK,
GQL_CONNECTION_ERROR
GQL_CONNECTION_ERROR,
GQL_COMPLETE,
)


Expand Down Expand Up @@ -81,34 +82,38 @@ 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
self.on_complete
))
connection_context.register_operation(op_id, disposable)
except Exception as e:
self.send_error(connection_context, op_id, str(e))

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)