dope-manifold 0.37.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
78
79
80
81

use std::{convert, pin};

use dope_core::driver::{receive::Stale, route::Identity};
use dope_net::{
    link::{
        event,
        pool::recv::{arrival, delivery, input, reserve::Choice},
    },
    wire::{self, receive},
};
use receive::batch::Items;

use crate::app::receive::{
    Batch,
    ingress::{self, Dispatch, Strategy, batch::Grant, dispatch::flow},
};

pub(crate) struct Batched;

impl<W, M> Strategy<W, M> for Batched
where
    W: wire::Spec,
    M: Batch + Dispatch<W, Strategy = Batched>,
    M::Kind: delivery::BorrowedData,
{
    type Value<'input, 'w, 'd>
        = Items<'input, W>
    where
        'd: 'w,
        'w: 'input,
        W: 'w + 'd;

    type Pending<'w, 'd>
        = convert::Infallible
    where
        'd: 'w,
        W: 'w + 'd;

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

    fn drive_data<'p, 'a, 'c, 'w, 'd, ID: Identity + 'd, P, A>(
        mut 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.with(|driver, turn, bound| {
            let mut backing = None;
            let control = 'admit: {
                let prep = match P::reserve(policy.as_mut(), data, driver.region_token()) {
                    event::Reservation::Ready(prepared) => prepared,
                    event::Reservation::Control(next) => break 'admit Ok(next),
                    event::Reservation::Parked(parked) => break 'admit Err(parked),
                    event::Reservation::Stale(Stale) | event::Reservation::Drop => {
                        return ((), bound.empty());
                    }
                };
                let (next, unused) = prep.dispatch(&mut backing, bound);
                flow::apply_done(policy.as_mut(), next, turn, driver);
                return ((), unused);
            };
            let unused = bound.empty();
            match control {
                Ok(next) => ingress::apply_control(policy.as_mut(), next, turn, driver),
                Err(parked) => ingress::apply_parked(policy.as_mut(), parked, turn, driver),
            }
            ((), unused)
        })
    }
}