Support subscriptions

[LegNeato]

Looks like subscriptions made it into the spec.

While most users will implement subscription logic outside juniper (like how Facebook interfaces with a MQTT server handling the actual delivery and connections) it might be useful to add some integrations as optional features.

[mhallin]

Wow, I hadn't seen that happen. It certainly complicates things :)

I'll have to look and see how different implementations solve this, but it seems that there are (at least) three parts to this feature:

• Storing/managing active subscriptions.
• An API to push messages into Juniper.
• An API to get transformed messages from Juniper out to clients.

We could probably use Stream from the futures crate to represent the two latter points, while the first one is just a trait that you have to implement for the backing storage you're interested in.

For convenience, it would be nice to have some reasonable implementations for these points behind feature flags just like the Iron and Rocket integrations. What it'll be specifically depends on the state of respective crates - I haven't looked at any of the AMQP/MQTT or websockets crates that are available.

[LegNeato]

That sounds about right to me! The AMQP and MQTT crates look pretty early days sadly...

[japrogramer]

In graphene they just went with a consumer type interface where on next the user decides how the message is transported. I was thinking that for the websocket part of sending we could use pheonix.js https://hexdocs.pm/phoenix/js/ by writting a ffi crate for it .. It has a grate track record.

[tailhook]

I've just prototyped subscriptions interface for my app. The basic idea is to rewrite subscription to a query using grapqhql-parser and execute normal juniper query each time underlying data is updated. This works fine except serializing request and parsing it again is surely quite costly.

Another finding is that subscriptions-transport-ws protocol fits serde model quite fine with some annotations (at least parts of the proto I needed so far).

So in the first iteration, I would say that API for subscriptions may look like this:

let (data, sub) = execute_with_subscription(...);

Where the function works similarly to execute, and data is result of a query. Except it's unclear what sub here means.

In the first iteration, it might even be true if request contains subscriptions. Meaning that you must manually inspect AST to find out events you need to subscribe to.

[japrogramer]

@tailhook hello I recently implemented subscriptions in python with observables,
here a sub could be an observable Stream that can both consume and send messages to the on_next method of the observable .. but the state of the rust observables library does not inspire confidence, it hasn't been touched in years.

[thedodd]

I'm definitely of the opinion that subscriptions should be implemented as part of, or following, the effort to cut this system over to a futures based API. Doing subscriptions without futures seems like we would be asking for major performance problems.

[JohnDoneth]

Subscriptions is the feature I want to see implemented the most. Polling is not really ideal for most of my use cases of GraphQL. I agree with @thedodd and with async-await semantics on the way, it will be much easier to create a futures based API.

[gklijs]

Really like rust, but to bad there is no support for subscriptions yet. For Clojure there is https://github.com/walmartlabs/lacinia which has subscriptions for some time now. But that was a lot easier, having already a good async web-socket implementation.

[jeffvandyke]

So after investigating, as of October 25 2018, it looks like the current status of subscriptions in juniper is that it is blocked by:

• Interop with async? (Tokio/Futures) #2 (Interop with async) to use futures, blocked by:
• futures crate reaching v0.3.0, (currently at 0.3.0-alpha.9), tracked/blocked by:
• futures 0.3.0 milestone, blocked by:
• (... about 6 various futures issues, some of which depend on other crates)

A lot of the futures 0.3 milestone stuff hasn't been discussed since March (though I didn't investigate commits in various forks/branches that might be working on it). It seems to me that we might be waiting a while for futures 0.3.0 to come out.

Is the best path forward for this project to start a branch to interop with futures in their current 0.3.0-alpha.9 state to have a good forward-compatible base for implementing usable GraphQL Subscriptions support?

A reason this is huge to me is that it allows a complete GraphQL implementation in a resource-constrained embedded systems Linux environment with a good language like Rust, which would be awesome!

[LegNeato]

Here is the interface Apollo uses to keep the lib delivery agnostic yet allow people to implement subscriptions for various 3rd party protocols:

https://github.com/apollographql/graphql-subscriptions/blob/master/src/pubsub-engine.ts

[mtyrrell]

Is there an ETA for this feature? What can we do to help get it unblocked?

[JohnDoneth]

@mtyrrell See https://areweasyncyet.rs/ for the blockers on the ergonomic language features that can help power subscriptions in the future.

[theduke]

We can totally implement subscriptions without tokio support though.

I will probably tackle this soon as I need it myself.

[D1plo1d]

Any updates on this? Would the asynchronous changes needed for subscription support be able to be reused in moving us towards @live directive support as well?

[japrogramer]

The asynchronous syntax for rust is about to be locked down, final proposal
https://boats.gitlab.io/blog/post/await-decision/

[jmandel1027]

Any updates on this? Adding support for this would be killer. Really like the work you guys have done with this package,

