rskit_worker/
bridge.rs

1//! Bridges between [`Handler`] and [`rskit_provider`] traits.
2//!
3//! - `from_provider` wraps a `RequestResponse<I,O>` as a `Handler<I,O>`.
4//!   The handler ignores the event channel (the provider has no concept of intermediate events)
5//!   and returns the provider's result directly.
6//!
7//! - `as_provider` wraps an `Arc<dyn Handler<I,O>>` behind a `RequestResponse` adapter that submits one task,
8//!   awaits its result, and returns it.
9
10use std::sync::Arc;
11
12use tokio::sync::mpsc;
13use tokio_util::sync::CancellationToken;
14
15use rskit_errors::AppResult;
16use rskit_provider::traits::{Provider, RequestResponse};
17
18use crate::event::Event;
19use crate::handler::Handler;
20
21// ---------------------------------------------------------------------------
22// from_provider
23// ---------------------------------------------------------------------------
24
25struct ProviderHandler<I, O, P> {
26    provider: Arc<P>,
27    _ph: std::marker::PhantomData<fn(I) -> O>,
28}
29
30#[async_trait::async_trait]
31impl<I, O, P> Handler<I, O> for ProviderHandler<I, O, P>
32where
33    I: Send + 'static,
34    O: Send + Clone + 'static,
35    P: RequestResponse<I, O> + Send + Sync + 'static,
36{
37    async fn handle(
38        &self,
39        task: I,
40        _emit: mpsc::Sender<Event<O>>,
41        _cancel: CancellationToken,
42    ) -> AppResult<O> {
43        self.provider.execute(task).await
44    }
45}
46
47/// Wrap a [`RequestResponse`] provider as a [`Handler`].
48///
49/// The handler passes no intermediate events — the provider produces only a final result.
50pub fn from_provider<I, O, P>(provider: Arc<P>) -> impl Handler<I, O>
51where
52    I: Send + 'static,
53    O: Send + Clone + 'static,
54    P: RequestResponse<I, O> + Send + Sync + 'static,
55{
56    ProviderHandler {
57        provider,
58        _ph: std::marker::PhantomData,
59    }
60}
61
62// ---------------------------------------------------------------------------
63// as_provider
64// ---------------------------------------------------------------------------
65
66struct HandlerProvider<I, O> {
67    name: &'static str,
68    handler: Arc<dyn Handler<I, O>>,
69}
70
71#[async_trait::async_trait]
72impl<I, O> Provider for HandlerProvider<I, O>
73where
74    I: Send + 'static,
75    O: Send + Clone + 'static,
76{
77    fn name(&self) -> &'static str {
78        self.name
79    }
80}
81
82#[async_trait::async_trait]
83impl<I, O> RequestResponse<I, O> for HandlerProvider<I, O>
84where
85    I: Send + 'static,
86    O: Send + Clone + 'static,
87{
88    async fn execute(&self, input: I) -> AppResult<O> {
89        let (emit_tx, mut emit_rx) = mpsc::channel::<Event<O>>(64);
90        let cancel = CancellationToken::new();
91
92        // Drain events in background —
93        // callers using the provider interface have no event subscription.
94        tokio::spawn(async move { while emit_rx.recv().await.is_some() {} });
95
96        self.handler.handle(input, emit_tx, cancel).await
97    }
98}
99
100/// Wrap an `Arc<dyn Handler<I,O>>` as a [`RequestResponse`] provider.
101pub fn as_provider<I, O>(
102    name: &'static str,
103    handler: Arc<dyn Handler<I, O>>,
104) -> impl RequestResponse<I, O>
105where
106    I: Send + 'static,
107    O: Send + Clone + 'static,
108{
109    HandlerProvider { name, handler }
110}
rskit_worker/bridge.rs

rskit_worker/
bridge.rs
