rskit-worker 0.2.0-alpha.2

Task worker pool with JoinSet, typed events, and provider bridges
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

//! Bridges between [`Handler`] and [`rskit_provider`] traits.
//!
//! - `from_provider` wraps a `RequestResponse<I,O>` as a `Handler<I,O>`.
//!   The handler ignores the event channel (the provider has no concept of
//!   intermediate events) and returns the provider's result directly.
//!
//! - `as_provider` wraps an `Arc<dyn Handler<I,O>>` behind a `RequestResponse`
//!   adapter that submits one task, awaits its result, and returns it.

use std::sync::Arc;

use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

use rskit_errors::AppResult;
use rskit_provider::traits::{Provider, RequestResponse};

use crate::event::Event;
use crate::handler::Handler;

// ---------------------------------------------------------------------------
// from_provider
// ---------------------------------------------------------------------------

struct ProviderHandler<I, O, P> {
    provider: Arc<P>,
    _ph: std::marker::PhantomData<fn(I) -> O>,
}

#[async_trait::async_trait]
impl<I, O, P> Handler<I, O> for ProviderHandler<I, O, P>
where
    I: Send + 'static,
    O: Send + Clone + 'static,
    P: RequestResponse<I, O> + Send + Sync + 'static,
{
    async fn handle(
        &self,
        task: I,
        _emit: mpsc::Sender<Event<O>>,
        _cancel: CancellationToken,
    ) -> AppResult<O> {
        self.provider.execute(task).await
    }
}

/// Wrap a [`RequestResponse`] provider as a [`Handler`].
///
/// The handler passes no intermediate events — the provider produces only a
/// final result.
pub fn from_provider<I, O, P>(provider: Arc<P>) -> impl Handler<I, O>
where
    I: Send + 'static,
    O: Send + Clone + 'static,
    P: RequestResponse<I, O> + Send + Sync + 'static,
{
    ProviderHandler {
        provider,
        _ph: std::marker::PhantomData,
    }
}

// ---------------------------------------------------------------------------
// as_provider
// ---------------------------------------------------------------------------

struct HandlerProvider<I, O> {
    name: &'static str,
    handler: Arc<dyn Handler<I, O>>,
}

#[async_trait::async_trait]
impl<I, O> Provider for HandlerProvider<I, O>
where
    I: Send + 'static,
    O: Send + Clone + 'static,
{
    fn name(&self) -> &'static str {
        self.name
    }
}

#[async_trait::async_trait]
impl<I, O> RequestResponse<I, O> for HandlerProvider<I, O>
where
    I: Send + 'static,
    O: Send + Clone + 'static,
{
    async fn execute(&self, input: I) -> AppResult<O> {
        let (emit_tx, mut emit_rx) = mpsc::channel::<Event<O>>(64);
        let cancel = CancellationToken::new();

        // Drain events in background — callers using the provider interface
        // have no event subscription.
        tokio::spawn(async move { while emit_rx.recv().await.is_some() {} });

        self.handler.handle(input, emit_tx, cancel).await
    }
}

/// Wrap an `Arc<dyn Handler<I,O>>` as a [`RequestResponse`] provider.
pub fn as_provider<I, O>(
    name: &'static str,
    handler: Arc<dyn Handler<I, O>>,
) -> impl RequestResponse<I, O>
where
    I: Send + 'static,
    O: Send + Clone + 'static,
{
    HandlerProvider { name, handler }
}