obmarg / graphql-ws-client Goto Github PK

I'm not very happy with the API of graphql-ws-client at the moment:

1. Too many generic parameters everywhere
2. Integrating into new clients takes too much boilerplate
3. Dropping connections when the client drops makes sense but is... kind of surprising. It's not how most other clients work for example.

Plans to fix:

• Remove the GraphqlClient trait #58
• Possibly get rid of WsMessage as a generic param on the AsyncWebsocketClient #61
  • Want to expose something that transforms generic type to a concrete internal type. new function remains generic but the output won't be.
  • Could maybe combine send & sink into a single parameter w/ impls for splittable streams etc? Not sure, need to toy with it and see what is most ergonomic.
• Try to get rid of the runtime: impl SpawnExt parameter - could just return a future and let users deal with it instead. Probably less code in the end. #61
• Change the drop behaviour - clients and subscriptions should both keep the receiver_loop running, assuming the user has spawned it somewhere persistent. If the user wants the old behaviour they drop the returned future. #61
• Add a method to subscription streams that takes the connection future and runs it inside the stream, to handle the common case of wanting to run a single subscription per connection. #65
  • Discovered while trying this in a project that you still need to run the actor while making the subscription. It's a bit clumsy, may need more thought.
• While I'm going nuts - is AsyncWebsocketClient the best name? Not sure (not super keen to go changing that though) #61
• Update the various websocket impls
  • async tungstenite #62
  • ws-stream-wasm #64
• Module tidy up #67
  • Move stuff out of the next module
  • Remove the allow(unused) #68
  • Deprecate old interface
• Document new stuff (including doctests)
• Update the examples to use the new API
• Remove the default feature on async tungstenite #69
• Make graphql_client example & test
• Make a release candidate

Post Release Candidate

Can solicit feedback with release candidate and decide on these:

• API concerns
  • Think about whether client::streaming_operation needs a &mut self
  • Think about whether I even need the ping/pong messages
  • run function vs IntoFuture

Hi,
I am looking for some help with a specific usage pattern for graphql-ws-client.
I couldn't find the solution in the examples or on the interwebs.

I ll try and describe the setup with some psuedo-code.

I have the use-case of subscribing to 5 different queries. From the cynic-multiple-subscriptions example, it is recommended that I re-use the Client struct and call client.subscribe with as many queries as I have.
And then loop over the returned Stream

In my case, the job of subscribing and then looping over the potentially never-ending stream is done on separate threads for each query, and the updates are passed through mpsc channels.

fn subscribe_to(graphql_client: Client, query: StreamingOperation) {
  std::thread::spawn(|| {
    let stream = client.subscribe(query);
    while let Some(item) = stream.next().await {
      println!("{:?}", item);
    }
  }
  });
}

I also have a requirement of being able to close all subscriptions when a certain event occurs. For this I looked at the Client::close function.
Now the trouble is that since I need to pass in the graphql client value to multiple threads, I decided to wrap that in an Arc<RwLock<Client>>
RwLock because the subscribe method needs a mutable self so...

I also changed up the subscribe_to* functions to take an Arc, acquire a write lock before calling subscribe
The issue is, when I need to close the client, I need an inner owned value to pass to the close function.
So I tried something like Arc::into_inner(arc_rwlock_graphql_client)

But since every subscribe_to clones the Arc, the into_inner will always return None because that's the requirement of the Arc::into_inner fn

So my question is, what is the recommended way to implement such a pattern? Where I have to subscribe to multiple queries from multiple threads and also retain the ability to close all of them at once.

I have a very simple proposal for a small change to the builder pattern naming.
I had no idea that I could use a builder to customize the client until I started looking into the code.
Most implementations include a builder method that returns the builder, then a build function that generates an instance of the (built) class.
Because the builder function returns an instance of the builder and the build method returns an instance of the "buit" class, there is a distinction between the two.
To help new users understand that the client is not built directly but rather provides a builder that must subsequently be invoked with the build function in order to return an instance of the class, I suggest renaming this to builder.

