bybit/ws/
handle.rs

1use tokio::sync::mpsc;
2
3use crate::{
4    Topic,
5    ws::{self, Command, OutgoingMessage},
6};
7
8/// Sends commands to a running [`Stream`](crate::ws::Stream); cheap to clone.
9#[derive(Debug, Clone)]
10pub struct Handle {
11    cmd_tx: mpsc::Sender<Command>,
12}
13
14impl Handle {
15    /// Wrap the command sender of a driver; normally obtained from [`Stream::new`](crate::ws::Stream::new).
16    pub fn new(cmd_tx: mpsc::Sender<Command>) -> Self {
17        Self { cmd_tx }
18    }
19
20    /// Open the connection (no-op if already connected).
21    pub async fn connect(&self) -> Result<(), ws::Error> {
22        self.send(Command::Connect).await
23    }
24
25    /// Like [`Handle::connect`], but fails with [`Error::QueueFull`](crate::ws::Error::QueueFull) instead of waiting.
26    pub fn try_connect(&self) -> Result<(), ws::Error> {
27        self.try_send(Command::Connect)
28    }
29
30    /// Close the connection; the driver stays alive and can connect again.
31    pub async fn disconnect(&self) -> Result<(), ws::Error> {
32        self.send(Command::Disconnect).await
33    }
34
35    /// Like [`Handle::disconnect`], but fails with [`Error::QueueFull`](crate::ws::Error::QueueFull) instead of waiting.
36    pub fn try_disconnect(&self) -> Result<(), ws::Error> {
37        self.try_send(Command::Disconnect)
38    }
39
40    /// Send a raw message on the current connection (not restored after a reconnect).
41    pub async fn send_command(&self, msg: OutgoingMessage) -> Result<(), ws::Error> {
42        self.send(Command::Send(msg)).await
43    }
44
45    /// Like [`Handle::send_command`], but fails with [`Error::QueueFull`](crate::ws::Error::QueueFull) instead of waiting.
46    pub fn try_send_command(&self, msg: OutgoingMessage) -> Result<(), ws::Error> {
47        self.try_send(Command::Send(msg))
48    }
49
50    /// Subscribe to `topics` now and again after every reconnect.
51    pub async fn subscribe(&self, topics: Vec<Topic>) -> Result<(), ws::Error> {
52        self.send(Command::Subscribe(topics)).await
53    }
54
55    /// Like [`Handle::subscribe`], but fails with [`Error::QueueFull`](crate::ws::Error::QueueFull) instead of waiting.
56    pub fn try_subscribe(&self, topics: Vec<Topic>) -> Result<(), ws::Error> {
57        self.try_send(Command::Subscribe(topics))
58    }
59
60    /// Unsubscribe from `topics` and stop restoring them.
61    pub async fn unsubscribe(&self, topics: Vec<Topic>) -> Result<(), ws::Error> {
62        self.send(Command::Unsubscribe(topics)).await
63    }
64
65    /// Like [`Handle::unsubscribe`], but fails with [`Error::QueueFull`](crate::ws::Error::QueueFull) instead of waiting.
66    pub fn try_unsubscribe(&self, topics: Vec<Topic>) -> Result<(), ws::Error> {
67        self.try_send(Command::Unsubscribe(topics))
68    }
69
70    async fn send(&self, cmd: Command) -> Result<(), ws::Error> {
71        self.cmd_tx
72            .send(cmd)
73            .await
74            .map_err(|_| ws::Error::DriverGone)
75    }
76
77    fn try_send(&self, cmd: Command) -> Result<(), ws::Error> {
78        self.cmd_tx.try_send(cmd).map_err(map_try_send_err)
79    }
80}
81
82fn map_try_send_err(e: mpsc::error::TrySendError<Command>) -> ws::Error {
83    match e {
84        mpsc::error::TrySendError::Full(_) => ws::Error::QueueFull,
85        mpsc::error::TrySendError::Closed(_) => ws::Error::DriverGone,
86    }
87}
bybit/ws/handle.rs

bybit/ws/
handle.rs
