rocketmq_protocol/protocol/header/
end_transaction_request_header.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 rocketmq_macros::RequestHeaderCodec;
17use serde::Deserialize;
18use serde::Serialize;
19
20use crate::rpc::rpc_request_header::RpcRequestHeader;
21
22#[derive(Clone, Debug, Serialize, Deserialize, Default, RequestHeaderCodec)]
23#[header(
24    type_id = "rocketmq_protocol::protocol::header::end_transaction_request_header::EndTransactionRequestHeader",
25    java_class = "org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader",
26    validate = "Self::validate"
27)]
28#[serde(rename_all = "camelCase")]
29pub struct EndTransactionRequestHeader {
30    #[serde(default)]
31    #[header(default, default_semantic = "literal:")]
32    pub topic: CheetahString,
33
34    #[header(required)]
35    pub producer_group: CheetahString,
36
37    //ConsumeQueue Offset
38    #[header(required)]
39    pub tran_state_table_offset: i64,
40
41    // Offset of the message in the CommitLog
42    #[header(required)]
43    pub commit_log_offset: i64,
44
45    //TRANSACTION_COMMIT_TYPE,TRANSACTION_ROLLBACK_TYPE,TRANSACTION_NOT_TYPE
46    #[header(required)]
47    pub commit_or_rollback: i32,
48
49    //Whether the check-back is initiated by the Broker
50    #[serde(default)]
51    #[header(default, default_semantic = "literal:false")]
52    pub from_transaction_check: bool,
53
54    #[header(required)]
55    pub msg_id: CheetahString,
56
57    pub transaction_id: Option<CheetahString>,
58
59    #[serde(flatten)]
60    #[header(flatten, presence = "always")]
61    pub rpc_request_header: RpcRequestHeader,
62}
63
64impl EndTransactionRequestHeader {
65    fn validate(&self) -> Result<(), crate::ProtocolContractViolation> {
66        use rocketmq_model::common::sys_flag::message_sys_flag::MessageSysFlag;
67
68        match self.commit_or_rollback {
69            MessageSysFlag::TRANSACTION_NOT_TYPE
70            | MessageSysFlag::TRANSACTION_COMMIT_TYPE
71            | MessageSysFlag::TRANSACTION_ROLLBACK_TYPE => Ok(()),
72            _ => Err(crate::ProtocolContractViolation::Validation {
73                header:
74                    "rocketmq_protocol::protocol::header::end_transaction_request_header::EndTransactionRequestHeader",
75                rule: "supported_transaction_state",
76            }),
77        }
78    }
79}
80
81#[cfg(test)]
82impl EndTransactionRequestHeader {
83    const TOPIC: &'static str = "topic";
84    const PRODUCER_GROUP: &'static str = "producerGroup";
85    const TRAN_STATE_TABLE_OFFSET: &'static str = "tranStateTableOffset";
86    const COMMIT_LOG_OFFSET: &'static str = "commitLogOffset";
87    const COMMIT_OR_ROLLBACK: &'static str = "commitOrRollback";
88    const FROM_TRANSACTION_CHECK: &'static str = "fromTransactionCheck";
89    const MSG_ID: &'static str = "msgId";
90    const TRANSACTION_ID: &'static str = "transactionId";
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[test]
98    fn end_transaction_request_header_serialization() {
99        let header = EndTransactionRequestHeader {
100            topic: CheetahString::from("topic1"),
101            producer_group: CheetahString::from("group1"),
102            tran_state_table_offset: 123,
103            commit_log_offset: 456,
104            commit_or_rollback: 8,
105            from_transaction_check: true,
106            msg_id: CheetahString::from("msg1"),
107            transaction_id: Some(CheetahString::from("tran1")),
108            rpc_request_header: RpcRequestHeader {
109                broker_name: Some(CheetahString::from("broker1")),
110                ..Default::default()
111            },
112        };
113        let json = serde_json::to_string(&header).unwrap();
114        assert!(json.contains("\"topic\":\"topic1\""));
115        assert!(json.contains("\"producerGroup\":\"group1\""));
116        assert!(json.contains("\"tranStateTableOffset\":123"));
117        assert!(json.contains("\"commitLogOffset\":456"));
118        assert!(json.contains("\"commitOrRollback\":8"));
119        assert!(json.contains("\"fromTransactionCheck\":true"));
120        assert!(json.contains("\"msgId\":\"msg1\""));
121        assert!(json.contains("\"transactionId\":\"tran1\""));
122        assert!(json.contains("\"bname\":\"broker1\""));
123    }
124
125    #[test]
126    fn end_transaction_request_header_deserialization() {
127        let json = r#"{"topic":"topic1","producerGroup":"group1","tranStateTableOffset":123,"commitLogOffset":456,"commitOrRollback":8,"fromTransactionCheck":true,"msgId":"msg1","transactionId":"tran1","bname":"broker1"}"#;
128        let header: EndTransactionRequestHeader = serde_json::from_str(json).unwrap();
129        assert_eq!(header.topic, "topic1");
130        assert_eq!(header.producer_group, "group1");
131        assert_eq!(header.tran_state_table_offset, 123);
132        assert_eq!(header.commit_log_offset, 456);
133        assert_eq!(header.commit_or_rollback, 8);
134        assert!(header.from_transaction_check);
135        assert_eq!(header.msg_id, "msg1");
136        assert_eq!(header.transaction_id.as_ref().unwrap(), "tran1");
137        assert_eq!(header.rpc_request_header.broker_name.as_ref().unwrap(), "broker1");
138    }
139    #[test]
140    fn end_transaction_request_header_required_fields() {
141        let json = r#"{"topic":"topic1","producerGroup":"group1","tranStateTableOffset":123,"commitLogOffset":456,"commitOrRollback":8,"fromTransactionCheck":true,"msgId":"msg1"}"#;
142        let header: EndTransactionRequestHeader = serde_json::from_str(json).unwrap();
143        assert_eq!(header.topic, "topic1");
144        assert_eq!(header.producer_group, "group1");
145        assert_eq!(header.tran_state_table_offset, 123);
146        assert_eq!(header.commit_log_offset, 456);
147        assert_eq!(header.commit_or_rollback, 8);
148        assert!(header.from_transaction_check);
149        assert_eq!(header.msg_id, "msg1");
150        assert!(header.transaction_id.is_none());
151    }
152
153    #[test]
154    fn end_transaction_request_header_to_map() {
155        use crate::protocol::command_custom_header::CommandCustomHeader;
156
157        let header = EndTransactionRequestHeader {
158            topic: CheetahString::from("topic1"),
159            producer_group: CheetahString::from("group1"),
160            tran_state_table_offset: 123,
161            commit_log_offset: 456,
162            commit_or_rollback: 8,
163            from_transaction_check: true,
164            msg_id: CheetahString::from("msg1"),
165            transaction_id: Some(CheetahString::from("tran1")),
166            rpc_request_header: RpcRequestHeader::default(),
167        };
168
169        let map = header.to_map().unwrap();
170        assert_eq!(
171            map.get(&CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC))
172                .unwrap(),
173            "topic1"
174        );
175        assert_eq!(
176            map.get(&CheetahString::from_static_str(
177                EndTransactionRequestHeader::PRODUCER_GROUP
178            ))
179            .unwrap(),
180            "group1"
181        );
182        assert_eq!(
183            map.get(&CheetahString::from_static_str(
184                EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET
185            ))
186            .unwrap(),
187            "123"
188        );
189        assert_eq!(
190            map.get(&CheetahString::from_static_str(
191                EndTransactionRequestHeader::COMMIT_LOG_OFFSET
192            ))
193            .unwrap(),
194            "456"
195        );
196        assert_eq!(
197            map.get(&CheetahString::from_static_str(
198                EndTransactionRequestHeader::COMMIT_OR_ROLLBACK
199            ))
200            .unwrap(),
201            "8"
202        );
203        assert_eq!(
204            map.get(&CheetahString::from_static_str(
205                EndTransactionRequestHeader::FROM_TRANSACTION_CHECK
206            ))
207            .unwrap(),
208            "true"
209        );
210        assert_eq!(
211            map.get(&CheetahString::from_static_str(EndTransactionRequestHeader::MSG_ID))
212                .unwrap(),
213            "msg1"
214        );
215        assert_eq!(
216            map.get(&CheetahString::from_static_str(
217                EndTransactionRequestHeader::TRANSACTION_ID
218            ))
219            .unwrap(),
220            "tran1"
221        );
222    }
223
224    #[test]
225    fn end_transaction_request_header_to_map_without_transaction_id() {
226        use crate::protocol::command_custom_header::CommandCustomHeader;
227
228        let header = EndTransactionRequestHeader {
229            topic: CheetahString::from("topic1"),
230            producer_group: CheetahString::from("group1"),
231            tran_state_table_offset: 123,
232            commit_log_offset: 456,
233            commit_or_rollback: 8,
234            from_transaction_check: true,
235            msg_id: CheetahString::from("msg1"),
236            transaction_id: None,
237            rpc_request_header: RpcRequestHeader::default(),
238        };
239
240        let map = header.to_map().unwrap();
241        assert_eq!(
242            map.get(&CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC))
243                .unwrap(),
244            "topic1"
245        );
246        assert!(!map.contains_key(&CheetahString::from_static_str(
247            EndTransactionRequestHeader::TRANSACTION_ID
248        )));
249    }
250
251    #[test]
252    fn end_transaction_request_header_from_map() {
253        use crate::protocol::command_custom_header::FromMap;
254        use std::collections::HashMap;
255
256        let mut map = HashMap::new();
257        map.insert(
258            CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC),
259            CheetahString::from_static_str("topic1"),
260        );
261        map.insert(
262            CheetahString::from_static_str(EndTransactionRequestHeader::PRODUCER_GROUP),
263            CheetahString::from_static_str("group1"),
264        );
265        map.insert(
266            CheetahString::from_static_str(EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET),
267            CheetahString::from_static_str("123"),
268        );
269        map.insert(
270            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_LOG_OFFSET),
271            CheetahString::from_static_str("456"),
272        );
273        map.insert(
274            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_OR_ROLLBACK),
275            CheetahString::from_static_str("8"),
276        );
277        map.insert(
278            CheetahString::from_static_str(EndTransactionRequestHeader::FROM_TRANSACTION_CHECK),
279            CheetahString::from_static_str("true"),
280        );
281        map.insert(
282            CheetahString::from_static_str(EndTransactionRequestHeader::MSG_ID),
283            CheetahString::from_static_str("msg1"),
284        );
285        map.insert(
286            CheetahString::from_static_str(EndTransactionRequestHeader::TRANSACTION_ID),
287            CheetahString::from_static_str("tran1"),
288        );
289
290        let header = <EndTransactionRequestHeader as FromMap>::from(&map).unwrap();
291        assert_eq!(header.topic, "topic1");
292        assert_eq!(header.producer_group, "group1");
293        assert_eq!(header.tran_state_table_offset, 123);
294        assert_eq!(header.commit_log_offset, 456);
295        assert_eq!(header.commit_or_rollback, 8);
296        assert!(header.from_transaction_check);
297        assert_eq!(header.msg_id, "msg1");
298        assert_eq!(header.transaction_id.unwrap(), "tran1");
299    }
300
301    #[test]
302    fn end_transaction_request_header_preserves_java_signed_long_offsets() {
303        use crate::protocol::command_custom_header::CommandCustomHeader;
304        use crate::protocol::command_custom_header::FromMap;
305        use std::collections::HashMap;
306
307        let header = EndTransactionRequestHeader {
308            topic: CheetahString::from("topic1"),
309            producer_group: CheetahString::from("group1"),
310            tran_state_table_offset: -123,
311            commit_log_offset: -456,
312            commit_or_rollback: 8,
313            from_transaction_check: true,
314            msg_id: CheetahString::from("msg1"),
315            transaction_id: None,
316            rpc_request_header: RpcRequestHeader::default(),
317        };
318        let map = header.to_map().unwrap();
319        assert_eq!(
320            map.get(&CheetahString::from_static_str(
321                EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET
322            )),
323            Some(&CheetahString::from_static_str("-123"))
324        );
325        assert_eq!(
326            map.get(&CheetahString::from_static_str(
327                EndTransactionRequestHeader::COMMIT_LOG_OFFSET
328            )),
329            Some(&CheetahString::from_static_str("-456"))
330        );
331
332        let restored = <EndTransactionRequestHeader as FromMap>::from(&HashMap::from([
333            (
334                CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC),
335                CheetahString::from_static_str("topic1"),
336            ),
337            (
338                CheetahString::from_static_str(EndTransactionRequestHeader::PRODUCER_GROUP),
339                CheetahString::from_static_str("group1"),
340            ),
341            (
342                CheetahString::from_static_str(EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET),
343                CheetahString::from_static_str("-123"),
344            ),
345            (
346                CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_LOG_OFFSET),
347                CheetahString::from_static_str("-456"),
348            ),
349            (
350                CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_OR_ROLLBACK),
351                CheetahString::from_static_str("8"),
352            ),
353            (
354                CheetahString::from_static_str(EndTransactionRequestHeader::FROM_TRANSACTION_CHECK),
355                CheetahString::from_static_str("true"),
356            ),
357            (
358                CheetahString::from_static_str(EndTransactionRequestHeader::MSG_ID),
359                CheetahString::from_static_str("msg1"),
360            ),
361        ]))
362        .unwrap();
363
364        assert_eq!(restored.tran_state_table_offset, -123);
365        assert_eq!(restored.commit_log_offset, -456);
366    }
367
368    #[test]
369    fn end_transaction_request_header_from_map_without_transaction_id() {
370        use crate::protocol::command_custom_header::FromMap;
371        use std::collections::HashMap;
372
373        let mut map = HashMap::new();
374        map.insert(
375            CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC),
376            CheetahString::from_static_str("topic1"),
377        );
378        map.insert(
379            CheetahString::from_static_str(EndTransactionRequestHeader::PRODUCER_GROUP),
380            CheetahString::from_static_str("group1"),
381        );
382        map.insert(
383            CheetahString::from_static_str(EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET),
384            CheetahString::from_static_str("123"),
385        );
386        map.insert(
387            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_LOG_OFFSET),
388            CheetahString::from_static_str("456"),
389        );
390        map.insert(
391            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_OR_ROLLBACK),
392            CheetahString::from_static_str("8"),
393        );
394        map.insert(
395            CheetahString::from_static_str(EndTransactionRequestHeader::FROM_TRANSACTION_CHECK),
396            CheetahString::from_static_str("true"),
397        );
398        map.insert(
399            CheetahString::from_static_str(EndTransactionRequestHeader::MSG_ID),
400            CheetahString::from_static_str("msg1"),
401        );
402
403        let header = <EndTransactionRequestHeader as FromMap>::from(&map).unwrap();
404        assert_eq!(header.topic, "topic1");
405        assert_eq!(header.producer_group, "group1");
406        assert_eq!(header.tran_state_table_offset, 123);
407        assert_eq!(header.commit_log_offset, 456);
408        assert_eq!(header.commit_or_rollback, 8);
409        assert!(header.from_transaction_check);
410        assert_eq!(header.msg_id, "msg1");
411        assert!(header.transaction_id.is_none());
412    }
413
414    #[test]
415    fn end_transaction_request_header_from_map_missing_required_field() {
416        use crate::protocol::command_custom_header::FromMap;
417        use std::collections::HashMap;
418
419        #[derive(Debug, Clone, Copy, PartialEq, Eq)]
420        enum RequiredField {
421            ProducerGroup,
422            TranStateTableOffset,
423            CommitLogOffset,
424            CommitOrRollback,
425            MsgId,
426        }
427
428        impl RequiredField {
429            fn as_str(&self) -> &'static str {
430                match self {
431                    RequiredField::ProducerGroup => EndTransactionRequestHeader::PRODUCER_GROUP,
432                    RequiredField::TranStateTableOffset => EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET,
433                    RequiredField::CommitLogOffset => EndTransactionRequestHeader::COMMIT_LOG_OFFSET,
434                    RequiredField::CommitOrRollback => EndTransactionRequestHeader::COMMIT_OR_ROLLBACK,
435                    RequiredField::MsgId => EndTransactionRequestHeader::MSG_ID,
436                }
437            }
438
439            fn test_value(&self) -> &'static str {
440                match self {
441                    RequiredField::ProducerGroup => "group1",
442                    RequiredField::TranStateTableOffset => "123",
443                    RequiredField::CommitLogOffset => "456",
444                    RequiredField::CommitOrRollback => "8",
445                    RequiredField::MsgId => "msg1",
446                }
447            }
448        }
449
450        let all_required_fields: &[RequiredField] = &[
451            RequiredField::ProducerGroup,
452            RequiredField::TranStateTableOffset,
453            RequiredField::CommitLogOffset,
454            RequiredField::CommitOrRollback,
455            RequiredField::MsgId,
456        ];
457
458        for &missing_field in all_required_fields {
459            let mut map = HashMap::new();
460            for &field in all_required_fields {
461                if field != missing_field {
462                    map.insert(
463                        CheetahString::from_static_str(field.as_str()),
464                        CheetahString::from_static_str(field.test_value()),
465                    );
466                }
467            }
468
469            let result = <EndTransactionRequestHeader as FromMap>::from(&map);
470            assert!(
471                result.is_err(),
472                "Expected failure when missing required field: {:?}",
473                missing_field
474            );
475        }
476    }
477
478    #[test]
479    fn end_transaction_request_header_from_map_invalid_numeric_field() {
480        use crate::protocol::command_custom_header::FromMap;
481        use std::collections::HashMap;
482
483        let mut map = HashMap::new();
484        map.insert(
485            CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC),
486            CheetahString::from_static_str("topic1"),
487        );
488        map.insert(
489            CheetahString::from_static_str(EndTransactionRequestHeader::PRODUCER_GROUP),
490            CheetahString::from_static_str("group1"),
491        );
492        map.insert(
493            CheetahString::from_static_str(EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET),
494            CheetahString::from_static_str("invalid"),
495        );
496        map.insert(
497            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_LOG_OFFSET),
498            CheetahString::from_static_str("456"),
499        );
500        map.insert(
501            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_OR_ROLLBACK),
502            CheetahString::from_static_str("8"),
503        );
504        map.insert(
505            CheetahString::from_static_str(EndTransactionRequestHeader::FROM_TRANSACTION_CHECK),
506            CheetahString::from_static_str("true"),
507        );
508        map.insert(
509            CheetahString::from_static_str(EndTransactionRequestHeader::MSG_ID),
510            CheetahString::from_static_str("msg1"),
511        );
512
513        let result = <EndTransactionRequestHeader as FromMap>::from(&map);
514        assert!(result.is_err());
515    }
516
517    #[test]
518    fn end_transaction_request_header_from_map_invalid_bool_field() {
519        use crate::protocol::command_custom_header::FromMap;
520        use std::collections::HashMap;
521
522        let mut map = HashMap::new();
523        map.insert(
524            CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC),
525            CheetahString::from_static_str("topic1"),
526        );
527        map.insert(
528            CheetahString::from_static_str(EndTransactionRequestHeader::PRODUCER_GROUP),
529            CheetahString::from_static_str("group1"),
530        );
531        map.insert(
532            CheetahString::from_static_str(EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET),
533            CheetahString::from_static_str("123"),
534        );
535        map.insert(
536            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_LOG_OFFSET),
537            CheetahString::from_static_str("456"),
538        );
539        map.insert(
540            CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_OR_ROLLBACK),
541            CheetahString::from_static_str("8"),
542        );
543        map.insert(
544            CheetahString::from_static_str(EndTransactionRequestHeader::FROM_TRANSACTION_CHECK),
545            CheetahString::from_static_str("invalid"),
546        );
547        map.insert(
548            CheetahString::from_static_str(EndTransactionRequestHeader::MSG_ID),
549            CheetahString::from_static_str("msg1"),
550        );
551
552        let result = <EndTransactionRequestHeader as FromMap>::from(&map);
553        assert!(result.is_err());
554    }
555
556    #[test]
557    fn end_transaction_request_header_from_map_transaction_state_values() {
558        use crate::protocol::command_custom_header::FromMap;
559        use rocketmq_model::common::sys_flag::message_sys_flag::MessageSysFlag;
560        use std::collections::HashMap;
561
562        struct TestCase {
563            value: i32,
564        }
565
566        let test_cases = &[
567            TestCase {
568                value: MessageSysFlag::TRANSACTION_NOT_TYPE,
569            },
570            TestCase {
571                value: MessageSysFlag::TRANSACTION_COMMIT_TYPE,
572            },
573            TestCase {
574                value: MessageSysFlag::TRANSACTION_ROLLBACK_TYPE,
575            },
576        ];
577
578        for case in test_cases {
579            let mut map = HashMap::new();
580            map.insert(
581                CheetahString::from_static_str(EndTransactionRequestHeader::TOPIC),
582                CheetahString::from_static_str("topic1"),
583            );
584            map.insert(
585                CheetahString::from_static_str(EndTransactionRequestHeader::PRODUCER_GROUP),
586                CheetahString::from_static_str("group1"),
587            );
588            map.insert(
589                CheetahString::from_static_str(EndTransactionRequestHeader::TRAN_STATE_TABLE_OFFSET),
590                CheetahString::from_static_str("123"),
591            );
592            map.insert(
593                CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_LOG_OFFSET),
594                CheetahString::from_static_str("456"),
595            );
596            map.insert(
597                CheetahString::from_static_str(EndTransactionRequestHeader::COMMIT_OR_ROLLBACK),
598                CheetahString::from(case.value.to_string()),
599            );
600            map.insert(
601                CheetahString::from_static_str(EndTransactionRequestHeader::FROM_TRANSACTION_CHECK),
602                CheetahString::from_static_str("true"),
603            );
604            map.insert(
605                CheetahString::from_static_str(EndTransactionRequestHeader::MSG_ID),
606                CheetahString::from_static_str("msg1"),
607            );
608
609            let result = <EndTransactionRequestHeader as FromMap>::from(&map);
610            assert!(
611                result.is_ok(),
612                "commit_or_rollback={} should parse successfully but got {:?}",
613                case.value,
614                result
615            );
616            let header = result.unwrap();
617            assert_eq!(
618                header.commit_or_rollback, case.value,
619                "commit_or_rollback should be {}",
620                case.value
621            );
622        }
623    }
624
625    #[test]
626    fn end_transaction_request_header_rejects_unsupported_transaction_state() {
627        use crate::protocol::command_custom_header::FromMap;
628        use std::collections::HashMap;
629
630        let fields = HashMap::from([
631            ("producerGroup".into(), "group-a".into()),
632            ("tranStateTableOffset".into(), "1".into()),
633            ("commitLogOffset".into(), "2".into()),
634            ("commitOrRollback".into(), "999".into()),
635            ("msgId".into(), "msg-a".into()),
636        ]);
637        assert!(<EndTransactionRequestHeader as FromMap>::from(&fields).is_err());
638    }
639
640    #[test]
641    fn end_transaction_request_header_roundtrip() {
642        use crate::protocol::command_custom_header::CommandCustomHeader;
643        use crate::protocol::command_custom_header::FromMap;
644        use rocketmq_model::common::sys_flag::message_sys_flag::MessageSysFlag;
645
646        let original = EndTransactionRequestHeader {
647            topic: CheetahString::from("topic1"),
648            producer_group: CheetahString::from("group1"),
649            tran_state_table_offset: 123,
650            commit_log_offset: 456,
651            commit_or_rollback: MessageSysFlag::TRANSACTION_COMMIT_TYPE,
652            from_transaction_check: true,
653            msg_id: CheetahString::from("msg1"),
654            transaction_id: Some(CheetahString::from("tran1")),
655            rpc_request_header: RpcRequestHeader::default(),
656        };
657
658        let map = original.to_map().unwrap();
659        let restored = <EndTransactionRequestHeader as FromMap>::from(&map).unwrap();
660
661        assert_eq!(original.topic, restored.topic);
662        assert_eq!(original.producer_group, restored.producer_group);
663        assert_eq!(original.tran_state_table_offset, restored.tran_state_table_offset);
664        assert_eq!(original.commit_log_offset, restored.commit_log_offset);
665        assert_eq!(original.commit_or_rollback, restored.commit_or_rollback);
666        assert_eq!(original.from_transaction_check, restored.from_transaction_check);
667        assert_eq!(original.msg_id, restored.msg_id);
668        assert_eq!(original.transaction_id, restored.transaction_id);
669    }
670
671    #[test]
672    fn end_transaction_request_header_via_remoting_command() {
673        use crate::code::request_code::RequestCode;
674        use crate::protocol::remoting_command::RemotingCommand;
675        use bytes::BytesMut;
676        use rocketmq_model::common::sys_flag::message_sys_flag::MessageSysFlag;
677
678        let original = EndTransactionRequestHeader {
679            topic: CheetahString::from("topic1"),
680            producer_group: CheetahString::from("group1"),
681            tran_state_table_offset: 123,
682            commit_log_offset: 456,
683            commit_or_rollback: MessageSysFlag::TRANSACTION_COMMIT_TYPE,
684            from_transaction_check: true,
685            msg_id: CheetahString::from("msg1"),
686            transaction_id: Some(CheetahString::from("tran1")),
687            rpc_request_header: RpcRequestHeader::default(),
688        };
689
690        let mut command = RemotingCommand::create_request_command(RequestCode::EndTransaction, original);
691        let header_bytes = command.encode_header().expect("Failed to encode header");
692
693        let mut buf = BytesMut::from(header_bytes);
694        let decoded_command = RemotingCommand::decode(&mut buf).expect("Failed to decode command");
695        let decoded_command = decoded_command.expect("Decoded command is None");
696
697        assert_eq!(decoded_command.code(), RequestCode::EndTransaction as i32);
698
699        let decoded_header = decoded_command
700            .decode_command_custom_header::<EndTransactionRequestHeader>()
701            .expect("Failed to decode header");
702
703        assert_eq!(decoded_header.topic, "topic1");
704        assert_eq!(decoded_header.producer_group, "group1");
705        assert_eq!(decoded_header.tran_state_table_offset, 123);
706        assert_eq!(decoded_header.commit_log_offset, 456);
707        assert_eq!(
708            decoded_header.commit_or_rollback,
709            MessageSysFlag::TRANSACTION_COMMIT_TYPE
710        );
711        assert!(decoded_header.from_transaction_check);
712        assert_eq!(decoded_header.msg_id, "msg1");
713        assert_eq!(decoded_header.transaction_id.unwrap(), "tran1");
714    }
715}
rocketmq_protocol/protocol/header/end_transaction_request_header.rs

rocketmq_protocol/protocol/header/
end_transaction_request_header.rs
