google_cloud_pubsub/subscriber/
handler.rs

1// Copyright 2025 Google LLC
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     https://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use tokio::sync::mpsc::UnboundedSender;
16
17/// The action an application does with a message.
18#[derive(Debug, PartialEq)]
19pub(super) enum AckResult {
20    Ack(String),
21    Nack(String),
22    // TODO(#3964) - support exactly once acking
23}
24
25/// A handler for acknowledging or rejecting messages.
26#[derive(Debug)]
27#[non_exhaustive]
28pub enum Handler {
29    AtLeastOnce(AtLeastOnce),
30    // TODO(#3964) - support exactly once acking
31}
32
33impl Handler {
34    /// Acknowledge the message associated with this handler.
35    ///
36    /// Note that the acknowledgement is best effort. The message may still be
37    /// redelivered to this client, or another client.
38    pub fn ack(self) {
39        match self {
40            Handler::AtLeastOnce(h) => h.ack(),
41        }
42    }
43
44    /// Rejects the message associated with this handler.
45    ///
46    /// The message will be removed from this `Subscriber`'s lease management.
47    /// The service will redeliver this message, possibly to another client.
48    pub fn nack(self) {
49        match self {
50            Handler::AtLeastOnce(h) => h.nack(),
51        }
52    }
53}
54
55/// A handler for at-least-once delivery.
56#[derive(Debug)]
57pub struct AtLeastOnce {
58    pub(super) ack_id: String,
59    pub(super) ack_tx: UnboundedSender<AckResult>,
60}
61
62impl AtLeastOnce {
63    /// Acknowledge the message associated with this handler.
64    ///
65    /// Note that the acknowledgement is best effort. The message may still be
66    /// redelivered to this client, or another client.
67    pub fn ack(self) {
68        let _ = self.ack_tx.send(AckResult::Ack(self.ack_id));
69    }
70
71    /// Rejects the message associated with this handler.
72    ///
73    /// The message will be removed from this `Subscriber`'s lease management.
74    /// The service will redeliver this message, possibly to another client.
75    pub fn nack(self) {
76        let _ = self.ack_tx.send(AckResult::Nack(self.ack_id));
77    }
78}
79
80#[cfg(test)]
81mod tests {
82    use super::super::lease_state::tests::test_id;
83    use super::*;
84    use tokio::sync::mpsc::error::TryRecvError;
85    use tokio::sync::mpsc::unbounded_channel;
86
87    #[test]
88    fn handler_ack() -> anyhow::Result<()> {
89        let (ack_tx, mut ack_rx) = unbounded_channel();
90        let h = Handler::AtLeastOnce(AtLeastOnce {
91            ack_id: test_id(1),
92            ack_tx,
93        });
94        assert_eq!(ack_rx.try_recv(), Err(TryRecvError::Empty));
95
96        h.ack();
97        let ack = ack_rx.try_recv()?;
98        assert_eq!(ack, AckResult::Ack(test_id(1)));
99
100        Ok(())
101    }
102
103    #[test]
104    fn handler_nack() -> anyhow::Result<()> {
105        let (ack_tx, mut ack_rx) = unbounded_channel();
106        let h = Handler::AtLeastOnce(AtLeastOnce {
107            ack_id: test_id(1),
108            ack_tx,
109        });
110        assert_eq!(ack_rx.try_recv(), Err(TryRecvError::Empty));
111
112        h.nack();
113        let ack = ack_rx.try_recv()?;
114        assert_eq!(ack, AckResult::Nack(test_id(1)));
115
116        Ok(())
117    }
118
119    #[test]
120    fn at_least_once_ack() -> anyhow::Result<()> {
121        let (ack_tx, mut ack_rx) = unbounded_channel();
122        let h = AtLeastOnce {
123            ack_id: test_id(1),
124            ack_tx,
125        };
126        assert_eq!(ack_rx.try_recv(), Err(TryRecvError::Empty));
127
128        h.ack();
129        let ack = ack_rx.try_recv()?;
130        assert_eq!(ack, AckResult::Ack(test_id(1)));
131
132        Ok(())
133    }
134
135    #[test]
136    fn at_least_once_nack() -> anyhow::Result<()> {
137        let (ack_tx, mut ack_rx) = unbounded_channel();
138        let h = AtLeastOnce {
139            ack_id: test_id(1),
140            ack_tx,
141        };
142        assert_eq!(ack_rx.try_recv(), Err(TryRecvError::Empty));
143
144        h.nack();
145        let ack = ack_rx.try_recv()?;
146        assert_eq!(ack, AckResult::Nack(test_id(1)));
147
148        Ok(())
149    }
150}
google_cloud_pubsub/subscriber/handler.rs

google_cloud_pubsub/subscriber/
handler.rs
