sosistab 0.5.43

An obfuscated datagram transport for horrible networks
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
93
94
95
96
97
98

use crate::{buffer::Buff, runtime, Session};
use smol::channel::{Receiver, Sender};
use std::sync::Arc;
mod congestion;
mod multiplex_actor;
pub mod pkt_trace;
mod relconn;
mod structs;
// pub use congestion::*;
pub use relconn::RelConn;

/// A multiplex session over a sosistab session, implementing both reliable "streams" and unreliable messages.
pub struct Multiplex {
    urel_send: Sender<Buff>,
    urel_recv: Receiver<Buff>,
    conn_open: Sender<(Option<String>, Sender<RelConn>)>,
    conn_accept: Receiver<RelConn>,
    send_session: Sender<Session>,
    _task: smol::Task<()>,
}

fn to_ioerror<T: Into<Box<dyn std::error::Error + Send + Sync>>>(val: T) -> std::io::Error {
    std::io::Error::new(std::io::ErrorKind::ConnectionReset, val)
}

impl Multiplex {
    /// Creates a new multiplexed session
    pub fn new(session: Session) -> Self {
        let (send_session, recv_session) = smol::channel::unbounded();
        let (urel_send, urel_send_recv) = smol::channel::bounded(256);
        let (urel_recv_send, urel_recv) = smol::channel::bounded(256);
        let (conn_open, conn_open_recv) = smol::channel::unbounded();
        let (conn_accept_send, conn_accept) = smol::channel::bounded(100);
        send_session.try_send(session).unwrap();
        let _task = runtime::spawn(async move {
            let retval = multiplex_actor::multiplex(
                recv_session,
                urel_send_recv,
                urel_recv_send,
                conn_open_recv,
                conn_accept_send,
            )
            .await;
            tracing::debug!("multiplex actor returned {:?}", retval);
        });
        Multiplex {
            urel_send,
            urel_recv,
            conn_open,
            conn_accept,
            send_session,
            _task,
        }
    }

    /// Sends an unreliable message to the other side
    pub async fn send_urel(&self, msg: impl Into<Buff>) -> std::io::Result<()> {
        self.urel_send.send(msg.into()).await.map_err(to_ioerror)
    }

    pub async fn recv_urel(&self) -> std::io::Result<Buff> {
        self.urel_recv.recv().await.map_err(to_ioerror)
    }

    /// Receive an unreliable message if there is one available.
    pub fn try_recv_urel(&self) -> std::io::Result<Buff> {
        self.urel_recv.try_recv().map_err(to_ioerror)
    }

    // /// Gets a reference to the underlying Session
    // pub async fn get_session(&self) -> impl '_ + Deref<Target = Session> {
    //     self.sess_ref.read().clone()
    // }

    /// Replaces the internal Session. This drops the previous Session, but this is not guaranteed to happen immediately.
    pub async fn replace_session(&self, sess: Session) {
        let _ = self.send_session.try_send(sess);
    }

    /// Open a reliable conn to the other end.
    pub async fn open_conn(&self, additional: Option<String>) -> std::io::Result<RelConn> {
        let (send, recv) = smol::channel::unbounded();
        self.conn_open
            .send((additional.clone(), send))
            .await
            .map_err(to_ioerror)?;
        if let Ok(s) = recv.recv().await {
            Ok(s)
        } else {
            smol::future::pending().await
        }
    }

    /// Accept a reliable conn from the other end.
    pub async fn accept_conn(&self) -> std::io::Result<RelConn> {
        self.conn_accept.recv().await.map_err(to_ioerror)
    }
}