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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
//! Remote channels.
//!
//! This module contains channels that can be used to exchange data of
//! arbitrary type with a remote endpoint.
//!
//! The data type must be serializable and deserializable by [serde] and
//! [sendable across thread boundaries](Send).
//! The [RemoteSend](crate::RemoteSend) trait is implemented automatically
//! for all types that fulfill these requirements.
//!
//! # Establishing a channel connection with a remote endpoint
//!
//! Except for a [base channel](base), a channel connection is established by sending
//! the Sender or Receiver half of a channel to a remote endpoint over an already
//! existing channel.
//! The transmitted channel-half can also be part of a larger object, such as a struct, tuple or enum.
//! Most channel types can even be forwarded over multiple connections.
//!
//! The primary purpose of a [base channel](base) is to provide an initial channel after
//! establishing a connection over a physical transport.
//! In most cases there is no need to create it directly and you will probably only use it
//! after it has been returned from an [initial connect function](crate::Connect).
//!
//! # Channel types
//!
//! [Broadcast](broadcast), [MPSC](mpsc), [oneshot] and [watch] channels are closely
//! modelled after the channels found in [tokio::sync].
//! They also work when both halves of the channel are local and can be forwarded
//! over multiple connections.
//!
//! An [local/remote channel](lr) is a more restricted version of an [MPSC](mpsc) channel.
//! It does not support forwarding and exactly one half of it must be on a remote endpoint.
//! Its benefit is, that it does not spawn an async task and thus uses less resources.
//! When in doubt use an [MPSC channel](mpsc) instead.
//!
//! A [binary channel](bin) can be used to exchange binary data over a channel.
//! It skips serialization and deserialization and thus is more efficient for binary data,
//! especially when using text codecs such as JSON.
//! It does support forwarding.
//! However, at least one half of it must be on a remote endpoint.
//!
//! # Acknowledgements and connection latency
//!
//! The channels do not wait for acknowledgement of transmitted values.
//! Thus, the next value can be queued for transmission before the previous value
//! has been received by the remote endpoint.
//! Hence, throughput is unaffected by the connection roundtrip time.
//!
//! A successful send does not guarantee that the transmitted value will be received
//! by the remote endpoint.
//! However, values cannot be lost intermittently, i.e. if a transmission error
//! occurs on the underlying physical connection the whole [chmux] connection will
//! be terminated eventually and sending will fail.
//!
//! If you need confirmation for every sent value, consider sending a [`oneshot::Sender`]`<()>`
//! along with it (for example as a tuple).
//! The remote endpoint can then send back an empty message as confirmation over the oneshot channel
//! after it has processed the received value.
//!
//! # Size considerations
//!
//! The size of the objects exchanged is not limited.
//! If the object is small enough it is first serialized into a buffer and then sent as
//! one message to the remote endpoint.
//! For larger objects, a serialization thread is spawned and the message is sent
//! chunk-by-chunk to avoid the need for a large temporary memory buffer.
//! Deserialization is also performed on-the-fly as data is received for large objects.
//!
//! Large values are always sent in chunks, so that one channel does not block other channels
//! for a long period of time.
//!
//! To avoid denial of service attacks when you exchange data with untrusted remote endpoints,
//! make sure that an object cannot grow to infinite size during deserialization.
//! For example, avoid using containers of unlimited size like [Vec] or [HashMap](std::collections::HashMap)
//! in your data structure or limit their maximum size during deserialization by applying
//! the `#[serde(deserialize_with="...")]` attribute to the fields containing them.
//!
//! # Static buffer size configuration for received channel-halves
//!
//! This is used to specify the local buffer size (in items) via a const generic type parameter
//! when the sender or receiver half of a channel *is received*.
//!
//! The default buffer size is [DEFAULT_BUFFER], which is currently 2 items.
//! It can be increased to improve performance, but this will also increase
//! the maximum amount of memory used per channel.
//! Setting the buffer size too high must be avoided, since this can lead to
//! the program running out of memory.
//!
//! # Error handling
//!
//! During sending it is often useful to treat errors that occur due to the disconnection, either graceful
//! or non-graceful, of the remote endpoint as an indication that the receiver is not
//! interested anymore in the transmitted data, rather than a hard error.
//! To avoid complicated error checks in your code, you can use [SendResultExt::into_disconnected]
//! to query whether a send error occurred due to the above cause and exit the sending process gracefully.
//!

use serde::{Deserialize, Serialize};
use std::{error::Error, fmt};

use crate::chmux;

mod interlock;

pub mod base;
pub mod bin;
pub mod broadcast;
//pub mod buffer;
pub mod lr;
pub mod mpsc;
pub mod oneshot;
pub mod watch;

/// Error connecting a remote channel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ConnectError {
    /// The corresponding sender or receiver has been dropped.
    Dropped,
    /// Error initiating chmux connection.
    Connect(chmux::ConnectError),
    /// Error listening for or accepting chmux connection.
    Listen(chmux::ListenerError),
}

impl fmt::Display for ConnectError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            Self::Dropped => write!(f, "other part was dropped"),
            Self::Connect(err) => write!(f, "connect error: {err}"),
            Self::Listen(err) => write!(f, "listen error: {err}"),
        }
    }
}

impl Error for ConnectError {}

/// Reason for closure of sender.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum ClosedReason {
    /// Channel was closed because receiver has been closed.
    Closed,
    /// Channel was closed because receiver has been dropped.
    Dropped,
    /// Channel was closed because connection between sender and receiver failed.
    Failed,
}

