use std::pin::Pin;use std::time::Duration; use anyhow::Result;use cloud_api_types::websocket_protocol::MessageToClient;use futures::channel::mpsc::unbounded;use futures::stream::{SplitSink, SplitStream};use futures::{FutureExt as _, SinkExt as _, Stream, StreamExt as _, TryStreamExt as _, pin_mut};use gpui::{App, BackgroundExecutor, Task};use yawc::WebSocket;use yawc::frame::{FrameView, OpCode}; const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1); pub type MessageStream = Pin<Box<dyn Stream<Item = Result<MessageToClient>>>>; pub struct Connection { tx: SplitSink<WebSocket, FrameView>, rx: SplitStream<WebSocket>,} impl Connection { pub fn new(ws: WebSocket) -> Self { let (tx, rx) = ws.split(); Self { tx, rx } } pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) { let (mut tx, rx) = (self.tx, self.rx); let (message_tx, message_rx) = unbounded(); let handle_io = |executor: BackgroundExecutor| async move { // Send messages on this frequency so the connection isn't closed. let keepalive_timer = executor.timer(KEEPALIVE_INTERVAL).fuse(); futures::pin_mut!(keepalive_timer); let rx = rx.fuse(); pin_mut!(rx); loop { futures::select_biased! { _ = keepalive_timer => { let _ = tx.send(FrameView::ping(Vec::new())).await; keepalive_timer.set(executor.timer(KEEPALIVE_INTERVAL).fuse()); } frame = rx.next() => { let Some(frame) = frame else { break; }; match frame.opcode { OpCode::Binary => { let message_result = MessageToClient::deserialize(&frame.payload); message_tx.unbounded_send(message_result).ok(); } OpCode::Close => { break; } _ => {} } } } } }; let task = cx.spawn(async move |cx| handle_io(cx.background_executor().clone()).await); (message_rx.into_stream().boxed(), task) }}
use std::pin::Pin;
use std::time::Duration;
use anyhow::Result;
use cloud_api_types::websocket_protocol::MessageToClient;
use futures::channel::mpsc::unbounded;
use futures::stream::{SplitSink, SplitStream};
use futures::{FutureExt as _, SinkExt as _, Stream, StreamExt as _, TryStreamExt as _, pin_mut};
use gpui::{App, BackgroundExecutor, Task};
use yawc::WebSocket;
use yawc::frame::{FrameView, OpCode};
const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1);
pub type MessageStream = Pin<Box<dyn Stream<Item = Result<MessageToClient>>>>;
pub struct Connection {
tx: SplitSink<WebSocket, FrameView>,
rx: SplitStream<WebSocket>,
}
impl Connection {
pub fn new(ws: WebSocket) -> Self {
let (tx, rx) = ws.split();
Self { tx, rx }
}
pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) {
let (mut tx, rx) = (self.tx, self.rx);
let (message_tx, message_rx) = unbounded();
let handle_io = |executor: BackgroundExecutor| async move {
// Send messages on this frequency so the connection isn't closed.
let keepalive_timer = executor.timer(KEEPALIVE_INTERVAL).fuse();
futures::pin_mut!(keepalive_timer);
let rx = rx.fuse();
pin_mut!(rx);
loop {
futures::select_biased! {
_ = keepalive_timer => {
let _ = tx.send(FrameView::ping(Vec::new())).await;
keepalive_timer.set(executor.timer(KEEPALIVE_INTERVAL).fuse());
}
frame = rx.next() => {
let Some(frame) = frame else {
break;
};
match frame.opcode {
OpCode::Binary => {
let message_result = MessageToClient::deserialize(&frame.payload);
message_tx.unbounded_send(message_result).ok();
}
OpCode::Close => {
break;
}
_ => {}
}
}
}
}
};
let task = cx.spawn(async move |cx| handle_io(cx.background_executor().clone()).await);
(message_rx.into_stream().boxed(), task)
}
}