feat(lazer): add resilient client in rust #2859
Conversation
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎
|
lazer/sdk/rust/client/src/client.rs
Outdated
| /// * `access_token` - The access token for authentication | ||
| /// * `num_connections` - The number of WebSocket connections to maintain | ||
| pub fn new( | ||
| endpoints: Vec<String>, |
| let (sender, receiver) = mpsc::channel::<AnyResponse>(CHANNEL_CAPACITY); | ||
|
|
||
| for i in 0..self.num_connections { | ||
| let endpoint = self.endpoints[i % self.endpoints.len()].clone(); |
There was a problem hiding this comment.
It will panic if endpoints is empty. We should probably check for that in the constructor.
lazer/sdk/rust/client/src/client.rs
Outdated
| } | ||
|
|
||
| pub async fn start(&mut self) -> Result<mpsc::Receiver<AnyResponse>> { | ||
| let (sender, receiver) = mpsc::channel::<AnyResponse>(CHANNEL_CAPACITY); |
There was a problem hiding this comment.
Channel capacity should be configurable by the caller.
There was a problem hiding this comment.
Done. Internal channel don't need to be configurable right?
lazer/sdk/rust/client/src/client.rs
Outdated
| } | ||
|
|
||
| let streams: Vec<_> = self.receivers.drain(..).map(ReceiverStream::new).collect(); | ||
| let mut merged_stream = stream::select_all(streams); |
There was a problem hiding this comment.
I think it's easier and more efficient to use a single channel here.
There was a problem hiding this comment.
right. fixed.
| if last_failure_time.elapsed() > BACKOFF_RESET_DURATION { | ||
| self.backoff.reset(); | ||
| } |
There was a problem hiding this comment.
This feels like a hack. I think we should reset the backoff object in start() when we successfully connect and subscribe.
There was a problem hiding this comment.
I want the behavior to be a little different than that. For example, if client can connect and subscribe, but get disconnected immediately after we should not reset.
| request: SubscribeRequest, | ||
| ) -> Result<()> { | ||
| self.subscriptions.push(request.clone()); | ||
| return ws_connection.subscribe(request).await; |
| request.subscription_id | ||
| ); | ||
| } | ||
| return ws_connection.unsubscribe(request).await; |
| endpoints: Vec<String>, | ||
| access_token: String, | ||
| num_connections: usize, | ||
| backoff: ExponentialBackoff, |
There was a problem hiding this comment.
this seems to be part of your public interface. you need to reexport it to downstream consumers. Other thing is to have a wrapper that has what you need (because i see you are not supporting everything there)
There was a problem hiding this comment.
added a builder wrapper.
Summary
This PR updates the rust client to be a resilient client with multiple connections.
Rationale
We need this both for consumers and for monitor service.
How has this been tested?
Manually ran and tested the connections.