photon_backend/delivery/
worker_pool.rs

1//! Bounded handler worker pool per subscription partition.
2
3use std::sync::Arc;
4
5use tokio::sync::Semaphore;
6
7/// Limits concurrent handler tasks for a subscription.
8pub struct WorkerPool {
9    semaphore: Arc<Semaphore>,
10}
11
12impl WorkerPool {
13    /// Create a pool with at most `max_concurrent` in-flight handlers.
14    #[must_use]
15    pub fn new(max_concurrent: usize) -> Self {
16        Self {
17            semaphore: Arc::new(Semaphore::new(max_concurrent.max(1))),
18        }
19    }
20
21    /// Pool size from `PHOTON_HANDLER_POOL_SIZE` (default 64).
22    #[must_use]
23    pub fn from_env() -> Self {
24        let max = std::env::var("PHOTON_HANDLER_POOL_SIZE")
25            .ok()
26            .and_then(|s| s.parse().ok())
27            .unwrap_or(64);
28        Self::new(max)
29    }
30
31    /// Acquire a permit before running a handler task.
32    ///
33    /// # Panics
34    ///
35    /// Panics if an internal lock is poisoned.
36    pub async fn acquire(&self) -> tokio::sync::OwnedSemaphorePermit {
37        self.semaphore
38            .clone()
39            .acquire_owned()
40            .await
41            .expect("semaphore closed")
42    }
43}
photon_backend/delivery/worker_pool.rs

photon_backend/delivery/
worker_pool.rs
