apalis_core/backend/ext/
with_codec.rs

1//! A utility module for modifying the codec used by a backend.
2//!
3//! This is useful for changing the serialization format of task arguments without altering the underlying backend logic.
4use std::{
5    pin::Pin,
6    task::{Context, Poll},
7};
8
9use futures_sink::Sink;
10use futures_util::SinkExt;
11
12use crate::{backend::*, task::Task, worker::context::WorkerContext};
13
14/// A `Backend` wrapper that swaps out the serialization codec entirely (JSON,
15/// MessagePack, Protobuf, ...) without touching storage logic.
16#[derive(Debug, Clone)]
17#[doc = features_table! {
18    setup = r#"
19        # {
20        #   use apalis_core::backend::memory::MemoryStorage;
21        #   use apalis_core::backend::ext::with_codec::WithCodec;
22        #   let memory = MemoryStorage::new();
23        #.  pub struct MyCodec;
24        #   WithCodec::new(memory, MyCodec)
25        # };
26    "#,
27    Backend => supported("Basic Backend functionality", false),
28    TaskSink => supported("Ability to push new tasks", false),
29    InheritsFeatures => limited("Inherits features from the underlying backend", false),
30}]
31pub struct WithCodec<B, NewCodec> {
32    backend: B,
33    codec: NewCodec,
34}
35
36impl<B, NewCodec> WithCodec<B, NewCodec> {
37    /// Create a new `WithCodec` wrapping the given backend and codec.
38    pub fn new(backend: B, codec: NewCodec) -> Self {
39        Self { backend, codec }
40    }
41}
42
43impl<B, NewCodec> Backend for WithCodec<B, NewCodec>
44where
45    B: Backend,
46{
47    type Task = B::Task;
48    type Error = B::Error;
49
50    fn poll_ready(
51        &mut self,
52        cx: &mut Context<'_>,
53        worker: &WorkerContext,
54    ) -> Poll<Result<(), Self::Error>> {
55        self.backend.poll_ready(cx, worker)
56    }
57
58    fn poll_next(
59        &mut self,
60        cx: &mut Context<'_>,
61        worker: &WorkerContext,
62    ) -> Poll<Option<Result<Self::Task, Self::Error>>> {
63        self.backend.poll_next(cx, worker)
64    }
65    fn poll_close(
66        &mut self,
67        cx: &mut Context<'_>,
68        worker: &WorkerContext,
69    ) -> Poll<Result<(), Self::Error>> {
70        self.backend.poll_close(cx, worker)
71    }
72}
73
74impl<B, NewCodec, Args> WireFormatBackend for WithCodec<B, NewCodec>
75where
76    NewCodec: Codec<Args> + Send + 'static,
77    B: BackendConfig<Args = Args>,
78{
79    type Codec = NewCodec;
80
81    type Compact = NewCodec::Compact;
82
83    fn codec(&self) -> &Self::Codec {
84        &self.codec
85    }
86}
87
88delegate_sink!(WithCodec<B, NewCodec>, backend);
89
90delegate_deref!(WithCodec<B, NewCodec>, backend);
91
92delegate_config!(WithCodec<B, NewCodec>, backend);
93
94delegate_expose!(
95    impl<B, C> for WithCodec<B, C>
96    where {
97        B: Send + Sync + Backend + BackendConfig + WireFormatBackend,
98        C: Send + Sync + 'static,
99        C: Codec<B::Args, Compact = B::Compact> + Send + 'static,
100    }
101    => backend
102);
apalis_core/backend/ext/with_codec.rs

apalis_core/backend/ext/
with_codec.rs
