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

#[allow(dead_code)]
/// Abstracts crossbeam-channel and std::sync::mpsc
pub mod channel {
    #[cfg(feature = "crossbeam-channel")]
    mod inner {
        /// The sender side
        pub type Tx<T> = crossbeam_channel::Sender<T>;
        /// The receiver side
        pub type Rx<T> = crossbeam_channel::Receiver<T>;
        /// An error returned on Send
        pub use crossbeam_channel::SendError;
    }

    #[cfg(not(feature = "crossbeam-channel"))]
    mod inner {
        /// The sender side
        pub type Tx<T> = std::sync::mpsc::Sender<T>;
        /// The receiver side
        pub type Rx<T> = std::sync::mpsc::Receiver<T>;
        /// An error returned on Send
        pub use std::sync::mpsc::SendError;
    }

    pub use inner::*;

    #[cfg(feature = "crossbeam-channel")]
    /// Create an unbounded channel
    pub fn channel<T>() -> (Tx<T>, Rx<T>) {
        crossbeam_channel::unbounded()
    }

    #[cfg(not(feature = "crossbeam-channel"))]
    /// Create an unbounded channel
    pub fn channel<T>() -> (Tx<T>, Rx<T>) {
        std::sync::mpsc::channel()
    }
}

use channel::*;
use std::io::{Error, ErrorKind, Write};

/// A writer that allows sending messages to the client
pub type SyncWriter = crate::Encoder<SyncMpscWriter>;

/// A channel-based synchronous writer
pub struct SyncMpscWriter {
    tx: Tx<Vec<u8>>,
    buf: Vec<u8>,
}

impl std::fmt::Debug for SyncMpscWriter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SyncMpscWriter").finish()
    }
}

impl Clone for SyncMpscWriter {
    fn clone(&self) -> Self {
        Self {
            tx: self.tx.clone(),
            buf: Vec::new(),
        }
    }
}

impl SyncMpscWriter {
    /// Create a new SyncMpscWriter from this channel's sender
    pub fn new(tx: Tx<Vec<u8>>) -> Self {
        Self {
            tx,
            buf: Vec::new(),
        }
    }
}

impl Write for SyncMpscWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.buf.extend_from_slice(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        // if the receiver has been dropped, then we should return an error.
        self.tx
            .send(std::mem::take(&mut self.buf))
            .map_err(|_| Error::new(ErrorKind::UnexpectedEof, "cannot write, other half dropped"))
    }
}