dope-manifold 0.35.0

Owned application and service composition for dope
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
106
107
108
109
110
111
112
113
114
115

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

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

use crate::app::receive::ingress::{
    self, Admission as _, Dispatch, Replay, Strategy, dispatch::flow,
};

pub(crate) struct Scope;

impl<W, M> Strategy<W, M> for Scope
where
    W: wire::Spec,
    M: Dispatch<W, Strategy = Scope>,
    M::Kind: delivery::ScopedData + input::Replay<W>,
{
    type Value<'input, 'w, 'd>
        = Chunk<'w, 'd, W, Disabled>
    where
        'd: 'w,
        'w: 'input,
        W: 'w + 'd;

    type Pending<'w, 'd>
        = Chunk<'w, 'd, W, Disabled>
    where
        'd: 'w,
        W: 'w + 'd;

    type Admission<'a, 'c, 'd>
        = ingress::Single<'a, 'c, 'd>
    where
        'c: 'a,
        'd: 'a + 'c,
        W: 'd;

    fn drive_data<'p, 'a, 'c, 'w, 'd, ID: Identity + 'd, P, A>(
        policy: pin::Pin<&mut P>,
        data: A,
        work: Self::Admission<'a, 'c, 'd>,
    ) where
        'c: 'a,
        'd: 'a + 'c + 'w + 'p,
        W: 'w + 'd,
        P: ingress::Policy<'p, 'w, 'd, ID, Wire = W, Input = M>,
        M::Reserve: Choice<W, M::Ingress>,
        A: arrival::Arrival<'d, input::Receipt<W, M>, Observed = event::Data<'d>>,
    {
        work.drive(
            policy,
            data,
            |prep| prep.scoped(),
            |mut policy, next, turn, driver| 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);
                    let done = P::receive(policy.as_mut(), key, chunk, turn.reborrow(), driver);
                    let _ = flow::settle(policy, key, done, turn, driver);
                }
            },
        )
    }

    fn resume_tail<'p, 'w, 'd, ID: Identity + 'd, P>(
        mut policy: pin::Pin<&mut P>,
        item: &delivery::Resumed<'d, ID>,
        first: bool,
        turn: schedule::Turn<'_, 'd>,
        driver: &mut driver::Context<'_, 'd>,
    ) -> ControlFlow<(), bool>
    where
        'd: 'w + 'p,
        W: 'w + 'd,
        P: ingress::Policy<'p, 'w, 'd, ID, Wire = W, Input = M>,
        M::Reserve: Choice<W, M::Ingress>,
    {
        let (key, chunk, work) = {
            let mut recv = P::storage(policy.as_mut()).recv();
            let Some(taken) = recv.take_tail(item)? else {
                return ControlFlow::Continue(false);
            };
            let replay = if first {
                Replay::First
            } else {
                Replay::Waiting
            };
            let work = match <Self as Strategy<W, M>>::Admission::replay(
                replay,
                item.ready(),
                turn.reborrow(),
                driver,
            ) {
                schedule::Capacity::Available(work) => work,
                schedule::Capacity::Exhausted => return ControlFlow::Break(()),
            };
            (item.key(), taken.commit(), work)
        };
        let (driver, step) = work.into();
        ingress::prepare_terminal(policy.as_mut(), key);
        let done = P::receive(policy.as_mut(), key, chunk, step.reborrow(), driver);
        flow::settle(policy.as_mut(), key, done, step.reborrow(), driver)?;
        P::storage(policy).recv().finish_tail(item);
        ControlFlow::Continue(true)
    }
}