/// Back channel message that receiver has been closed.
pub(crate) const BACKCHANNEL_MSG_CLOSE: u8 = 0x01;

/// Back channel message that error has occurred.
pub(crate) const BACKCHANNEL_MSG_ERROR: u8 = 0x02;

/// Remote sending error.
#[derive(Clone)]
pub(crate) enum RemoteSendError {
    /// Send error.
    Send(base::SendErrorKind),
    /// Connect error.
    Connect(chmux::ConnectError),
    /// Listen for connect error.
    Listen(chmux::ListenerError),
    /// Forwarding error.
    Forward,
    /// Receiver was closed.
    Closed,
}

/// Common functions to query send errors for details.
///
/// This is implemented for all send errors in this module.
///
/// Since [watch] channels have no method to close them,
/// [is_closed](Self::is_closed) and [is_disconnected](Self::is_disconnected) are
/// equivalent for them.
pub trait SendErrorExt {
    /// Whether the remote endpoint closed the channel.
    fn is_closed(&self) -> bool;

    /// Whether the remote endpoint closed the channel, was dropped or the connection failed.
    fn is_disconnected(&self) -> bool;

    /// Whether the error is final, i.e. no further send operation can succeed.
    fn is_final(&self) -> bool;

    /// Whether the error is caused by the item to be sent.
    fn is_item_specific(&self) -> bool;
}

/// Common functions to query results of send operations for details.
///
/// This is implemented by all results from send operations in this module.
pub trait SendResultExt {
    /// The error type of this result.
    type Err: SendErrorExt;

    /// Whether the remote endpoint closed the channel.
    ///
    /// Returns `Ok(false)` if the send was successful and `Ok(true)` if the send failed for the
    /// above reason.
    /// The original error is returned if the failure had another cause.
    ///
    /// # Example
    ///
    /// In the following example the client counts upwards and the server closes the connections once
    /// it receives the number 5.
    /// The client stops counting as soon as sending fails.
    /// Note that the panic error path in the client would only be triggered by another error,
    /// such as a serialization error.
    ///
    /// ```
    /// use remoc::prelude::*;
    ///
    /// // This would be run on the client.
    /// async fn client(mut tx: rch::base::Sender<u32>) {
    ///     let mut n = 0;
    ///
    ///     loop {
    ///         if tx.send(n).await.into_closed().unwrap() {
    ///             break;
    ///         }
    ///         n += 1;
    ///     }
    ///
    ///     assert!(n > 5);
    /// }
    ///
    /// // This would be run on the server.
    /// async fn server(mut rx: rch::base::Receiver<u32>) {
    ///     while let Some(n) = rx.recv().await.unwrap() {
    ///         if n == 5 {
    ///             rx.close().await;
    ///         }
    ///     }
    /// }
    /// # tokio_test::block_on(remoc::doctest::client_server(client, server));
    /// ```
    fn into_closed(self) -> Result<bool, Self::Err>;

    /// Whether the remote endpoint closed the channel, was dropped or the connection failed.
    ///
    /// Returns `Ok(false)` if the send was successful and `Ok(true)` if the send failed for the
    /// above reasons.
    /// The original error is returned if the failure had another cause.
    ///
    /// # Example
    ///
    /// In the following example the client counts upwards and the server drops the connections once
    /// it receives the number 5.
    /// The client stops counting as soon as sending fails.
    /// Note that the panic error path in the client would only be triggered by another error,
    /// such as a serialization error.
    ///
    /// ```
    /// use remoc::prelude::*;
    ///
    /// // This would be run on the client.
    /// async fn client(mut tx: rch::base::Sender<u32>) {
    ///     let mut n = 0;
    ///
    ///     loop {
    ///         if tx.send(n).await.into_disconnected().unwrap() {
    ///             break;
    ///         }
    ///         n += 1;
    ///     }
    ///
    ///     assert!(n > 5);
    /// }
    ///
    /// // This would be run on the server.
    /// async fn server(mut rx: rch::base::Receiver<u32>) {
    ///     while let Some(n) = rx.recv().await.unwrap() {
    ///         if n == 5 {
    ///             break;
    ///         }
    ///     }
    /// }
    /// # tokio_test::block_on(remoc::doctest::client_server(client, server));
    /// ```
    fn into_disconnected(self) -> Result<bool, Self::Err>;
}

impl<E> SendResultExt for Result<(), E>
where
    E: SendErrorExt,
{
    type Err = E;

    fn into_closed(self) -> Result<bool, E> {
        match self {
            Ok(()) => Ok(false),
            Err(err) => {
                if err.is_closed() {
                    Ok(true)
                } else {
                    Err(err)
                }
            }
        }
    }

    fn into_disconnected(self) -> Result<bool, E> {
        match self {
            Ok(()) => Ok(false),
            Err(err) => {
                if err.is_disconnected() {
                    Ok(true)
                } else {
                    Err(err)
                }
            }
        }
    }
}

/// Default buffer size in items, when the sender or receiver half of a channel *is received*.
///
/// This can be changed via a const generic type parameter of a sender or receiver.
///
/// The current default buffer size is 2.
pub const DEFAULT_BUFFER: usize = 2;

/// Default maximum allowed item size for sending and receiving items over remote channels.
///
/// The current default maximum allowed item size is 16 MB.
pub const DEFAULT_MAX_ITEM_SIZE: usize = 16_777_216;