I wrote this library primarily for cynic, but with the intention of supporting other client libraries also.

This issue is to track adding support for graphql_client - should just be a case of providing a GraphqlClient impl for it's types.

Currently this library just supports subscriptions via streaming_operation.

But GraphQL over websockets can also be used for normal queries & mutations. We should add support for these (via the commented out operation function in the code.

The client has an internal channel thats buffer size is currently controlled by a constant SUBSCRIPTION_BUFFER_SIZE. We should probably allow users to customise the size of this.

Probably want a ClientBuilder that allows these sorts of parameters to be set before construction of a client.

In #33 we discussed what the WebsocketMessage::error_message type should return. I'm not sure if it's important, but wanted to note it incase we want to come back to it.

Currently this library does not work with the latest version of cynic.

I changed https://github.com/obmarg/graphql-ws-client/blob/main/src/graphql.rs#L76 to also take Variables which allows the code based on the example to compile, but I'm unsure whether other changes are required.

Add derive Debug to Client

The client is using a few Arcs - this issue is to do some investigation to make sure we've not got any cycles that would cause memory leaks.

Not certain we do, but just want to make sure it has been thought about.

I think we should refactor examples to accommodate for WASM and graphql-client.
Probably using different packages.
What do you think?

I've seen that the channel is always bounded (default to 5).
What do you think if we also support unbounded channel (Option<usize>).
Of course we should rethink on how to wrap the Sender and Receiver but I think this should be a nice feature.

Update async-tungstenite to least version (0.26.1).

The crate is currently very rubbish at handling errors: there's unwraps all over the place, some errors are just ignored, the Error type is just an empty placeholder struct.

This issue is to track fixing those various issues.

• All the unwraps.
• The return inside sender_loop probably needs to do something

For example this cause panic:

        let mut stream = client
            .streaming_operation(WorkspaceVoiceVoicemail::build(()))
            .await;

        for _ in 0..5 {
            let data = stream
                .next()
                .await
                .unwrap()
                .data
                .unwrap();
        }

        stream = client
            .streaming_operation(WorkspaceVoiceVoicemail::build(()))
            .await;

        for _ in 0..5 {
            let data = stream
                .next()
                .await
                .unwrap()
                .data
                .unwrap();
        }

Somehow need send Complete to server when the stream is droping.

There are quite a few functions that would benefit from some doctest style examples. This issue is to track adding them.

• AsyncWebsocketClient & it's functions.

By the way, how use with tokio?
This code give me panic: thread 'main' panicked at 'Cannot drop a runtime in a context where blocking is not allowed. This happens when a runtime is dropped from within an asynchronous context.', #C:\Users\asinotov\.cargo\registry\src\github.com-1ecc6299db9ec823\tokio-1.6.0\src\runtime\blocking\shutdown.rs:51:21

#[tokio::main]
async fn main() {
    let mut request = "ws://localhost:5000/graphql".into_client_request().unwrap();
    request.headers_mut().insert(
        "Sec-WebSocket-Protocol",
        HeaderValue::from_str("graphql-transport-ws").unwrap(),
    );

    let (connection, res) = async_tungstenite::tokio::connect_async(request)
        .await
        .unwrap();

    let (sink, stream) = connection.split();

    assert_eq!(StatusCode::SWITCHING_PROTOCOLS, res.status());

    let mut _client = CynicClientBuilder::new()
        .payload("data")
        .build(
            stream,
            sink,
            async_executors::TokioTpBuilder::new().build().unwrap(),
        )
        .await
        .unwrap();
}

Haven't got round to writing any tests yet. This needs fixed before a release.

Also need CI etc (probably github actions?)

#23 (and it's predecessor #19) added a stop_operation function that can be called to send a Complete message when a subscription is done. There's some improvements that could be made though:

1. Stopped operations will still be in the operation map. Should figure out a way to remove them.
2. If you drop a StreamingOperation it'll never be stopped. Ideally need a way to run stop_operation on drop (currently it's an async operation which makes this tricky - might need a way to do this that doesn't rely on async/await
3. Not quite related, but I'm noting it down here anyway: if the server stopped the subscription we ideally need a way to tell how & why on the StreamingOperation itself.

The client would benefit from adding logs so we have a place to print out errors or warnigns when things go wrong. This is to track adding them.

For now lib support only graphql-transport-ws protocol.
May be add subscriptions-transport-ws as well (they very similar):
https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md

Hi there! As I was playing around with the repo, I noticed that AsyncWebsocketClientBuilder is not Send because it has a field called phantom of type PhantomData<*const GraphqlClient>. Since *const GraphqlClient is not Send (because it's a pointer), this causes the struct AsyncWebsocketClientBuilder itself to not be Send.

I think this isn't an issue with the examples in the repo because they use async_std::main/tokio::main which both internally use block_on under-the-hood. And block_on doesn't have Send requirements on the future, whereas the spawn equivalents do (tokio example here).

I think there's a few ways around this.

1. change the phantom type to PhantomData<fn() -> GraphqlClient>, which has the same variance/drop constraints as the current approach
2. explicitly mark AsyncWebsocketClientBuilder as Send via an unsafe impl Send for AsyncWebsocketClientBuilder

I think the first option is cleaner because the AsyncWebsocketClientBuilder will be Send for free if everything in it is also Send.

I'm happy to create a PR for this if you think one of these approaches is reasonable 😄

AsyncWebsocketClient::streaming_operation currently returns an anonymous Stream type. This is fine for reading the results of a subscription, but there's a few things it can't support:

1. Stopping the subscription (arguably dropping the stream might do this, although not sure if it's handled very cleanly).
2. Checking why a stream ended.

Ideally we'd return a custom type that implements Stream but also:

1. Provides a stop function.
2. Stops the subscription cleanly on drop.
3. Provides a way to check whether the subscription stopped "cleanly" or because of some sort of connectivity error etc.

This lib currently supports async-tungetenite for websockets. The intention was to support other clients though - this issue is to track adding that support.

• rust-websocket-lite
• #33

Hey there!
I built a prototype implementation for GraphQLClient (over here https://github.com/caido/graphql-ws-client).
Do you still plan on working on this lib, if so I will upstream my changes otherwise I will go my own way and most likely try to integrate it directly in GraphQLClient.

Like in #46, I'm getting this error:

error[E0277]: the trait bound `async_tungstenite::tungstenite::Message: WebsocketMessage` is not satisfied
   --> src/main.rs:154:22
    |
154 |                     .build(stream, sink, TokioSpawner::current())
    |                      ^^^^^ the trait `WebsocketMessage` is not implemented for `async_tungstenite::tungstenite::Message`
    |
    = help: the trait `WebsocketMessage` is implemented for `tungstenite::protocol::message::Message`
note: required by a bound in `AsyncWebsocketClientBuilder::<GraphqlClient, Payload>::build`

However, unlike #46, I'm using graphql-ws-client 0.4.0 and async-tungstenite 0.22.2. I get this with both the Tokio example in the latest main branch and at the 0.4.0 tag. It looks like this might be a library compatibility issue again? I'm not sure how to resolve it; can you advise?

I'm encountering this error:

error[E0277]: the trait bound `Message: WebsocketMessage` is not satisfied
   --> src/subscription.rs:69:10
    |
69  |         .build(stream, sink, as...
    |          ^^^^^ the trait `WebsocketMessage` is not implemented for `Message`
    |

when using the example in https://github.com/obmarg/graphql-ws-client/blob/main/examples/examples/subscriptions.rs.

I'm using the examples as-is, but perhaps I'm missing something obvious?