lenso_runtime_conformance/
interaction.rs

1use std::{
2    any::Any,
3    cell::{Cell, RefCell},
4    collections::VecDeque,
5    rc::Rc,
6};
7
8use futures::future::{LocalBoxFuture, ready};
9use lenso_app_plan::PluginInstancePlan;
10use lenso_kernel::{
11    EventCapability, InvocationContext, NativeEventEndpoint, NativeStreamEndpoint,
12    NativeStreamHandle, NativeStreamItem, NativeStreamSession, NoopPluginLifecycle, RuntimeFailure,
13    StreamCapability,
14};
15
16use super::{ConformancePlugin, ConformancePluginFactory};
17
18/// Stable identity for the bidirectional stream conformance Capability.
19pub const STREAM_PROBE_CAPABILITY_ID: &str = "lenso.runtime.conformance.stream-probe@1";
20/// Exact stream conformance Descriptor version.
21pub const STREAM_PROBE_DESCRIPTOR_VERSION: &str = "1.0.0";
22/// Stream Operation exercised by Runtime Driver and Execution Adapter tests.
23pub const STREAM_PROBE_OPERATION: &str = "exchange";
24/// Provider package used by the stream conformance suite.
25pub const STREAM_PROBE_PROVIDER_PACKAGE_ID: &str =
26    "lenso.runtime.conformance.stream-probe-provider";
27/// Consumer package used by the stream conformance suite.
28pub const STREAM_PROBE_CONSUMER_PACKAGE_ID: &str =
29    "lenso.runtime.conformance.stream-probe-consumer";
30
31/// Request used to open one conformance stream.
32#[derive(Clone, Debug, Eq, PartialEq)]
33pub struct StreamProbeOpen {
34    pub value: String,
35}
36
37/// Ordered value exchanged in both directions by the conformance stream.
38#[derive(Clone, Debug, Eq, PartialEq)]
39pub struct StreamProbeMessage {
40    pub sequence: u64,
41    pub value: String,
42}
43
44/// Capability-defined stream opening or terminal failure.
45#[derive(Clone, Debug, Eq, PartialEq)]
46pub enum StreamProbeError {
47    Rejected,
48}
49
50/// Typed bidirectional stream conformance Capability.
51#[derive(Debug)]
52pub struct StreamProbe;
53
54impl StreamCapability for StreamProbe {
55    type OpenRequest = StreamProbeOpen;
56    type Message = StreamProbeMessage;
57    type DomainError = StreamProbeError;
58
59    const ID: &'static str = STREAM_PROBE_CAPABILITY_ID;
60    const DESCRIPTOR_VERSION: &'static str = STREAM_PROBE_DESCRIPTOR_VERSION;
61}
62
63/// Consumer wrapper shared by Runtime Driver and Execution Adapter tests.
64#[derive(Debug)]
65pub struct StreamProbeClient {
66    handle: NativeStreamHandle<StreamProbe>,
67}
68
69impl StreamProbeClient {
70    pub fn new(handle: NativeStreamHandle<StreamProbe>) -> Self {
71        Self { handle }
72    }
73
74    pub fn from_dependencies(
75        dependencies: &lenso_kernel::PluginDependencies,
76    ) -> Result<Self, RuntimeFailure> {
77        Ok(Self::new(dependencies.one_stream::<StreamProbe>()?))
78    }
79
80    pub async fn open(
81        &self,
82        request: StreamProbeOpen,
83    ) -> Result<Result<lenso_kernel::NativeStream<StreamProbe>, StreamProbeError>, RuntimeFailure>
84    {
85        self.handle.open(STREAM_PROBE_OPERATION, request).await
86    }
87}
88
89/// In-memory endpoint that echoes sent values while preserving stream protocol semantics.
90#[derive(Clone, Debug, Default)]
91pub struct StreamProbeEndpoint;
92
93impl NativeStreamEndpoint for StreamProbeEndpoint {
94    fn capability_id(&self) -> &'static str {
95        STREAM_PROBE_CAPABILITY_ID
96    }
97
98    fn descriptor_version(&self) -> &'static str {
99        STREAM_PROBE_DESCRIPTOR_VERSION
100    }
101
102    fn operations(&self) -> &'static [&'static str] {
103        &[STREAM_PROBE_OPERATION]
104    }
105
106    fn open(
107        &self,
108        operation: &str,
109        request: Box<dyn Any>,
110        context: InvocationContext,
111    ) -> LocalBoxFuture<
112        'static,
113        Result<Result<Box<dyn NativeStreamSession>, Box<dyn Any>>, RuntimeFailure>,
114    > {
115        if operation != STREAM_PROBE_OPERATION {
116            return Box::pin(ready(Err(RuntimeFailure::UnknownOperation {
117                capability: STREAM_PROBE_CAPABILITY_ID,
118                operation: operation.to_owned(),
119            })));
120        }
121        let Ok(request) = request.downcast::<StreamProbeOpen>() else {
122            return Box::pin(ready(Err(RuntimeFailure::ProtocolViolation {
123                capability: STREAM_PROBE_CAPABILITY_ID,
124            })));
125        };
126        if request.value == "reject" {
127            return Box::pin(ready(Ok(Err(
128                Box::new(StreamProbeError::Rejected) as Box<dyn Any>
129            ))));
130        }
131        let session: Box<dyn NativeStreamSession> =
132            Box::new(EchoStreamSession::new(context.request_id(), request.value));
133        Box::pin(ready(Ok(Ok(session))))
134    }
135}
136
137#[derive(Debug)]
138struct EchoStreamState {
139    open_value: String,
140    pending: VecDeque<NativeStreamItem>,
141    send_closed: bool,
142    cancelled: bool,
143}
144
145#[derive(Debug)]
146struct EchoStreamSession {
147    request_id: u64,
148    state: Rc<RefCell<EchoStreamState>>,
149}
150
151impl EchoStreamSession {
152    fn new(request_id: u64, open_value: String) -> Self {
153        Self {
154            request_id,
155            state: Rc::new(RefCell::new(EchoStreamState {
156                open_value,
157                pending: VecDeque::new(),
158                send_closed: false,
159                cancelled: false,
160            })),
161        }
162    }
163}
164
165impl NativeStreamSession for EchoStreamSession {
166    fn send(&self, message: Box<dyn Any>) -> LocalBoxFuture<'static, Result<(), RuntimeFailure>> {
167        let Ok(message) = message.downcast::<StreamProbeMessage>() else {
168            return Box::pin(ready(Err(RuntimeFailure::ProtocolViolation {
169                capability: STREAM_PROBE_CAPABILITY_ID,
170            })));
171        };
172        let mut state = self.state.borrow_mut();
173        if state.cancelled {
174            return Box::pin(ready(Err(RuntimeFailure::Cancelled {
175                request_id: self.request_id,
176            })));
177        }
178        if state.send_closed {
179            return Box::pin(ready(Err(RuntimeFailure::ProtocolViolation {
180                capability: STREAM_PROBE_CAPABILITY_ID,
181            })));
182        }
183        let message = StreamProbeMessage {
184            sequence: message.sequence,
185            value: format!("{}: {}", state.open_value, message.value),
186        };
187        state
188            .pending
189            .push_back(NativeStreamItem::Message(Box::new(message)));
190        Box::pin(ready(Ok(())))
191    }
192
193    fn receive(&self) -> LocalBoxFuture<'static, Result<NativeStreamItem, RuntimeFailure>> {
194        let result =
195            self.state
196                .borrow_mut()
197                .pending
198                .pop_front()
199                .ok_or_else(|| RuntimeFailure::Internal {
200                    detail: "stream conformance fixture has no pending item".to_owned(),
201                });
202        Box::pin(ready(result))
203    }
204
205    fn close_send(&self) -> LocalBoxFuture<'static, Result<(), RuntimeFailure>> {
206        let mut state = self.state.borrow_mut();
207        if state.cancelled {
208            return Box::pin(ready(Err(RuntimeFailure::Cancelled {
209                request_id: self.request_id,
210            })));
211        }
212        if !state.send_closed {
213            state.send_closed = true;
214            state.pending.push_back(NativeStreamItem::PeerHalfClosed);
215            state.pending.push_back(NativeStreamItem::Terminal(Ok(())));
216        }
217        Box::pin(ready(Ok(())))
218    }
219
220    fn cancel(&self) {
221        let mut state = self.state.borrow_mut();
222        state.cancelled = true;
223        state.pending.clear();
224    }
225}
226
227/// Factory for the default stream conformance endpoint.
228#[derive(Debug)]
229pub struct StreamProbeProviderFactory;
230
231impl ConformancePluginFactory for StreamProbeProviderFactory {
232    fn package_id(&self) -> &'static str {
233        STREAM_PROBE_PROVIDER_PACKAGE_ID
234    }
235
236    fn package_version(&self) -> &'static str {
237        env!("CARGO_PKG_VERSION")
238    }
239
240    fn instantiate(
241        &self,
242        _instance: &PluginInstancePlan,
243    ) -> Result<ConformancePlugin, RuntimeFailure> {
244        Ok(ConformancePlugin::with_stream_endpoints(
245            vec![Rc::new(StreamProbeEndpoint)],
246            NoopPluginLifecycle,
247        ))
248    }
249}
250
251/// Stable identity for the ephemeral Event conformance Capability.
252pub const EVENT_PROBE_CAPABILITY_ID: &str = "lenso.runtime.conformance.event-probe@1";
253/// Exact Event conformance Descriptor version.
254pub const EVENT_PROBE_DESCRIPTOR_VERSION: &str = "1.0.0";
255/// Event Operation exercised by Runtime Driver and Execution Adapter tests.
256pub const EVENT_PROBE_OPERATION: &str = "publish";
257/// Provider package used by the Event conformance suite.
258pub const EVENT_PROBE_PROVIDER_PACKAGE_ID: &str = "lenso.runtime.conformance.event-probe-provider";
259/// Consumer package used by the Event conformance suite.
260pub const EVENT_PROBE_CONSUMER_PACKAGE_ID: &str = "lenso.runtime.conformance.event-probe-consumer";
261
262/// Ordered value published through the ephemeral Event conformance seam.
263#[derive(Clone, Debug, Eq, PartialEq)]
264pub struct EventProbeValue {
265    pub sequence: u64,
266    pub value: String,
267}
268
269/// Typed ephemeral Event conformance Capability.
270#[derive(Debug)]
271pub struct EventProbe;
272
273impl EventCapability for EventProbe {
274    type Event = EventProbeValue;
275
276    const ID: &'static str = EVENT_PROBE_CAPABILITY_ID;
277    const DESCRIPTOR_VERSION: &'static str = EVENT_PROBE_DESCRIPTOR_VERSION;
278}
279
280/// Shared observation handle for accepted Event values.
281#[derive(Clone, Debug, Default)]
282pub struct EventProbeRecorder {
283    seen: Rc<RefCell<Vec<EventProbeValue>>>,
284}
285
286impl EventProbeRecorder {
287    pub fn values(&self) -> Vec<EventProbeValue> {
288        self.seen.borrow().clone()
289    }
290}
291
292/// In-memory Event endpoint used by the conformance Adapter.
293#[derive(Clone, Debug)]
294pub struct EventProbeEndpoint {
295    recorder: EventProbeRecorder,
296    available: Rc<Cell<bool>>,
297}
298
299impl EventProbeEndpoint {
300    pub fn new(recorder: EventProbeRecorder) -> Self {
301        Self {
302            recorder,
303            available: Rc::new(Cell::new(true)),
304        }
305    }
306
307    pub fn set_available(&self, available: bool) {
308        self.available.set(available);
309    }
310}
311
312impl NativeEventEndpoint for EventProbeEndpoint {
313    fn capability_id(&self) -> &'static str {
314        EVENT_PROBE_CAPABILITY_ID
315    }
316
317    fn descriptor_version(&self) -> &'static str {
318        EVENT_PROBE_DESCRIPTOR_VERSION
319    }
320
321    fn operations(&self) -> &'static [&'static str] {
322        &[EVENT_PROBE_OPERATION]
323    }
324
325    fn publish(
326        &self,
327        operation: &str,
328        event: Box<dyn Any>,
329        _context: InvocationContext,
330    ) -> LocalBoxFuture<'static, Result<(), RuntimeFailure>> {
331        if operation != EVENT_PROBE_OPERATION {
332            return Box::pin(ready(Err(RuntimeFailure::UnknownOperation {
333                capability: EVENT_PROBE_CAPABILITY_ID,
334                operation: operation.to_owned(),
335            })));
336        }
337        if !self.available.get() {
338            return Box::pin(ready(Err(RuntimeFailure::Unavailable {
339                capability: EVENT_PROBE_CAPABILITY_ID,
340            })));
341        }
342        let Ok(event) = event.downcast::<EventProbeValue>() else {
343            return Box::pin(ready(Err(RuntimeFailure::ProtocolViolation {
344                capability: EVENT_PROBE_CAPABILITY_ID,
345            })));
346        };
347        self.recorder.seen.borrow_mut().push(*event);
348        Box::pin(ready(Ok(())))
349    }
350}
351
352/// Factory for one observable Event conformance endpoint.
353#[derive(Clone, Debug)]
354pub struct EventProbeProviderFactory {
355    recorder: EventProbeRecorder,
356}
357
358impl EventProbeProviderFactory {
359    pub fn new(recorder: EventProbeRecorder) -> Self {
360        Self { recorder }
361    }
362}
363
364impl ConformancePluginFactory for EventProbeProviderFactory {
365    fn package_id(&self) -> &'static str {
366        EVENT_PROBE_PROVIDER_PACKAGE_ID
367    }
368
369    fn package_version(&self) -> &'static str {
370        env!("CARGO_PKG_VERSION")
371    }
372
373    fn instantiate(
374        &self,
375        _instance: &PluginInstancePlan,
376    ) -> Result<ConformancePlugin, RuntimeFailure> {
377        Ok(ConformancePlugin::with_event_endpoints(
378            vec![Rc::new(EventProbeEndpoint::new(self.recorder.clone()))],
379            NoopPluginLifecycle,
380        ))
381    }
382}
lenso_runtime_conformance/interaction.rs

lenso_runtime_conformance/
interaction.rs
