rocketmq_protocol/protocol/body/
batch_ack.rs

1// Copyright 2023 The RocketMQ Rust Authors
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//     http://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 bitvec::prelude::BitVec;
16use bitvec::prelude::Lsb0;
17use cheetah_string::CheetahString;
18use serde::Deserialize;
19use serde::Deserializer;
20use serde::Serialize;
21use serde::Serializer;
22
23#[derive(Debug, Serialize, Deserialize)]
24pub struct BatchAck {
25    #[serde(rename = "c", alias = "consumerGroup")]
26    pub consumer_group: CheetahString,
27
28    #[serde(rename = "t", alias = "topic")]
29    pub topic: CheetahString,
30
31    #[serde(rename = "r", alias = "retry")]
32    pub retry: CheetahString, // "1" if it's retry topic
33
34    #[serde(rename = "so", alias = "startOffset")]
35    pub start_offset: i64,
36
37    #[serde(rename = "q", alias = "queueId")]
38    pub queue_id: i32,
39
40    #[serde(rename = "rq", alias = "reviveQueueId")]
41    pub revive_queue_id: i32,
42
43    #[serde(rename = "pt", alias = "popTime")]
44    pub pop_time: i64,
45
46    #[serde(rename = "it", alias = "invisibleTime")]
47    pub invisible_time: i64,
48
49    #[serde(rename = "b", alias = "bitSet")]
50    pub bit_set: SerializableBitVec,
51}
52
53#[derive(Debug)]
54pub struct SerializableBitVec(pub BitVec<u64, Lsb0>);
55
56impl Serialize for SerializableBitVec {
57    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
58    where
59        S: Serializer,
60    {
61        let slice = bytemuck::cast_slice(self.0.as_raw_slice());
62        serializer.serialize_bytes(slice)
63    }
64}
65
66impl<'de> Deserialize<'de> for SerializableBitVec {
67    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
68    where
69        D: Deserializer<'de>,
70    {
71        let bytes: Vec<u8> = Vec::deserialize(deserializer)?;
72        let inner: &[u64] = bytemuck::cast_slice(bytes.as_slice());
73        Ok(SerializableBitVec(BitVec::<u64, Lsb0>::from_slice(inner)))
74    }
75}
76
77#[cfg(test)]
78mod tests {
79    use cheetah_string::CheetahString;
80
81    use super::*;
82
83    #[test]
84    fn serde_round_trip_preserves_the_batch_ack() {
85        let bit_set = BitVec::from_vec(vec![0u64; 8]);
86        let batch_ack = BatchAck {
87            consumer_group: CheetahString::from_static_str("group1"),
88            topic: CheetahString::from_static_str("topic1"),
89            retry: CheetahString::from_static_str("1"),
90            start_offset: 100,
91            queue_id: 1,
92            revive_queue_id: 2,
93            pop_time: 123456789,
94            invisible_time: 987654321,
95            bit_set: SerializableBitVec(bit_set.clone()),
96        };
97
98        let encoded = serde_json::to_vec(&batch_ack).expect("serialize batch ACK");
99        let decoded: BatchAck = serde_json::from_slice(&encoded).expect("deserialize batch ACK");
100        assert_eq!(decoded.consumer_group, batch_ack.consumer_group);
101        assert_eq!(decoded.topic, batch_ack.topic);
102        assert_eq!(decoded.retry, batch_ack.retry);
103        assert_eq!(decoded.start_offset, batch_ack.start_offset);
104        assert_eq!(decoded.queue_id, batch_ack.queue_id);
105        assert_eq!(decoded.revive_queue_id, batch_ack.revive_queue_id);
106        assert_eq!(decoded.pop_time, batch_ack.pop_time);
107        assert_eq!(decoded.invisible_time, batch_ack.invisible_time);
108        assert_eq!(decoded.bit_set.0, bit_set);
109    }
110}
rocketmq_protocol/protocol/body/batch_ack.rs

rocketmq_protocol/protocol/body/
batch_ack.rs
