dope-net 0.46.0

Owned network transports for dope
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

use dope_core::{
    driver::{self, retained::flight, route, route::Identity},
    io::fd::handles,
};

use crate::{
    link::{
        duplex::sending,
        slot::{transmit::outcome, types},
    },
    wire::{self, send, terminal},
};
pub(super) fn submit_pending<'a, 'w, 'd, ID: Identity + 'w, W: wire::Spec + 'w>(
    context: send::Context<'a, 'w, ID, W, send::Storage<'a, W::SendStorage<'w>>>,
    submit: sending::Submission<'_, 'd, <W::Send as send::Policy<W>>::Reclaim>,
    fd: &handles::Descriptor<'d>,
    slots: &flight::Slots<'d, ID>,
    driver: &mut driver::Context<'_, 'd>,
    target: route::Target<'d, ID>,
    end: terminal::Admission<terminal::Send<'_, terminal::Error<'w, W>>>,
) -> (terminal::Captured, outcome::Handoff)
where
    'd: 'w,
{
    let terminal::Admission::Open(end) = end else {
        return (terminal::Captured::None, outcome::Handoff::blocked());
    };
    let sending::Pending::Ready(seal) = submit.pending() else {
        return (terminal::Captured::None, outcome::Handoff::blocked());
    };
    let done = <W::Send as send::Policy<W>>::close::<ID>(context, end);
    let (ready, captured) = done.into_parts();
    let ready = ready.resolve(captured);
    let submit = seal.seal(ready.is_waiting());
    let submit = submit.submit(fd, slots, driver, target, ready);
    let handoff = outcome::Handoff::from_attempt(submit);
    (captured, handoff)
}

pub(super) fn flush<'w, 'd, ID: Identity + 'w, W>(
    slot: &mut types::Control<'w, 'd, ID, W>,
    slots: &flight::Slots<'d, ID>,
    driver: &mut driver::Context<'_, 'd>,
    target: route::Target<'d, ID>,
    source: sending::Source,
) -> outcome::Progress
where
    'd: 'w,
    W: wire::Spec + 'w,
{
    if slot.engine.operations.lifecycle.close.is_discarded() {
        return outcome::Progress::Idle;
    }
    let end = slot.engine.send(&mut slot.terminal);
    let Some(fd) = slot.engine.establish.fd() else {
        return outcome::Progress::Idle;
    };
    let Some(op) = slot.engine.operations.submission(source).try_ready() else {
        return outcome::Progress::Idle;
    };
    let context = send::Context::new(&mut slot.wire, send::Storage::new(&mut slot.send));
    let (attempt, captured) = match end {
        terminal::Admission::Open(end) => {
            let done = <W::Send as send::Policy<W>>::flush::<ID>(context, end);
            let (ready, captured) = done.into_parts();
            let ready = ready.resolve(captured);
            let attempt = op.submit(fd, slots, driver, target, ready);
            (attempt, captured)
        }
        terminal::Admission::Closed => {
            let end = slot.terminal.drain();
            let ready = <W::Send as send::Policy<W>>::drain::<ID>(context, end).close_after();
            let captured = terminal::Captured::None;
            let attempt = op.submit(fd, slots, driver, target, ready);
            (attempt, captured)
        }
    };
    let result = outcome::Handoff::from_attempt(attempt);
    slot.engine.captured(captured);
    result.progress()
}