interprocess 2.4.2

Interprocess communication toolkit
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
99
100
101
102
103
104
105

use {
    super::drive_server,
    crate::{
        os::windows::named_pipe::{
            pipe_mode,
            tokio::{DuplexPipeStream, PipeListener, RecvPipeStream, SendPipeStream},
        },
        tests::util::{message, TestResult, WrapErrExt},
    },
    std::sync::Arc,
    tokio::{
        io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
        sync::oneshot::Sender,
        try_join,
    },
};

fn msg(server: bool) -> Box<str> { message(None, server, Some('\n')) }

pub async fn server_duplex(
    id: &str,
    name_sender: Sender<Arc<str>>,
    num_clients: u32,
) -> TestResult {
    drive_server(
        id,
        name_sender,
        num_clients,
        |plo| plo.create_tokio_duplex::<pipe_mode::Bytes>(),
        handle_conn_duplex,
    )
    .await
}
pub async fn server_cts(id: &str, name_sender: Sender<Arc<str>>, num_clients: u32) -> TestResult {
    drive_server(
        id,
        name_sender,
        num_clients,
        |plo| plo.create_tokio_recv_only::<pipe_mode::Bytes>(),
        handle_conn_cts,
    )
    .await
}
pub async fn server_stc(id: &str, name_sender: Sender<Arc<str>>, num_clients: u32) -> TestResult {
    drive_server(
        id,
        name_sender,
        num_clients,
        |plo| plo.create_tokio_send_only::<pipe_mode::Bytes>(),
        handle_conn_stc,
    )
    .await
}

async fn handle_conn_duplex(
    listener: Arc<PipeListener<pipe_mode::Bytes, pipe_mode::Bytes>>,
) -> TestResult {
    let (mut recver, mut sender) = listener.accept().await.opname("accept")?.split();
    try_join!(recv(&mut recver, msg(false)), send(&mut sender, msg(true)))?;
    DuplexPipeStream::reunite(recver, sender).opname("reunite")?;
    Ok(())
}
async fn handle_conn_cts(
    listener: Arc<PipeListener<pipe_mode::Bytes, pipe_mode::None>>,
) -> TestResult {
    let mut recver = listener.accept().await.opname("accept")?;
    recv(&mut recver, msg(false)).await
}
async fn handle_conn_stc(
    listener: Arc<PipeListener<pipe_mode::None, pipe_mode::Bytes>>,
) -> TestResult {
    let mut sender = listener.accept().await.opname("accept")?;
    send(&mut sender, msg(true)).await
}

pub async fn client_duplex(name: Arc<str>) -> TestResult {
    let (mut recver, mut sender) = DuplexPipeStream::<pipe_mode::Bytes>::connect_by_path(&*name)
        .await
        .opname("connect")?
        .split();
    try_join!(recv(&mut recver, msg(true)), send(&mut sender, msg(false)))?;
    DuplexPipeStream::reunite(recver, sender).opname("reunite")?;
    Ok(())
}
pub async fn client_cts(name: Arc<str>) -> TestResult {
    let mut sender =
        SendPipeStream::<pipe_mode::Bytes>::connect_by_path(&*name).await.opname("connect")?;
    send(&mut sender, msg(false)).await
}
pub async fn client_stc(name: Arc<str>) -> TestResult {
    let mut recver =
        RecvPipeStream::<pipe_mode::Bytes>::connect_by_path(&*name).await.opname("connect")?;
    recv(&mut recver, msg(true)).await
}

async fn recv(recver: &mut RecvPipeStream<pipe_mode::Bytes>, exp: impl AsRef<str>) -> TestResult {
    let mut buffer = String::with_capacity(128);
    let mut recver = BufReader::new(recver);
    recver.read_line(&mut buffer).await.opname("receive")?;
    ensure_eq!(buffer, exp.as_ref());
    Ok(())
}
async fn send(sender: &mut SendPipeStream<pipe_mode::Bytes>, snd: impl AsRef<str>) -> TestResult {
    sender.write_all(snd.as_ref().as_bytes()).await.opname("send")
}