Subscription must return Async Iterable or Observable. Received:

[japrogramer]

Hello, I have previously gotten subscriptions to crudely work graphql-python/graphene#500 (comment)

But now i want to try the new version of the graphql-core library and here is how far I've gotten. It is not very clear to me what a Async Iterable is meant to be I assumed that its a Async generator?

but from this line it seems to be anything that is an instance of an Observable
https://github.com/graphql-python/graphql-core/blob/0b0551e7dfb724dedc66759145b9221595ec8598/graphql/execution/executor.py#L296
Is there any documentation for graphql-cores new approach to subscriptions .. I would very much like to learn more.

Here is what im trying

[GraphQLError('Subscription must return Async Iterable or Observable. Received: <async_generator object ProductSubscritption.resolve_sub_product.<locals>.compat at 0x7f6335e89d48>',)]

    def resolve_sub_product(self, info, **input):
        async def compat(result, delay):
            yield result
            await asyncio.sleep(delay)
        return compat(make_sub(info, input.get('product')), .1)

[japrogramer]

Hmm I think I understand now, In the on next, I could also send to the reply channel in channels
My question is this now, will it block other normal request from occurring in django? hmm, I need to figure out how to connect django signals with this new mechanism ... any ideas?

currently im adding the reply channel to a group and returning the query, With this new layout would it even be necessary to add the reply_channel to a group ? hmm, I need to keep reading.

~   1         result = schema.execute(query, variable_values=foovar, allow_subscriptions=True, **kwargs)                                                                                                            
+   2         if isinstance(result, rx.Observable):                                                                                                                                                                 
+   3             class MyObserver(rx.Observer):                                                                                                                                                                    
+   4                                                                                                                                                                                                               
+   5                 def on_next(self, x):                                                                                                                                                                         
+   6                     nonlocal result                                                                                                                                                                           
+   7                     result = x                                                                                                                                                                                
+   8                                                                                                                                                                                                               
+   9                 def on_error(self, e):                                                                                                                                                                        
+  10                     ...                                                                                                                                                                                       
+  11                                                                                                                                                                                                               
+  12                 def on_completed(self):                                                                                                                                                                       
+  13                     ...                                                                                                                                                                                       
+  14             result.subscribe(MyObserver())                                                                                                                                                                    
   15         data = result.data       

[perrosnk]

+1

[AgentChris]

can you give us the complete code, to play with it?

[japrogramer]

most of it is in graphql-python/graphene#500 (comment)

this is the part that makes it rx compliant

  1 def make_sub(info, gid):
  2     inst = relay.Node.get_node_from_global_id(info, gid)
  3     try:
  4         gp_name = 'gqp.{0}-updated.{1}'.format(str.lower(inst.__class__.__name__), inst.pk)
  5         Group(gp_name).add(info.context.reply_channel)
  6         info.context.channel_session['Groups'] = ','.join(
  7             (gp_name, info.context.channel_session['Groups']))
  8     except:
  9         pass
 10     return iter([inst])



_   1     def resolve_sub_product(self, info, **input):                                                                                                                                                              
  162         return rx.Observable.from_iterable(make_sub(info, input.get('product')))

[eamigo86]

Hello @japrogramer, any progress returning an Async Iterable or Observable in the subscriptions?

[japrogramer]

@eamigo86 yes, rx.Observable.from_iterable( ... ) in the resolve method on the comment above I explain more.

[eamigo86]

Hi again @japrogramer, my problem is that the subscription always gives me back a promise with the observable inside.
In this image you can see that the function resolve_or_error within the function subscribe_field on execution/executor.py, always gives me back a promise. That's why I get the error.

I use the default SyncExecutor executor. So in when I debug in its execute function get this:

Any idea how to fix it?

[japrogramer]

@eamigo86 what query are you running and what does the resolve for those fields return?

[japrogramer]

are you using a dataloader in any of those fields?

[eamigo86]

No, I not use Dataloader, I just define my Subscription like a ObjectType class, and on my subscription_resolver function return a ObjectType, thanks to your code I now return a Observable, just like this:

[japrogramer]

ok, so when the schema returns a result it should be an instance of rx.Observable
than this code should help you out graphql-python/graphql-core#149 (comment)

[japrogramer]

the way Im doing subsrciptions is by adding the reply_channel to a group graphql-python/graphql-core#149 (comment)
and because Im serving subscriptions over a websocket, when i send in the group it goes to the people subscribed. So im subscribing to events and than in the client I determine if I want to query for the data based on the events.

socket = new WebSocket("ws://" + window.location.host + "/gql");
socket.onmessage = function(e) {
    alert(e.data);
}
socket.send(JSON.stringify({query: 'subscription {\
  subProduct (uuid: "714a3f4f-4a21-4ff9-b058-f12ad6389f72"){\
    id,\
    disabled,\
    title,\
  }\
}'}))

[eamigo86]

I understand the way you use the subscriptions, in my case, I use the subscriptions in graphql only to register a user connected by wescoket to the server, in a certain group, so when the event to which it has subscribed is given, it is notified via websocket, like this:

Everything works great, except that in version 2.0 of graphql-core an Iterable or an Observable Async must be returned, and I only return the confirmation that the subscription was made in a satisfactory way.

[eamigo86]

I only manage to obtain an Observable if I put in the settings:

GRAPHENE = {  
    'MIDDLEWARE': []
}