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.

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

[rpc module]: SubscriptionSink more user-friendly API for sending items from stream. #627

Closed
Tracked by #573
niklasad1 opened this issue Dec 23, 2021 · 0 comments · Fixed by #639
Closed
Tracked by #573

Comments

@niklasad1
Copy link
Member

niklasad1 commented Dec 23, 2021

We have noticed the API for SubscriptionSink is a bit messy to use if you want to await for stream to be sent via the SubscriptionSink if the connection has already been closed it might leak until the stream/fut "resolves" because the stream itself doesn't know anything of the connection state.

For example given that we have the following impl:

impl MyServerTrait for () 
{
	fn subscribe_to_something(&self, mut sink: SubscriptionSink) -> RpcResult<()> {
                // assume this stream lives forever and produces items "rarely"...
		let stream = create_stream();

		let fut = async move {
			stream
				.take_while(|sc| future::ready(sink.send(sc).map_or(false, |_| true)))
				.for_each(|_| future::ready(()))
				.await
		};

	       tokio::spawn(fut);
	}
}

If connection has already this case the tokio task will live until sink.send fails.
To be on the safe-side you would need to something like:

impl MyServerTrait for () {
    fn subscribe_to_something(&self, mut sink: SubscriptionSink) -> RpcResult<()> {
        // assume this stream lives forever and produces items "rarely"...
        let stream = create_stream();

        let fut = async move {
            loop {
                let timeout = tokio::time::sleep(std::time::Duration::from_secs(60));
                tokio::pin!(timeout);

                tokio::select! {
                    Some(item) = stream.next() => {
                        if let Err(e) = sink.send(&item) {
                          break;
                        }
                    },
                    _ = &mut timeout => {
                        if sink.is_closed() {
                            break;
                        }
                    }
                    else => break,
                };
            }
        };

        tokio::spawn(fut);
    }
}
@niklasad1 niklasad1 changed the title [rpc module]: SubscriptionSink more user-friendly API for sending items from stream or future. [rpc module]: SubscriptionSink more user-friendly API for sending items from stream. Jan 12, 2022
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging a pull request may close this issue.

1 participant