fix: add .subscribe / async support to wasm client #268
Conversation
dmccartney
force-pushed
the
daniel-subscribe-async
branch
3 times, most recently
from
5c2178c to
734b314
Compare
xmtp_api_grpc_gateway/src/lib.rs
Outdated
| vec![] | ||
| // HACK: this consumes from the stream whatever is already ready. | ||
| // TODO: implement a JS-friendly promise/future interface instead | ||
| async fn get_messages(&mut self) -> Vec<Envelope> { |
There was a problem hiding this comment.
After much-mashing to get this working, I think we should scrap this and change the abstraction for XmtpApiSubscription across the library. I suspect it will be cleaner either as a proper Stream (i.e. ✂️ the get_messages() stuff and use .collect() and other stream tools) and/or by designing outside-in to use JS promises (qua rust Futures) and callbacks.
There was a problem hiding this comment.
I have no objections to this. The stream polling stuff was a huge hack to make things work quickly on iOS and clearly there is a better way.
| // HACK: this tugs at the stream w/ poll_next() to initiate the HTTP req. | ||
| // we need to do this _before_ we publish, otherwise the | ||
| // publish will happen before the subscription is ready. | ||
| // TODO: implement a JS-friendly promise/future interface instead | ||
| let mut cx = Context::from_waker(noop_waker_ref()); | ||
| let stream = sub.stream.as_mut().unwrap(); | ||
| let init = stream.as_mut().poll_next(&mut cx); | ||
| assert_eq!(true, init.is_pending()); |
dmccartney
force-pushed
the
daniel-subscribe-async
branch
from
734b314 to
82855eb
Compare
dmccartney
force-pushed
the
daniel-subscribe-async
branch
from
82855eb to
0e740b7
Compare
| pub trait XmtpApiSubscription { | ||
| fn is_closed(&self) -> bool; | ||
| fn get_messages(&self) -> Vec<Envelope>; | ||
| fn close_stream(&mut self); | ||
| } | ||
| pub trait XmtpApiSubscription {} | ||
|
|
||
| impl<T: Stream<Item = Envelope>> XmtpApiSubscription for T {} |
There was a problem hiding this comment.
Beyond the WASM implementation, this is conceptually the big change to subscribe() across both grpc and grpc-gateway clients. Now the result is just an ordinary Stream of Envelopes.
There was a problem hiding this comment.
Do we need the XmtpApiSubscription trait at all, seeing as it's empty? Can we just mandate that subscribe() returns a Stream instead of an XmtpApiSubscription? I think that lets us call the Stream methods on consumers even if the concrete type for XmtpApiClient is not known.
| pub struct Client { | ||
| pub struct XmtpGrpcClient { |
There was a problem hiding this comment.
I renamed to be a little easier to understand alongside the other XmtpApiClient:
xmtp_api_grpc -> XmtpGrpcClient
xmtp_api_grpc_gateway -> XmtpGrpcGatewayClient
| impl XmtpApiClient for Client { | ||
| type Subscription = Subscription; | ||
| impl XmtpApiClient for XmtpGrpcClient { | ||
| type Subscription = BoxStream<'static, Envelope>; |
There was a problem hiding this comment.
If this has a static lifetime, are we sure the memory for these envelopes is ever going to get released?
| // grpc-gateway streams newline-delimited JSON bodies | ||
| let response = self | ||
| .http | ||
| .post(&format!("{}/message/v1/subscribe", self.url)) |
There was a problem hiding this comment.
This doesn't look like it needs to be a reference, does it?
| .await | ||
| .expect("published"); | ||
|
|
||
| if let Ok(env) = rx.await { |
There was a problem hiding this comment.
Nooby question, but what is the point of the channel here? If we are awaiting anyway, what difference does it make if we do rx.await rather than stream.next().await directly?
| // Discard any messages that we can't parse. | ||
| // TODO: consider surfacing these in a log somewhere | ||
| .filter_map(|r| async move { r.ok() }) | ||
| .boxed(); |
There was a problem hiding this comment.
Another nooby question - why do we need boxed() here? Can we use tonic::Streaming<Envelope> instead of BoxStream<'static, Envelope>?
|
This looks much better, thanks for cleaning it up! Just left some nooby questions about Streams |
This picks up where #222 left off implementing
Subscribein the wasm/grpc-gateway client.