actrpc-orchestrator 0.1.0

Configuration and orchestration utilities for ActRPC.
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

```rust
use actrpc_core::{
    convert::INTERCEPT_METHOD,
    interception::{InterceptionPhase, InterceptionRequest},
    json_rpc::JsonRpcMessage,
    participant::{Participant, ParticipantType},
};
use actrpc_transport::TransportTarget;

impl<I, A, P> DefaultOrchestrator<I, A, P>
where
    I: crate::InterceptorRegistry,
    A: crate::ActionRegistry,
    P: actrpc_transport::JsonRpcClientProvider,
{
    pub fn handle(
        &self,
        mut message: JsonRpcMessage,
        destination: TransportTarget,
    ) -> Result<JsonRpcMessage, OrchestratorError> {
        let mut prior_actions = Vec::new();

        // outbound
        for interceptor in self
            .interceptor_registry
            .interceptors_for_phase(InterceptionPhase::Outbound)?
        {
            let request = InterceptionRequest {
                origin: Participant {
                    kind: ParticipantType::Orchestrator,
                    id: "orchestrator".to_string(),
                },
                message: message.clone(),
                prior_actions: prior_actions.clone(),
            };

            let client = self.client_provider.get_client(&interceptor.target)?;
            let rpc_request = encode_interception_request(request)?;
            let rpc_response = client.send(rpc_request)?;
            let response = decode_interception_response(rpc_response)?;

            for requested_action in response.actions {
                let executor = self
                    .action_registry
                    .get_executor(requested_action.kind.as_str())
                    .ok_or_else(|| OrchestratorError::UnknownAction {
                        kind: requested_action.kind.clone(),
                    })?;

                let resolved = executor.execute(&request, requested_action)?;
                prior_actions.push(resolved);
            }

            if response.should_stop() {
                break;
            }
        }

        // downstream forward
        {
            let client = self.client_provider.get_client(&destination)?;
            message = client.send(message)?;
        }

        // inbound
        for interceptor in self
            .interceptor_registry
            .interceptors_for_phase(InterceptionPhase::Inbound)?
        {
            let request = InterceptionRequest {
                origin: Participant {
                    kind: ParticipantType::Orchestrator,
                    id: "orchestrator".to_string(),
                },
                message: message.clone(),
                prior_actions: prior_actions.clone(),
            };

            let client = self.client_provider.get_client(&interceptor.target)?;
            let rpc_request = encode_interception_request(request)?;
            let rpc_response = client.send(rpc_request)?;
            let response = decode_interception_response(rpc_response)?;

            for requested_action in response.actions {
                let executor = self
                    .action_registry
                    .get_executor(requested_action.kind.as_str())
                    .ok_or_else(|| OrchestratorError::UnknownAction {
                        kind: requested_action.kind.clone(),
                    })?;

                let resolved = executor.execute(&request, requested_action)?;
                prior_actions.push(resolved);
            }

            if response.should_stop() {
                break;
            }
        }

        Ok(message)
    }
}
```