dope-manifold 0.43.0

Owned application and service composition 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

use std::{convert, ops::ControlFlow, pin};

use dope_core::driver::{self, route::Identity, schedule};
use dope_net::{
    link::{
        event,
        pool::{
            self,
            recv::{Deferral, input},
        },
    },
    wire::receive::{Chunk, mode::Disabled},
};

use crate::app::receive::ingress::{self, Policy, Progress};

pub(in crate::app::receive::ingress) fn settle<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P>(
    mut policy: pin::Pin<&mut P>,
    key: pool::Key<'d, ID>,
    done: Progress<Chunk<'w, 'd, P::Wire, Disabled>>,
    turn: schedule::Turn<'_, 'd>,
    driver: &mut driver::Context<'_, 'd>,
) -> ControlFlow<()>
where
    P: Policy<'p, 'w, 'd, ID>,
    <P::Input as input::Mode<P::Wire>>::Kind: input::Replay<P::Wire>,
{
    match done {
        Progress::Pending(pending) => {
            let deferred = P::storage(policy.as_mut()).recv().defer(key, pending);
            let (pending, cause) = match deferred {
                Deferral::Ready => {
                    P::finish(policy, key, ingress::Finish::Chunk, turn, driver);
                    return ControlFlow::Break(());
                }
                Deferral::Unavailable(pending) => (pending, event::ParkError::Unavailable),
                Deferral::Occupied(pending) => (pending, event::ParkError::Occupied),
            };
            ingress::discard_pending(policy, key, pending, cause, turn, driver);
        }
        Progress::Done(done) => ingress::finish_chunk(policy, key, done, turn, driver),
    }
    ControlFlow::Continue(())
}

pub(in crate::app::receive::ingress) fn apply_done<
    'input,
    'p,
    'w: 'input,
    'd: 'w + 'p,
    ID: Identity + 'd,
    P,
>(
    mut policy: pin::Pin<&mut P>,
    next: event::DispatchRecv<'d, ID, crate::Incoming<'input, 'w, 'd, P::Wire, P::Input>>,
    turn: schedule::Turn<'_, 'd>,
    driver: &mut driver::Context<'_, 'd>,
) where
    P: Policy<'p, 'w, 'd, ID>,
    crate::Ingress<P::Wire, P::Input>:
        ingress::Strategy<P::Wire, P::Input, Pending<'w, 'd> = convert::Infallible>,
{
    match next {
        event::DispatchRecv::Control(next) => {
            ingress::apply_control(policy, next, turn, driver);
        }
        event::DispatchRecv::Chunk(key, chunk) => {
            ingress::prepare_terminal(policy.as_mut(), key);
            match P::receive(policy.as_mut(), key, chunk, turn.reborrow(), driver) {
                Progress::Pending(pending) => match pending {},
                Progress::Done(done) => {
                    ingress::finish_chunk(policy, key, done, turn.reborrow(), driver);
                }
            }
        }
    }
}