test it with redis and some workers

examples/django_subscriptions/django_subscriptions/asgi.py

@@ -0,0 +1,6 @@
+import os
+from channels.asgi import get_channel_layer
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_subscriptions.settings")
+
+channel_layer = get_channel_layer()

examples/django_subscriptions/django_subscriptions/schema.py

@@ -0,0 +1,23 @@
+import graphene
+from rx import Observable
+
+
+class Query(graphene.ObjectType):
+ hello = graphene.String()
+
+ def resolve_hello(self, info, **kwargs):
+ return 'world'
+
+class Subscription(graphene.ObjectType):
+
+ count_seconds = graphene.Int(up_to=graphene.Int())
+
+
+ def resolve_count_seconds(root, info, up_to=5):
+ return Observable.interval(1000)\
+ .map(lambda i: "{0}".format(i))\
+ .take_while(lambda i: int(i) <= up_to)
+
+
+
+schema = graphene.Schema(query=Query, subscription=Subscription)

examples/django_subscriptions/django_subscriptions/settings.py

@@ -122,36 +122,16 @@
 CHANNELS_WS_PROTOCOLS = ["graphql-ws", ]
 CHANNEL_LAYERS = {
  "default": {
- "BACKEND": "asgiref.inmemory.ChannelLayer",
+ "BACKEND": "asgi_redis.RedisChannelLayer",
+ "CONFIG": {
+ "hosts": [("localhost", 6379)],
+ },
  "ROUTING": "django_subscriptions.urls.channel_routing",
  },
 
 }
 
-import graphene
-from rx import Observable
-
-
-class Query(graphene.ObjectType):
- hello = graphene.String()
-
- def resolve_hello(self, info, **kwargs):
- return 'world'
-
-class Subscription(graphene.ObjectType):
-
- count_seconds = graphene.Int(up_to=graphene.Int())
-
-
- def resolve_count_seconds(root, info, up_to=5):
- return Observable.interval(1000)\
- .map(lambda i: "{0}".format(i))\
- .take_while(lambda i: int(i) <= up_to)
-
-
-
-schema = graphene.Schema(query=Query, subscription=Subscription)
 
 GRAPHENE = {
- 'SCHEMA': schema
+ 'SCHEMA': 'django_subscriptions.schema.schema'
 }

graphql_ws/django_channels.py

@@ -25,6 +25,13 @@ def __init__(self, message, request_context = None):
 
  def send(self, data):
  self.message.reply_channel.send(data)
+
+ def close(self, reason):
+ data = {
+ 'close': True,
+ 'text': reason
+ }
+ self.message.reply_channel.send(data)
 
 class DjangoChannelSubscriptionServer(BaseSubscriptionServer):
 
@@ -35,7 +42,7 @@ def get_graphql_params(self, *args, **kwargs):
 
  def handle(self, message, connection_context):
  self.on_message(connection_context, message)
- 
+
  def send_message(self, connection_context, op_id=None, op_type=None, payload=None):
  message = {}
  if op_id is not None:
@@ -46,7 +53,6 @@ def send_message(self, connection_context, op_id=None, op_type=None, payload=Non
  message['payload'] = payload
 
  assert message, "You need to send at least one thing"
-
  return connection_context.send({'text': json.dumps(message)})
 
  def on_open(self, connection_context):
@@ -98,10 +104,8 @@ class GraphQLSubscriptionConsumer(JsonWebsocketConsumer):
  strict_ordering = True
 
  def connect(self, message, **kwargs):
- schema = getattr(settings, "schema", None)
-
  message.reply_channel.send({"accept": True})
- print(self)
+
 
  def receive(self, content, **kwargs):
  """
@@ -113,13 +117,6 @@ def receive(self, content, **kwargs):
  self.subscription_server.on_open(self.connection_context)
  self.subscription_server.handle(content, self.connection_context)
 
- def disconnect(self, message, **kwargs):
- """
- Perform things on connection close
- """
- # self.subscription_cserver.on_close(self.connection_context)
-
-
 class SubscriptionObserver(Observer):
 
  def __init__(self, connection_context, op_id, send_execution_result, send_error, on_close):