later 0.0.48

Distributed Background jobs manager and runner for Rust
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

use std::collections::HashMap;

#[cfg(any(feature = "postgres", feature = "sqlite"))]
/// Database-backed delivery queues with renewable work leases.
pub mod sql;

/// Trace context headers attached to a delivered message.
///
/// This implements [`opentelemetry::propagation::Extractor`] so a consumer can
/// restore the producer's tracing context.
#[derive(Clone, Debug, Default)]
pub struct MessageHeaders(HashMap<String, String>);

impl opentelemetry::propagation::Extractor for MessageHeaders {
    fn get(&self, key: &str) -> Option<&str> {
        self.0.get(key).map(String::as_str)
    }

    fn keys(&self) -> Vec<&str> {
        self.0.keys().map(String::as_str).collect()
    }
}

#[async_trait::async_trait]
/// Creates publishers and competing consumers for one delivery system.
///
/// This is a low-level backend extension interface. The built-in backends
/// configure it for normal application use.
pub trait MqClient: Send + Sync {
    /// Creates a consumer for `routing_key` identified by `worker_id`.
    async fn new_consumer(
        &self,
        routing_key: &str,
        worker_id: i32,
    ) -> anyhow::Result<Box<dyn MqConsumer>>;
    /// Creates a publisher for `routing_key`.
    async fn new_publisher(&self, routing_key: &str) -> anyhow::Result<Box<dyn MqPublisher>>;

    /// Number of messages for `routing_key` that are ready to claim right
    /// now (published, not currently leased by a consumer) - backs
    /// [`crate::Config::worker_autoscale`], which needs a live backlog
    /// reading to decide whether to add or remove workers.
    ///
    /// The default returns `0` unconditionally, which is always a *safe*
    /// answer (autoscaling simply never sees a reason to scale up) but not
    /// a useful one for a backend that doesn't override it - the built-in
    /// SQLite and PostgreSQL backends do.
    async fn queue_depth(&self, _routing_key: &str) -> anyhow::Result<usize> {
        Ok(0)
    }
}

#[async_trait::async_trait]
/// Publishes work and reports whether a worker is ready to consume it.
pub trait MqPublisher: Send + Sync {
    /// Makes one encoded message available to a consumer.
    async fn publish(&self, payload: &[u8]) -> anyhow::Result<()>;
    /// Waits until at least one consumer is ready.
    async fn ensure_consumer(&self) -> anyhow::Result<()>;
    /// Returns whether at least one consumer is currently ready.
    async fn has_consumer(&self) -> anyhow::Result<bool>;
}

#[async_trait::async_trait]
/// Receives messages from a shared delivery queue.
pub trait MqConsumer: Send + Sync {
    /// Waits for the next message, or returns `None` after the consumer closes.
    async fn next(&mut self) -> Option<anyhow::Result<Box<dyn MqPayload>>>;
}

#[async_trait::async_trait]
/// One delivered message and its acknowledgement controls.
pub trait MqPayload: Send + Sync {
    /// Confirms successful processing so the message can be removed.
    async fn ack(&self) -> anyhow::Result<()>;
    /// Reports failed processing and asks for the message to be delivered again.
    async fn nack_requeue(&self) -> anyhow::Result<()>;
    /// Reports an invalid message that should not be delivered again.
    ///
    /// Implementations that cannot distinguish discard from acknowledgement
    /// may use the default acknowledgement behavior.
    async fn nack_discard(&self) -> anyhow::Result<()> {
        self.ack().await
    }
    /// Returns tracing headers attached by the producer, when available.
    fn get_headers(&self) -> Option<MessageHeaders>;
    /// Returns the encoded message body.
    fn data(&self) -> &[u8];
}