1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
//! Channel for users to send messages when calling [`Shard::send`] isn't
//! possible.
//!
//! [`Shard::send`]: crate::Shard::send
use crate::{
CloseFrame,
command::Command,
error::{ChannelError, ChannelErrorType},
json,
};
use tokio::sync::mpsc;
/// Channel between a user and shard for sending outgoing gateway messages.
#[derive(Debug)]
pub struct MessageChannel {
/// Receiving half for shards to receive users' close frames.
pub close_rx: mpsc::Receiver<CloseFrame<'static>>,
/// Sending half for users to send close frames via shards.
pub close_tx: mpsc::Sender<CloseFrame<'static>>,
/// Receiving half for shards to receive users' commands.
pub command_rx: mpsc::UnboundedReceiver<String>,
/// Sending half for users to send commands via shards.
pub command_tx: mpsc::UnboundedSender<String>,
}
impl MessageChannel {
/// Initialize a new message channel.
pub fn new() -> Self {
let (command_tx, command_rx) = mpsc::unbounded_channel();
let (close_tx, close_rx) = mpsc::channel(1);
Self {
close_rx,
close_tx,
command_rx,
command_tx,
}
}
/// Clone of the senders.
pub fn sender(&self) -> MessageSender {
MessageSender {
close: self.close_tx.clone(),
command: self.command_tx.clone(),
}
}
}
/// Channel to send messages over a [`Shard`] to the Discord gateway.
///
/// [`Shard`]: crate::Shard
#[derive(Clone, Debug)]
pub struct MessageSender {
/// Sending half of the close channel.
close: mpsc::Sender<CloseFrame<'static>>,
/// Sending half of the command channel.
command: mpsc::UnboundedSender<String>,
}
impl MessageSender {
/// Whether the channel is closed.
///
/// The channel will only be closed if the associated shard has been
/// dropped.
pub fn is_closed(&self) -> bool {
self.command.is_closed()
}
/// Send a command to the associated shard.
///
/// # Errors
///
/// Returns a [`ChannelErrorType::Closed`] error type if the channel is
/// closed.
#[allow(clippy::missing_panics_doc)]
pub fn command(&self, command: &impl Command) -> Result<(), ChannelError> {
self.send(json::to_string(command).expect("serialization cannot fail"))
}
/// Send a JSON encoded gateway event to the associated shard.
///
/// # Errors
///
/// Returns a [`ChannelErrorType::Closed`] error type if the channel is
/// closed.
pub fn send(&self, json: String) -> Result<(), ChannelError> {
self.command.send(json).map_err(|source| ChannelError {
kind: ChannelErrorType::Closed,
source: Some(Box::new(source)),
})
}
/// Send a Websocket close frame to the associated shard.
///
/// Subsequent calls may be queued up to be sent once the shard's
/// reestablished a Websocket connection or ignored if the queue is full.
/// The internal queue capacity is currently `1`.
///
/// See the [`Shard::close`] docs for further information.
///
/// # Errors
///
/// Returns a [`ChannelErrorType::Closed`] error type if the channel is
/// closed.
///
/// [`Shard::close`]: crate::Shard::close
pub fn close(&self, close_frame: CloseFrame<'static>) -> Result<(), ChannelError> {
if let Err(source @ mpsc::error::TrySendError::Closed(_)) = self.close.try_send(close_frame)
{
Err(ChannelError {
kind: ChannelErrorType::Closed,
source: Some(Box::new(source)),
})
} else {
Ok(())
}
}
}
#[cfg(test)]
mod tests {
use super::{MessageChannel, MessageSender};
use static_assertions::assert_impl_all;
use std::fmt::Debug;
assert_impl_all!(MessageChannel: Debug, Send, Sync);
assert_impl_all!(MessageSender: Clone, Debug, Send, Sync);
}