rocketmq_client_rust/consumer/
pull_result.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 crate::consumer::pull_status::PullStatus;
16use rocketmq_model::common::message::message_ext::MessageExt;
17
18/// Owned, runtime-neutral pull result used across crate boundaries.
19pub type PullOutcome = rocketmq_model::result::PullOutcome<MessageExt>;
20
21/// Compatibility name for the now-infallible owned pull result adapter.
22pub type PullOutcomeAdapterError = std::convert::Infallible;
23
24pub struct PullResult {
25    pub(crate) pull_status: PullStatus,
26    pub(crate) next_begin_offset: u64,
27    pub(crate) min_offset: u64,
28    pub(crate) max_offset: u64,
29    pub(crate) msg_found_list: Option<Vec<MessageExt>>,
30}
31
32impl PullResult {
33    pub fn new(
34        pull_status: PullStatus,
35        next_begin_offset: u64,
36        min_offset: u64,
37        max_offset: u64,
38        msg_found_list: Option<Vec<MessageExt>>,
39    ) -> Self {
40        Self {
41            pull_status,
42            next_begin_offset,
43            min_offset,
44            max_offset,
45            msg_found_list,
46        }
47    }
48
49    #[inline]
50    pub fn pull_status(&self) -> &PullStatus {
51        &self.pull_status
52    }
53
54    #[inline]
55    pub fn next_begin_offset(&self) -> u64 {
56        self.next_begin_offset
57    }
58
59    #[inline]
60    pub fn min_offset(&self) -> u64 {
61        self.min_offset
62    }
63
64    #[inline]
65    pub fn max_offset(&self) -> u64 {
66        self.max_offset
67    }
68
69    #[inline]
70    pub fn msg_found_list(&self) -> Option<&[MessageExt]> {
71        self.msg_found_list.as_deref()
72    }
73
74    #[inline]
75    pub fn set_msg_found_list(&mut self, msg_found_list: Option<Vec<MessageExt>>) {
76        self.msg_found_list = msg_found_list;
77    }
78}
79
80impl From<&PullResult> for PullOutcome {
81    fn from(value: &PullResult) -> Self {
82        let messages = value.msg_found_list.clone();
83        Self::new(
84            value.pull_status,
85            value.next_begin_offset,
86            value.min_offset,
87            value.max_offset,
88            messages,
89        )
90    }
91}
92
93impl From<PullOutcome> for PullResult {
94    fn from(value: PullOutcome) -> Self {
95        let pull_status = value.pull_status();
96        let next_begin_offset = value.next_begin_offset();
97        let min_offset = value.min_offset();
98        let max_offset = value.max_offset();
99        let messages = value.into_messages();
100        Self::new(pull_status, next_begin_offset, min_offset, max_offset, messages)
101    }
102}
103
104impl std::fmt::Display for PullResult {
105    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        write!(
107            f,
108            "PullResult [pullStatus={}, nextBeginOffset={}, minOffset={}, maxOffset={}, msgFoundList={}]",
109            self.pull_status,
110            self.next_begin_offset,
111            self.min_offset,
112            self.max_offset,
113            self.msg_found_list.as_ref().map_or(0, |v| v.len()),
114        )
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use super::*;
121
122    #[test]
123    fn pull_result_display_matches_java_to_string_shape() {
124        let pull_result = PullResult::new(PullStatus::NoNewMsg, 12, 3, 45, None);
125
126        assert_eq!(
127            pull_result.to_string(),
128            "PullResult [pullStatus=NO_NEW_MSG, nextBeginOffset=12, minOffset=3, maxOffset=45, msgFoundList=0]"
129        );
130    }
131
132    #[test]
133    fn pull_outcome_round_trip_preserves_non_empty_owned_messages() {
134        let mut first = MessageExt::default();
135        first.set_queue_offset(10);
136        let mut second = MessageExt::default();
137        second.set_queue_offset(11);
138        let result = PullResult::new(PullStatus::Found, 12, 1, 20, Some(vec![first, second]));
139
140        let round_trip = PullResult::from(PullOutcome::from(&result));
141
142        let offsets = round_trip
143            .msg_found_list()
144            .expect("non-empty messages should remain present")
145            .iter()
146            .map(|message| message.queue_offset)
147            .collect::<Vec<_>>();
148        assert_eq!(offsets, vec![10, 11]);
149    }
150}
rocketmq_client_rust/consumer/pull_result.rs

rocketmq_client_rust/consumer/
pull_result.rs
