wiring 0.1.0

An async binary serialization framework with channels support
Documentation 
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

use futures::FutureExt;
use std::{collections::HashMap, fmt::Debug};

use tokio::{
    io::{AsyncRead, AsyncWrite},
    net::{
        tcp::{OwnedReadHalf, OwnedWriteHalf},
        TcpStream,
    },
    sync::mpsc::UnboundedSender,
};
use unwire::Unwire;
use wire::Wire;
pub type WireId = u64;
use wire::{WireStream, Wiring};

pub(crate) mod listener;
pub(crate) mod unwire;
pub(crate) mod wire;

use listener::Local;

use self::listener::ConnectInfo;

pub trait SplitStream: Sized + Send + Sync + Unwire + Wire + Debug {
    type Unwire: Unwire;
    type Wire: Wire;
    fn split(self) -> Result<(Self::Unwire, Self::Wire), std::io::Error> {
        Err(std::io::Error::new(
            std::io::ErrorKind::Unsupported,
            "Wire doesn't support split",
        ))
    }
}

impl SplitStream for tokio::net::TcpStream {
    type Unwire = OwnedReadHalf;
    type Wire = OwnedWriteHalf;
    fn split(self) -> Result<(Self::Unwire, Self::Wire), std::io::Error> {
        Ok(self.into_split())
    }
}

impl<T, C> SplitStream for WireStream<T, C>
where
    T: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static + Sync,
    C: ConnectConfig,
{
    type Unwire = WireStream<tokio::io::ReadHalf<T>, C>;
    type Wire = WireStream<tokio::io::WriteHalf<T>, C>;

    fn split(self) -> Result<(Self::Unwire, Self::Wire), std::io::Error> {
        // to make this more generic, we're using io::split.
        let (r, w) = tokio::io::split(self.stream);

        let mut r = WireStream::<_, C>::new(r);
        if let Some(Local { wire_info, incoming }) = self.local {
            let local = Local::new(wire_info, incoming);
            r = r.with_local(local);
        };
        let mut w = WireStream::new(w);
        w.peer = self.peer;
        Ok((r, w))
    }
}

pub trait ConnectConfig: Clone + Send + Sync + 'static + Debug {
    type Stream: AsyncRead + AsyncWrite + Send + Sync + Unpin + Debug;
    fn connect_stream(
        &self,
        connect_info: &ConnectInfo,
    ) -> impl std::future::Future<Output = Result<Self::Stream, std::io::Error>> + Send;
}

#[derive(Debug, Clone)]
pub struct TcpStreamConfig;

impl ConnectConfig for TcpStreamConfig {
    type Stream = TcpStream;
    async fn connect_stream(&self, connect_info: &ConnectInfo) -> Result<Self::Stream, std::io::Error> {
        match connect_info {
            ConnectInfo::TcpStream(addr) => {
                let stream = TcpStream::connect(addr).await?;
                Ok(stream)
            }
            _ => Err(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "Provided peer connect info unsupported by tcpstreamconfig",
            )),
        }
    }
}