liminal-rs 0.2.0

A conversation-based messaging bus built on beamr
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
121
122
123

use uuid::Uuid;

use super::super::channels::ChannelName;
use super::super::codec::DispatchRequest;
use super::super::error::AionSurfaceError;
use super::super::types::ActivityRequest;
use super::{
    ConversationIdProvider, DispatchConversation, DispatchConversationEvent,
    DispatchConversationFactory, DispatchRecorder, DispatchWorker, DispatchWorkerPool,
    RecordedDispatchOutcome, dispatch_failed,
};
#[derive(Debug)]
pub(super) struct EmptyWorkerPool;

impl DispatchWorkerPool for EmptyWorkerPool {
    fn workers_for(
        &self,
        channel_name: &ChannelName,
        request: &ActivityRequest,
    ) -> Result<Vec<DispatchWorker>, AionSurfaceError> {
        let _ = (channel_name, request);
        Ok(Vec::new())
    }
}

#[derive(Debug)]
pub(super) struct NoopRouter;

impl super::DispatchRouter for NoopRouter {
    fn select_worker(
        &self,
        workflow_id: &str,
        channel_name: &ChannelName,
        request: &ActivityRequest,
        candidates: &[DispatchWorker],
        excluded_worker_ids: &[String],
    ) -> Result<Option<DispatchWorker>, AionSurfaceError> {
        let _ = (
            workflow_id,
            channel_name,
            request,
            candidates,
            excluded_worker_ids,
        );
        Ok(None)
    }
}

#[derive(Debug)]
pub(super) struct NoopConversationFactory;

impl DispatchConversationFactory for NoopConversationFactory {
    fn open(
        &self,
        workflow_id: &str,
        channel_name: &ChannelName,
        conversation_id: &str,
    ) -> Result<Box<dyn DispatchConversation>, AionSurfaceError> {
        let _ = conversation_id;
        Ok(Box::new(NoopConversation {
            workflow_id: workflow_id.to_owned(),
            channel_name: channel_name.clone(),
        }))
    }
}

#[derive(Debug)]
struct NoopConversation {
    workflow_id: String,
    channel_name: ChannelName,
}

impl DispatchConversation for NoopConversation {
    fn link_worker(&mut self, worker: &DispatchWorker) -> Result<(), AionSurfaceError> {
        let _ = worker;
        Ok(())
    }

    fn send(&mut self, request: DispatchRequest) -> Result<(), AionSurfaceError> {
        let _ = request;
        Ok(())
    }

    fn receive(&mut self) -> Result<DispatchConversationEvent, AionSurfaceError> {
        Err(dispatch_failed(
            &self.channel_name,
            self.workflow_id.as_str(),
            "NoWorkersAvailable: no response can be received without a worker",
        ))
    }

    fn close(&mut self) -> Result<(), AionSurfaceError> {
        Ok(())
    }
}

#[derive(Debug)]
pub(super) struct NoopRecorder;

impl DispatchRecorder for NoopRecorder {
    fn replay_outcome(
        &self,
        channel_name: &str,
        request: &ActivityRequest,
    ) -> Result<Option<RecordedDispatchOutcome>, AionSurfaceError> {
        let _ = (channel_name, request);
        Ok(None)
    }

    fn record(&self, operation: super::DispatchOperation) -> Result<(), AionSurfaceError> {
        let _ = operation;
        Ok(())
    }
}

#[derive(Debug)]
pub(super) struct UuidConversationIds;

impl ConversationIdProvider for UuidConversationIds {
    fn next_conversation_id(&self) -> String {
        Uuid::new_v4().to_string()
    }
}