dope-net 0.39.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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

use dope_core::driver::{
    receive::credit::{self, error::Arm},
    route::Identity,
};

use crate::{
    link::slot::{
        self,
        recv::{Blocked, Prepared},
        types,
    },
    wire::{
        self,
        receive::{self, mode, reserve},
        terminal,
    },
};

pub(in crate::link) enum Reserve<
    'a,
    'w,
    'd,
    ID: Identity + 'w,
    W: wire::Spec + 'w,
    P: mode::Flag,
    R: reserve::Try<W>,
> where
    'd: 'w,
{
    Blocked(Blocked<'a, 'w, 'd, ID, W, R>),
    Credit(Arm),
    Control(slot::Control),
    Ready(Prepared<'a, 'w, 'd, ID, W, P>),
}

impl<'a, 'w, 'd, ID: Identity + 'w, W, R> Reserve<'a, 'w, 'd, ID, W, mode::Waiting<R>, R>
where
    'd: 'w,
    W: wire::Spec + 'w,
    R: reserve::Try<W>,
    W::Receive: receive::Policy<W, Reserve = reserve::Waiting<R>>,
{
    pub(in crate::link) fn try_new(
        conn: &'a mut types::Control<'w, 'd, ID, W>,
        rt: &'a mut W::Runtime<'w, 'd, ID>,
        credit: credit::Slot<'a, 'd>,
        mut arm: bool,
    ) -> Self {
        let key = &conn.key;
        let end = match conn.engine.both(&mut conn.terminal) {
            terminal::Admission::Open(end) => end,
            terminal::Admission::Closed => {
                return Reserve::Control(slot::Control::Discarded {
                    needs_rearm: arm,
                    finished: false,
                });
            }
        };
        let send = if conn.engine.operations.sending.is_inflight() {
            None
        } else {
            Some(&mut conn.send)
        };
        let txn = match R::try_reserve(&mut conn.wire, send, rt) {
            reserve::Outcome::Ready(txn) => txn,
            reserve::Outcome::Blocked(block) => {
                let credit = match credit
                    .arm(key.target().dispatch(), conn.engine.ready_handle().target())
                {
                    Ok(credit) => credit,
                    Err(cause) => {
                        return Reserve::Credit(cause);
                    }
                };
                conn.engine.operations.recv.pause();
                return Reserve::Blocked(Blocked {
                    block,
                    credit,
                    key,
                    needs_cancel: conn.engine.operations.recv.has_cancel(),
                });
            }
            reserve::Outcome::Failed(cause) => {
                let ((), captured) = end
                    .terminal((), terminal::Terminal::Failed(cause))
                    .into_parts();
                Prepared::<ID, W, mode::Waiting<R>>::capture(&mut conn.engine, &mut arm, captured);
                return Reserve::Control(slot::Control::NoChunk { needs_rearm: arm });
            }
        };
        Reserve::Ready(Prepared {
            engine: &mut conn.engine,
            end,
            transaction: txn,
            credit,
            needs_rearm: arm,
        })
    }
}