apalis_core/backend/
queue.rs

1//! Represents a queue in the backend
2//!
3//! This module provides the `Queue` struct and related functionality for managing
4//! queues in the backend. A queue is identified by its name and is used to group
5//! tasks for processing by workers.
6//!
7//! The `Queue` struct is designed to be lightweight and easily clonable, allowing
8//! it to be passed around in various contexts. It uses an `Arc<String>` internally
9//! to store the queue name, ensuring efficient memory usage and thread safety.
10//!
11//! The module also includes an implementation of the `FromRequest` trait, allowing
12//! extraction of the queue information from a task context. This is useful for
13//! workers that need to know which queue they are processing tasks from.
14use std::{str::FromStr, sync::Arc};
15
16use crate::task::{
17    Task,
18    from_request::FromRequest,
19    metadata::{Metadata, MetadataStore},
20};
21
22/// Represents a queue in the backend
23#[derive(Debug, Clone, PartialEq, Eq, Hash)]
24pub struct Queue(Arc<str>);
25
26impl From<String> for Queue {
27    fn from(value: String) -> Self {
28        Self(Arc::from(value))
29    }
30}
31impl AsRef<str> for Queue {
32    fn as_ref(&self) -> &str {
33        &self.0
34    }
35}
36
37impl From<&str> for Queue {
38    fn from(value: &str) -> Self {
39        Self(Arc::from(value))
40    }
41}
42
43impl FromStr for Queue {
44    type Err = std::convert::Infallible;
45
46    fn from_str(s: &str) -> Result<Self, Self::Err> {
47        Ok(Self(Arc::from(s)))
48    }
49}
50
51impl std::fmt::Display for Queue {
52    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        write!(f, "{}", self.0)
54    }
55}
56
57#[cfg(feature = "serde")]
58impl serde::Serialize for Queue {
59    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
60    where
61        S: serde::Serializer,
62    {
63        serializer.serialize_str(&self.0)
64    }
65}
66
67#[cfg(feature = "serde")]
68impl<'de> serde::Deserialize<'de> for Queue {
69    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
70    where
71        D: serde::Deserializer<'de>,
72    {
73        let s = String::deserialize(deserializer)?;
74        Ok(Self(Arc::from(s)))
75    }
76}
77
78impl<Args> FromRequest<Task<Args>> for Queue
79where
80    Args: Sync,
81{
82    type Error = QueueError;
83
84    async fn from_request(req: &Task<Args>) -> Result<Self, Self::Error> {
85        let queue = req.queue().cloned().ok_or(QueueError::NotFound)?;
86        Ok(queue)
87    }
88}
89
90/// Errors that can occur when extracting queue information from a task context
91#[derive(Debug, thiserror::Error)]
92#[non_exhaustive]
93pub enum QueueError {
94    /// Queue data not found in task context
95    #[error("Queue data not found in task context. This is likely a bug. Please report it.")]
96    NotFound,
97}
98
99impl Metadata for Queue {
100    type Error = QueueError;
101
102    fn extract(store: &MetadataStore) -> Result<Self, Self::Error> {
103        store
104            .get("queue")
105            .map(|s| Self::from(s.as_str()))
106            .ok_or(QueueError::NotFound)
107    }
108
109    fn inject(&self, map: &mut MetadataStore) -> Result<(), Self::Error> {
110        map.insert("queue", self.0.to_string())
111            .map_err(|_| QueueError::NotFound)
112    }
113}
apalis_core/backend/queue.rs

apalis_core/backend/
queue.rs
