rocketmq_protocol/protocol/body/
query_assignment_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 std::collections::HashSet;
16
17use rocketmq_model::common::message::message_queue_assignment::MessageQueueAssignment;
18use serde::Deserialize;
19use serde::Serialize;
20
21#[derive(Debug, Clone, Serialize, Deserialize, Default)]
22#[serde(rename_all = "camelCase")]
23pub struct QueryAssignmentResponseBody {
24    pub message_queue_assignments: HashSet<MessageQueueAssignment>,
25}
26
27#[cfg(test)]
28mod tests {
29    use std::collections::HashMap;
30
31    use cheetah_string::CheetahString;
32    use rocketmq_model::common::message::message_enum::MessageRequestMode;
33    use rocketmq_model::message::MessageQueue;
34
35    use super::*;
36    use crate::protocol::RemotingDeserializable;
37    use crate::protocol::RemotingSerializable;
38
39    #[test]
40    fn remoting_round_trip_preserves_the_nested_assignment_contract() {
41        let mut queue = MessageQueue::new();
42        queue.set_topic(CheetahString::from_static_str("topic-a"));
43        queue.set_broker_name(CheetahString::from_static_str("broker-a"));
44        queue.set_queue_id(3);
45
46        let mut attachments = HashMap::new();
47        attachments.insert(
48            CheetahString::from_static_str("key"),
49            CheetahString::from_static_str("value"),
50        );
51        let assignment = MessageQueueAssignment {
52            message_queue: Some(queue),
53            mode: MessageRequestMode::Pop,
54            attachments: Some(attachments),
55        };
56        let body = QueryAssignmentResponseBody {
57            message_queue_assignments: [assignment].into_iter().collect(),
58        };
59
60        let encoded = body.encode().expect("encode query assignment response");
61        let value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded response should be JSON");
62        let assignments = value
63            .get("messageQueueAssignments")
64            .and_then(serde_json::Value::as_array)
65            .expect("encoded assignments should be an array");
66        assert_eq!(assignments.len(), 1);
67        assert!(value.get("message_queue_assignments").is_none());
68
69        let decoded = QueryAssignmentResponseBody::decode(&encoded).expect("decode query assignment response");
70        assert_eq!(decoded.message_queue_assignments.len(), 1);
71        let assignment = decoded
72            .message_queue_assignments
73            .iter()
74            .next()
75            .expect("one assignment should survive the round trip");
76        let queue = assignment
77            .message_queue
78            .as_ref()
79            .expect("the message queue should be preserved");
80        assert_eq!(queue.topic(), "topic-a");
81        assert_eq!(queue.broker_name(), "broker-a");
82        assert_eq!(queue.queue_id(), 3);
83        assert_eq!(assignment.mode, MessageRequestMode::Pop);
84        assert_eq!(
85            assignment
86                .attachments
87                .as_ref()
88                .and_then(|attachments| attachments.get("key")),
89            Some(&CheetahString::from_static_str("value"))
90        );
91    }
92}
rocketmq_protocol/protocol/body/query_assignment_response_body.rs

rocketmq_protocol/protocol/body/
query_assignment_response_body.rs
