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

rocketmq_remoting/protocol/header/
end_transaction_request_header.rs
