tcp_channel/
sender.rs

1use std::io::{BufWriter, Write};
2use std::marker::PhantomData;
3use std::net::{TcpStream, ToSocketAddrs};
4
5use bincode::Config;
6use byteorder::WriteBytesExt;
7use serde::Serialize;
8
9use crate::{ChannelSend, Endian, BigEndian, SendError};
10
11/// The sending side of a channel.
12pub struct Sender<T: Serialize, E: Endian, W: Write = BufWriter<TcpStream>> {
13    writer: W,
14    config: Config,
15    _marker: PhantomData<(T, E)>,
16}
17
18/// A more convenient way of initializing senders.
19pub struct SenderBuilder;
20
21pub struct TypedSenderBuilder<T, W, E> {
22    _marker: PhantomData<(T, W, E)>,
23}
24
25impl SenderBuilder {
26    /// Begin building a new, buffered channel.
27    pub fn new() -> TypedSenderBuilder<(), BufWriter<TcpStream>, BigEndian> {
28        Self::buffered()
29    }
30    /// Begin building a new, buffered channel.
31    pub fn buffered() -> TypedSenderBuilder<(), BufWriter<TcpStream>, BigEndian> {
32        TypedSenderBuilder {
33            _marker: PhantomData,
34        }
35    }
36    /// Begin building a new, non-buffered channel.
37    pub fn realtime() -> TypedSenderBuilder<(), TcpStream, BigEndian> {
38        TypedSenderBuilder {
39            _marker: PhantomData,
40        }
41    }
42}
43impl<T, W, E> TypedSenderBuilder<T, W, E> {
44    /// Specify the type to send.
45    pub fn with_type<U: Serialize>(self) -> TypedSenderBuilder<U, W, E> {
46        TypedSenderBuilder {
47            _marker: PhantomData,
48        }
49    }
50    /// Specify the underlying writer type.
51    pub fn with_writer<X: Write>(self) -> TypedSenderBuilder<T, X, E> {
52        TypedSenderBuilder {
53            _marker: PhantomData,
54        }
55    }
56    /// Specify the endianness.
57    pub fn with_endianness<F: Endian>(self) -> TypedSenderBuilder<T, W, F> {
58        TypedSenderBuilder {
59            _marker: PhantomData,
60        }
61    }
62}
63impl<T: Serialize, W: Write, E: Endian> TypedSenderBuilder<T, W, E> {
64    /// Initialize the sender with the current variables.
65    pub fn build(self, writer: W) -> Sender<T, E, W> {
66        Sender {
67            _marker: PhantomData,
68            writer,
69            config: E::config(),
70        }
71    }
72}
73impl<T: Serialize, E: Endian> TypedSenderBuilder<T, BufWriter<TcpStream>, E> {
74    /// Connect to a listening receiver, at a specified address.
75    pub fn connect<A: ToSocketAddrs>(self, address: A) -> std::io::Result<Sender<T, E, BufWriter<TcpStream>>> {
76        let stream = TcpStream::connect(address)?;
77
78        Ok(Sender {
79            writer: BufWriter::new(stream),
80            _marker: PhantomData,
81            config: E::config(),
82        })
83    }
84}
85impl<T: Serialize, E: Endian> TypedSenderBuilder<T, TcpStream, E> {
86    /// Connect to a listening receiver, at a specified address.
87    pub fn connect<A: ToSocketAddrs>(self, address: A) -> std::io::Result<Sender<T, E, TcpStream>> {
88        let stream = TcpStream::connect(address)?;
89        stream.set_nodelay(true)?;
90
91        Ok(Sender {
92            writer: stream,
93            _marker: PhantomData,
94            config: E::config(),
95        })
96    }
97}
98impl<T: Serialize, E: Endian, W: Write> Sender<T, E, W> {
99    pub fn flush(&mut self) -> std::io::Result<()> {
100        self.writer.flush()
101    }
102}
103impl<T: Serialize, E: Endian, W: Write> ChannelSend<T> for Sender<T, E, W> {
104    type Error = SendError;
105    fn send(&mut self, value: &T) -> Result<(), SendError> {
106        let buffer = self.config.serialize(value)?;
107        self.writer.write_u64::<E>(buffer.len() as u64)?;
108        self.writer.write(&buffer)?;
109        Ok(())
110    }
111}
tcp_channel/sender.rs

tcp_channel/
sender.rs