I have a apollo service talking to some iot devices (over ros / mqtt)and weve been thinking about switching it over to either the excellent gqlgen lib in go but also weighing graphql options in rust, having subscription support would deff be a big factor lending in rusts favor

[jasonlor]

+1 on subscription support.

Loving the implementation of juniper thus far.

[nWacky]

Here are some design thoughts we are about to try to implement on top of async-await branch:

If resolving query/mutation results in a value/future, then resolving subscription represents a series of values/futures, so results in an iterator/stream.

First, we need to add subscription support to GraphQLType and ability to have our own resolvers (and probably update macros, or introduce new #[juniper::subscription] macro).

Then, high-level processing logic on subscription request should look like that:
(stream may be replaced by iterator in sync environment)

got subscription request as `GraphQLRequest` ---> 
subscribe() --->
   request processed ---> 
   subscription handler -> Result<Stream, Error> { 
         manager.get_stream("key") 
   } ---> 
return element when got some from stream --->
server handles how element is returned --->
end subscription when stream ends (`Async::Ready(None)`).

request.subscribe() always returns Stream. If subscription handler returns something that is not Stream it will be internally wrapped into a trivial Stream.

And RootNode will now have to be like that: RootNode<Queries, Mutations, Subscriptions> (breaking change).

[gklijs]

Seems fine. Would the stream be supported by a web socket, or are there other ideas to implement that part? In lucinia, a Clojure GraphQL server implementation a subscription is backed up with web sockets. It's for example possible to consume some data stream, and then send it by websocket to all the clients that subscribed to certain data. How would something like that work with juniper?

[theduke]

Just like the rest of Juniper, subscriptions should be agnostic over the transport medium.

It should only produce streams of messages. The support libraries for the different servers will handle implementing the actual transport, with websockets/ H2 server side events, long polling fallback, ....

We'll probably have something like a trait SubscriptionHandler that abstracts over the transport semantics.

[tyranron]

@gklijs

We're using actix-web and pairing its WebSockets with such juniper subscription abstraction is quite a convenient. As WebSocket connection represents an Actor itself, all we need is just to inject returned Stream into Actor's Context and then we're able to handle it just like regular actix Messages.

I suppose other web-framework will propose near the same "abstraction glue".

It's for example possible to consume some data stream, and then send it by websocket to all the clients that subscribed to certain data. How would something like that work with juniper?

This thing, however, I suppose should be implemented on the other side of juniper, where actual subscription handlers are declared.
I mean, the decision to reuse some channel for all subscribers or do not do that is just a schema implementation detail and should not be hardcoded into juniper abstractions.

[tyranron]

@theduke what do you think about design thoughts described above? Should we start poking with it to make some MVP at least?

[D1plo1d]

+1 to transport agnostic subscriptions. I use WebRTC data channels to send GraphQL subscriptions and a subscribe function that returns a stream would be perfect for me. I'd add that the message format that encapsulates the GraphQL responses (Apollo calls this GRAPHQL_WS if memory serves and we should aim to be compatible with this defacto standard) should also be implemented as a transport agnostic stream (perhaps it consumes and wraps the unencoded GraphQL stream like a Tokio Codec?).

…

On Tue, Sep 24, 2019, 7:41 AM Kai Ren ***@***.***> wrote: @theduke <https://github.com/theduke> what do you think about design thoughts described above? Should we start poking with it to make some MVP at least? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#54?email_source=notifications&email_token=AABDOIF5RNLYIJC2CEPN42TQLH4HFA5CNFSM4DPUQQP2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD7OB6OI#issuecomment-534519609>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AABDOIAFRAM3UJDTR75BDDLQLH4HFANCNFSM4DPUQQPQ> .

[gklijs]

Sounds entirely valid, library I was referring to can by used without middleware, but also has a 'default' implementation. Reason I'm asking all this is I would really like to compare my Clojure implementation with one in Rust.

[LegNeato]

I think a trait returning a Stream (and maybe optionally an AsyncStream from tokio?) would be a good way to start prototyping and playing around @tyranron.

Though would we need to support things like timeouts, backpressure, etc? If so it starts to look more like a Tower service.

[tyranron]

@LegNeato

Though would we need to support things like timeouts, backpressure, etc? If so it starts to look more like a Tower service.

I don't think so. juniper is all about GraphQL schema definition and resolving. Timeouts and backpressure are just implementation details on both sides of GraphQL schema (either select(request.execute(schema, context), timeout) when receiving HTTP request, or backpressure baked into Stream returning on resolving GraphQL subscription).

While on-top-of-juniper crates like (juniper_rocket, etc) may provide such capabilities/abstractions if necessary/useful.

[LegNeato]

This has landed on master! Please check it out and give feedback on usage. Thanks a ton @nWacky and @tyranron ! 🍻 🍾 ❤️