rocketmq_protocol/protocol/body/
query_consume_queue_response_body.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 cheetah_string::CheetahString;
16use serde::Deserialize;
17use serde::Serialize;
18
19use crate::protocol::body::consume_queue_data::ConsumeQueueData;
20use crate::protocol::heartbeat::subscription_data::SubscriptionData;
21
22#[derive(Serialize, Deserialize, Debug, Clone, Default)]
23#[serde(rename_all = "camelCase")]
24pub struct QueryConsumeQueueResponseBody {
25    pub subscription_data: Option<SubscriptionData>,
26    pub filter_data: Option<CheetahString>,
27    pub queue_data: Option<Vec<ConsumeQueueData>>,
28    pub max_queue_index: i64,
29    pub min_queue_index: i64,
30}
31
32#[cfg(test)]
33mod tests {
34    use super::*;
35
36    #[test]
37    fn populated_response_preserves_java_field_names_and_signed_indexes() {
38        let body = QueryConsumeQueueResponseBody {
39            subscription_data: Some(SubscriptionData::default()),
40            filter_data: Some("filter expression".into()),
41            queue_data: Some(vec![ConsumeQueueData::default()]),
42            max_queue_index: i64::MAX,
43            min_queue_index: -1,
44        };
45        let value = serde_json::to_value(&body).unwrap();
46        assert_eq!(
47            value,
48            serde_json::json!({
49                "subscriptionData": body.subscription_data,
50                "filterData": "filter expression",
51                "queueData": body.queue_data,
52                "maxQueueIndex": i64::MAX,
53                "minQueueIndex": -1
54            })
55        );
56        let decoded: QueryConsumeQueueResponseBody = serde_json::from_value(value.clone()).unwrap();
57        assert_eq!(decoded.max_queue_index, i64::MAX);
58        assert_eq!(decoded.min_queue_index, -1);
59        assert_eq!(serde_json::to_value(decoded).unwrap(), value);
60    }
61
62    #[test]
63    fn null_optionals_remain_distinct_from_an_empty_queue_array() {
64        let value = serde_json::json!({
65            "subscriptionData": null, "filterData": null, "queueData": null,
66            "maxQueueIndex": 0, "minQueueIndex": 0
67        });
68        assert_eq!(
69            serde_json::to_value(QueryConsumeQueueResponseBody::default()).unwrap(),
70            value
71        );
72        let mut decoded: QueryConsumeQueueResponseBody = serde_json::from_value(value).unwrap();
73        assert!(decoded.subscription_data.is_none());
74        assert!(decoded.filter_data.is_none());
75        assert!(decoded.queue_data.is_none());
76        decoded.queue_data = Some(Vec::new());
77        let value = serde_json::to_value(decoded).unwrap();
78        assert_eq!(value["queueData"], serde_json::json!([]));
79        let decoded: QueryConsumeQueueResponseBody = serde_json::from_value(value).unwrap();
80        assert!(decoded.queue_data.unwrap().is_empty());
81    }
82}
rocketmq_protocol/protocol/body/query_consume_queue_response_body.rs

rocketmq_protocol/protocol/body/
query_consume_queue_response_body.rs
