atap 0.1.0

Threadsafe futureless async runtime for macOS
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

//! # Exchange
//! Connecting, sending one request and reading the answer to the
//! end, for any family whose connect hands back a byte stream
//!
//! TCP and TLS both make requests this way. The only difference
//! is how they connect, which is the connect task they step

use crate::modules::input::token;
use crate::{
    RuntimeError,
    futures::{
        net::stream::{RecvTask, SendTask},
        task::{Task, sealed::Step},
    },
};
use std::sync::Arc;

/// Where a request is
#[derive(Default)]
pub(crate) enum Stage {
    /// Opening the connection, and whatever the family does to it
    /// before it can carry the request
    #[default]
    Connecting,

    /// Sending the request down it
    Sending(SendTask),

    /// Reading the answer
    Reading(RecvTask),
}

/// A connection a request can be sent down
pub(crate) trait Sends {
    /// Sends every byte of `data`
    fn send_all(&self, data: Arc<[u8]>) -> SendTask;
}

/// Takes a request as far as it can go without waiting
pub(crate) fn advance<C, S>(
    connect: &mut C,
    stage: &mut Stage,
    data: &Arc<[u8]>,
    reactor_id: i32,
    task_id: usize,
) -> Step<Result<Vec<u8>, RuntimeError>>
where
    C: Task<Output = Result<S, RuntimeError>>,
    S: Sends,
{
    loop {
        match &mut *stage {
            Stage::Connecting => match connect.step(token(), reactor_id, task_id) {
                Step::Done(Ok(conn)) => {
                    *stage = Stage::Sending(conn.send_all(Arc::clone(data)));
                }

                Step::Done(Err(error)) => return Step::Done(Err(error)),
                Step::Park(park) => return Step::Park(park),
            },

            Stage::Sending(send) => match send.step(token(), reactor_id, task_id) {
                Step::Done(Ok(_)) => {
                    let read = RecvTask::to_end(send.source().clone());

                    *stage = Stage::Reading(read);
                }

                Step::Done(Err(error)) => return Step::Done(Err(error)),
                Step::Park(park) => return Step::Park(park),
            },

            Stage::Reading(read) => return read.step(token(), reactor_id, task_id),
        }
    }
}