bombay-rs 0.1.0

A pluggable process runtime over Bombay Behavior.
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
116
117
118
119
120

//! Child and peer observation interpretation for one actor incarnation.

use core::convert::Infallible;

use behavior::{
    Address, ChildEvent, ChildStopped, ObserveChild, ObservePeer, PeerEvent, PeerStopped,
    ServiceSends, UnwatchPeer,
};
use tokio::time::Instant;

use super::{Completion, IncarnationEffects, IntoPeerOutcome, classify_task};
use crate::{ChildLease, EventSender, PeerObservationError, PeerObserver, RouteSends};

impl<A, R, Edge, T, S, P>
    RouteSends<A, IncarnationEffects<R, A::Nonce, ChildLease<Edge, T>, S, P, A>>
    for ServiceSends<ObserveChild<A::Nonce>>
where
    A: Address + Send + Sync + 'static,
    A::Nonce: Send + 'static,
    R: Send,
    Edge: Send + 'static,
    T: IntoPeerOutcome<A> + Send + Sync + 'static,
    S: EventSender + Clone + Send + Sync + 'static,
    S::Event: ChildEvent<Addr = A> + Send + 'static,
    P: Send,
{
    type Error = super::child_scope::ChildObservationError<A::Nonce>;

    async fn route(
        self,
        _from: A,
        services: &mut IncarnationEffects<R, A::Nonce, ChildLease<Edge, T>, S, P, A>,
    ) -> Result<(), Self::Error> {
        for request in self {
            if !services.children.contains(request.nonce)
                && services.creation_was_rejected(request.nonce)
            {
                // The paired creation was rejected; the recoverable
                // result travels through the creation lane instead.
                continue;
            }
            let child = services.children.get_mut(request.nonce)?;
            let completion: Completion<T> = child
                .take_completion()
                .expect("a selectable child must own its completion seat");
            let response = services.response.clone();
            services.monitor(async move {
                let completion = completion.wait().await;
                let outcome = classify_task::<A, _>(&completion);
                if let Some(event) = <S::Event as ChildEvent>::child_stopped(ChildStopped {
                    nonce: request.nonce,
                    outcome,
                    at: Instant::now(),
                }) {
                    let _ = response.send(event).await;
                }
            });
        }
        Ok(())
    }
}

impl<A, R, N, L, S, P> RouteSends<A, IncarnationEffects<R, N, L, S, P, A>>
    for ServiceSends<ObservePeer<A>>
where
    A: Address + Send + Sync + 'static,
    R: PeerObserver<A> + Send,
    N: Send,
    L: Send,
    S: EventSender + Clone + Send + Sync + 'static,
    S::Event: PeerEvent<Addr = A> + Send + 'static,
    P: Send,
{
    type Error = PeerObservationError<A>;

    async fn route(
        self,
        _from: A,
        services: &mut IncarnationEffects<R, N, L, S, P, A>,
    ) -> Result<(), Self::Error> {
        for request in self {
            let completion = services.router.observe_peer(request.peer)?;
            let response = services.response.clone();
            services.monitor_peer(request.peer, async move {
                let outcome = completion.await;
                if let Some(event) = <S::Event as PeerEvent>::peer_stopped(PeerStopped {
                    peer: request.peer,
                    outcome,
                }) {
                    let _ = response.send(event).await;
                }
            });
        }
        Ok(())
    }
}

impl<A, R, N, L, S, P> RouteSends<A, IncarnationEffects<R, N, L, S, P, A>>
    for ServiceSends<UnwatchPeer<A>>
where
    A: Address + Send,
    R: Send,
    N: Send,
    L: Send,
    S: Send,
    P: Send,
{
    type Error = Infallible;

    async fn route(
        self,
        _from: A,
        services: &mut IncarnationEffects<R, N, L, S, P, A>,
    ) -> Result<(), Self::Error> {
        for request in self {
            services.unwatch_peer(&request.peer);
        }
        Ok(())
    }
}