rocketmq_client_rust/
legacy.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
15#![allow(deprecated)]
16
17use std::collections::HashMap;
18use std::collections::HashSet;
19use std::future::Future;
20use std::pin::Pin;
21use std::sync::Arc;
22use std::sync::Mutex as StdMutex;
23use std::sync::RwLock as StdRwLock;
24use std::time::Duration;
25use std::time::Instant;
26
27use crate::ClientError;
28use crate::ClientResult;
29use cheetah_string::CheetahString;
30use rocketmq_model::common::message::message_ext::MessageExt;
31use rocketmq_model::common::message::message_queue::MessageQueue;
32use rocketmq_protocol::protocol::heartbeat::message_model::MessageModel;
33use rocketmq_runtime::ChildServiceContext;
34use tokio::sync::Mutex;
35
36use crate::producer::local_transaction_state::LocalTransactionState;
37use crate::runtime::schedule_client_fixed_delay_task_with_context;
38use crate::runtime::ClientRuntime;
39use crate::runtime::ClientScheduledTaskHandle;
40
41pub use crate::consumer::consumer_impl::default_mq_pull_consumer_impl::DefaultMQPullConsumerImpl;
42pub use crate::consumer::default_mq_pull_consumer::DefaultMQPullConsumer;
43pub use crate::consumer::default_mq_pull_consumer::MQPullConsumer;
44
45const MODERN_TRANSACTION_LISTENER: &str = "TransactionListener";
46const MODERN_TRACE_HOOKS: &str = "RocketMQ trace hooks";
47
48fn unsupported_legacy_api(api: &str, replacement: &str) -> ClientError {
49    ClientError::illegal_argument(format!(
50        "{api} is deprecated in the RocketMQ Java client and is not supported by rocketmq-client-rust; use \
51         {replacement} instead"
52    ))
53}
54
55fn unsupported_impl_api(api: &str) -> ClientError {
56    ClientError::illegal_argument(format!(
57        "{api} is a RocketMQ Java impl-package type and is not part of the rocketmq-client-rust public API"
58    ))
59}
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62enum ScheduleServiceState {
63    Created,
64    Running,
65    Failed,
66    Shutdown,
67}
68
69struct PullScheduleCore {
70    consumer: DefaultMQPullConsumer,
71    service_context: ChildServiceContext,
72    callbacks: StdRwLock<HashMap<CheetahString, Arc<dyn PullTaskCallback>>>,
73    refresh_interval: StdRwLock<Duration>,
74    state: Mutex<ScheduleServiceState>,
75    coordinator: StdMutex<Option<ClientScheduledTaskHandle>>,
76}
77
78impl PullScheduleCore {
79    fn callback_snapshot(&self) -> HashMap<CheetahString, Arc<dyn PullTaskCallback>> {
80        self.callbacks
81            .read()
82            .unwrap_or_else(|poisoned| poisoned.into_inner())
83            .clone()
84    }
85
86    async fn start(self: &Arc<Self>) -> ClientResult<()> {
87        let mut state = self.state.lock().await;
88        match *state {
89            ScheduleServiceState::Created => {}
90            ScheduleServiceState::Running => {
91                return Err(crate::mq_client_err!("MQPullConsumerScheduleService already started"));
92            }
93            ScheduleServiceState::Failed => {
94                return Err(crate::mq_client_err!(
95                    "MQPullConsumerScheduleService start previously failed; create a new service"
96                ));
97            }
98            ScheduleServiceState::Shutdown => {
99                return Err(crate::mq_client_err!(
100                    "MQPullConsumerScheduleService has been shut down"
101                ));
102            }
103        }
104        if self.callback_snapshot().is_empty() {
105            *state = ScheduleServiceState::Failed;
106            return Err(crate::mq_client_err!("no pull task callback is registered"));
107        }
108        if let Err(error) = self.consumer.start().await {
109            *state = ScheduleServiceState::Failed;
110            return Err(error);
111        }
112        if let Err(error) = self.spawn_coordinator() {
113            self.consumer.shutdown().await?;
114            *state = ScheduleServiceState::Failed;
115            return Err(error);
116        }
117        *state = ScheduleServiceState::Running;
118        Ok(())
119    }
120
121    fn spawn_coordinator(self: &Arc<Self>) -> ClientResult<()> {
122        let core = self.clone();
123        let due = Arc::new(Mutex::new(HashMap::<MessageQueue, Instant>::new()));
124        let refresh_interval = *self
125            .refresh_interval
126            .read()
127            .unwrap_or_else(|poisoned| poisoned.into_inner());
128        let handle = schedule_client_fixed_delay_task_with_context(
129            &self.service_context,
130            "rocketmq-client-classic-pull-schedule",
131            refresh_interval,
132            refresh_interval,
133            move || {
134                let core = core.clone();
135                let due = due.clone();
136                async move {
137                    let mut due = due.lock().await;
138                let mut active_queues = HashSet::new();
139                for (topic, callback) in core.callback_snapshot() {
140                    let queues = match core.consumer.fetch_subscribe_message_queues(topic.as_str()).await {
141                        Ok(queues) => queues,
142                        Err(error) => {
143                            tracing::warn!(topic = %topic, error = %error, "scheduled Classic Pull route refresh failed");
144                            continue;
145                        }
146                    };
147                    for queue in queues {
148                        active_queues.insert(queue.clone());
149                        let now = Instant::now();
150                        if due.get(&queue).is_some_and(|deadline| *deadline > now) {
151                            continue;
152                        }
153                        let mut context = PullTaskContext::with_pull_consumer(core.consumer.clone());
154                        match callback.do_pull_task(&queue, &mut context).await {
155                            Ok(()) => {
156                                let delay = u64::try_from(context.pull_next_delay_time_millis.max(0)).unwrap_or(0);
157                                due.insert(queue, now + Duration::from_millis(delay));
158                            }
159                            Err(error) => {
160                                tracing::warn!(message_queue = %queue, error = %error, "scheduled Classic Pull callback failed");
161                                due.insert(queue, now + Duration::from_millis(3000));
162                            }
163                        }
164                    }
165                }
166                due.retain(|queue, _| active_queues.contains(queue));
167                }
168            },
169        )
170        .map_err(|source| ClientError::service_source("classic_pull_schedule", source))?;
171        *self.coordinator.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(handle);
172        Ok(())
173    }
174
175    async fn shutdown(&self) -> ClientResult<()> {
176        let mut state = self.state.lock().await;
177        match *state {
178            ScheduleServiceState::Shutdown => return Ok(()),
179            ScheduleServiceState::Created => {
180                *state = ScheduleServiceState::Shutdown;
181                return Ok(());
182            }
183            ScheduleServiceState::Running | ScheduleServiceState::Failed => {}
184        }
185        let coordinator = self
186            .coordinator
187            .lock()
188            .unwrap_or_else(|poisoned| poisoned.into_inner())
189            .take();
190        let stopped = match coordinator {
191            Some(coordinator) => coordinator.shutdown(Duration::from_secs(5)).await.is_healthy(),
192            None => true,
193        };
194        self.consumer.shutdown().await?;
195        *state = ScheduleServiceState::Shutdown;
196        if !stopped {
197            return Err(ClientError::timeout("classic pull schedule shutdown", 5000));
198        }
199        Ok(())
200    }
201}
202
203#[deprecated(
204    since = "0.9.0",
205    note = "Classic Pull scheduling is retained for compatibility; prefer application-owned scheduling for new code"
206)]
207#[derive(Clone)]
208pub struct MQPullConsumerScheduleService {
209    consumer_group: CheetahString,
210    core: Option<Arc<PullScheduleCore>>,
211}
212
213impl MQPullConsumerScheduleService {
214    /// Creates a detached compatibility value.
215    ///
216    /// Use [`Self::with_client_runtime`] for a runnable schedule service.
217    pub fn new(consumer_group: impl Into<CheetahString>) -> Self {
218        Self {
219            consumer_group: consumer_group.into(),
220            core: None,
221        }
222    }
223
224    /// Creates a schedule service backed by an application-owned client runtime.
225    ///
226    /// # Errors
227    ///
228    /// Returns an error when the consumer group or underlying Classic Pull configuration is
229    /// invalid.
230    pub fn with_client_runtime(
231        client_runtime: Arc<ClientRuntime>,
232        consumer_group: impl Into<CheetahString>,
233    ) -> ClientResult<Self> {
234        let consumer_group = consumer_group.into();
235        let consumer = DefaultMQPullConsumer::builder(client_runtime.clone())
236            .consumer_group(consumer_group.clone())
237            .build()?;
238        Ok(Self {
239            consumer_group,
240            core: Some(Arc::new(PullScheduleCore {
241                consumer,
242                service_context: client_runtime.component("classic-pull-schedule"),
243                callbacks: StdRwLock::new(HashMap::new()),
244                refresh_interval: StdRwLock::new(Duration::from_secs(1)),
245                state: Mutex::new(ScheduleServiceState::Created),
246                coordinator: StdMutex::new(None),
247            })),
248        })
249    }
250
251    fn core(&self) -> ClientResult<&Arc<PullScheduleCore>> {
252        self.core.as_ref().ok_or_else(|| {
253            ClientError::not_initialized("MQPullConsumerScheduleService has no ClientRuntime; use with_client_runtime")
254        })
255    }
256
257    /// Returns the configured consumer group.
258    pub fn consumer_group(&self) -> &CheetahString {
259        &self.consumer_group
260    }
261
262    /// Starts the consumer and its runtime-owned schedule coordinator.
263    ///
264    /// # Errors
265    ///
266    /// Returns an initialization error for a detached value, a stable lifecycle error for a
267    /// repeated or invalid start, or an underlying consumer or task-spawn error.
268    pub async fn start(&self) -> ClientResult<()> {
269        self.core()?.start().await
270    }
271
272    /// Cancels the coordinator, awaits it, and shuts down the consumer.
273    ///
274    /// # Errors
275    ///
276    /// Returns an initialization error for a detached value, an underlying consumer shutdown
277    /// error, or a timeout when the coordinator does not stop within the shutdown bound.
278    pub async fn shutdown(&self) -> ClientResult<()> {
279        self.core()?.shutdown().await
280    }
281
282    /// Sets how frequently the coordinator refreshes registered topic routes.
283    ///
284    /// # Errors
285    ///
286    /// Returns an error for a zero duration or a detached schedule service.
287    pub fn set_refresh_interval(&self, refresh_interval: Duration) -> ClientResult<()> {
288        if refresh_interval.is_zero() {
289            return Err(crate::mq_client_err!("schedule refresh interval must be positive"));
290        }
291        *self
292            .core()?
293            .refresh_interval
294            .write()
295            .unwrap_or_else(|poisoned| poisoned.into_inner()) = refresh_interval;
296        Ok(())
297    }
298
299    /// Registers or replaces the callback for a topic.
300    ///
301    /// # Errors
302    ///
303    /// Returns an error for a blank topic or a detached schedule service.
304    pub fn register_pull_task_callback<C>(&self, topic: impl Into<CheetahString>, callback: C) -> ClientResult<()>
305    where
306        C: PullTaskCallback,
307    {
308        let topic = topic.into();
309        if topic.trim().is_empty() {
310            return Err(crate::mq_client_err!("topic is blank"));
311        }
312        let core = self.core()?;
313        core.callbacks
314            .write()
315            .unwrap_or_else(|poisoned| poisoned.into_inner())
316            .insert(topic, Arc::new(callback));
317        Ok(())
318    }
319
320    /// Returns the schedule service's Classic Pull consumer.
321    ///
322    /// # Errors
323    ///
324    /// Returns an initialization error for a detached schedule service.
325    pub fn default_mq_pull_consumer(&self) -> ClientResult<DefaultMQPullConsumer> {
326        Ok(self.core()?.consumer.clone())
327    }
328}
329
330#[deprecated(
331    since = "0.9.0",
332    note = "Java PullTaskCallback is deprecated with MQPullConsumerScheduleService"
333)]
334pub trait PullTaskCallback: Send + Sync + 'static {
335    /// Runs one scheduled pull task for a currently readable queue.
336    ///
337    /// The callback may set the next delay through `context`. Returning an error applies the
338    /// schedule service's bounded retry delay.
339    fn do_pull_task<'a>(
340        &'a self,
341        _message_queue: &'a MessageQueue,
342        _context: &'a mut PullTaskContext,
343    ) -> Pin<Box<dyn Future<Output = ClientResult<()>> + Send + 'a>> {
344        Box::pin(async { Ok(()) })
345    }
346}
347
348#[deprecated(
349    since = "0.9.0",
350    note = "Java PullTaskContext is deprecated with MQPullConsumerScheduleService"
351)]
352#[derive(Clone)]
353pub struct PullTaskContext {
354    pull_next_delay_time_millis: i32,
355    pull_consumer: Option<DefaultMQPullConsumer>,
356}
357
358impl std::fmt::Debug for PullTaskContext {
359    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
360        formatter
361            .debug_struct("PullTaskContext")
362            .field("pull_next_delay_time_millis", &self.pull_next_delay_time_millis)
363            .field("has_pull_consumer", &self.pull_consumer.is_some())
364            .finish()
365    }
366}
367
368impl Default for PullTaskContext {
369    fn default() -> Self {
370        Self {
371            pull_next_delay_time_millis: 200,
372            pull_consumer: None,
373        }
374    }
375}
376
377impl PullTaskContext {
378    /// Creates a context without an attached consumer.
379    pub fn new() -> Self {
380        Self::default()
381    }
382
383    fn with_pull_consumer(pull_consumer: DefaultMQPullConsumer) -> Self {
384        Self {
385            pull_consumer: Some(pull_consumer),
386            ..Self::default()
387        }
388    }
389
390    /// Returns the configured delay before this queue's next callback.
391    pub fn pull_next_delay_time_millis(&self) -> i32 {
392        self.pull_next_delay_time_millis
393    }
394
395    /// Java-compatible alias for [`Self::pull_next_delay_time_millis`].
396    pub fn get_pull_next_delay_time_millis(&self) -> i32 {
397        self.pull_next_delay_time_millis()
398    }
399
400    /// Sets the delay before this queue's next callback.
401    pub fn set_pull_next_delay_time_millis(&mut self, pull_next_delay_time_millis: i32) {
402        self.pull_next_delay_time_millis = pull_next_delay_time_millis;
403    }
404
405    /// Returns the consumer attached by the schedule service.
406    ///
407    /// # Errors
408    ///
409    /// Returns an initialization error when the context was constructed directly and no consumer
410    /// has been attached.
411    pub fn get_pull_consumer(&self) -> ClientResult<DefaultMQPullConsumer> {
412        self.pull_consumer
413            .clone()
414            .ok_or_else(|| ClientError::not_initialized("PullTaskContext has no pull consumer"))
415    }
416
417    /// Replaces the consumer attached to this context.
418    pub fn set_pull_consumer(&mut self, pull_consumer: DefaultMQPullConsumer) {
419        self.pull_consumer = Some(pull_consumer);
420    }
421}
422
423#[deprecated(
424    since = "0.9.0",
425    note = "Java PullTaskImpl is deprecated; use MQPullConsumerScheduleService"
426)]
427#[derive(Debug, Clone, Default)]
428pub struct PullTaskImpl {
429    message_queue: Option<MessageQueue>,
430}
431
432impl PullTaskImpl {
433    pub fn new(message_queue: MessageQueue) -> Self {
434        Self {
435            message_queue: Some(message_queue),
436        }
437    }
438
439    pub fn message_queue(&self) -> Option<&MessageQueue> {
440        self.message_queue.as_ref()
441    }
442
443    pub fn run(&self) -> ClientResult<()> {
444        self.message_queue
445            .as_ref()
446            .map(|_| ())
447            .ok_or_else(|| ClientError::not_initialized("PullTaskImpl has no message queue"))
448    }
449}
450
451#[deprecated(
452    since = "0.9.0",
453    note = "Java RebalancePullImpl is deprecated with DefaultMQPullConsumer"
454)]
455#[derive(Debug, Clone, Default)]
456pub struct RebalancePullImpl;
457
458impl RebalancePullImpl {
459    pub fn new() -> ClientResult<Self> {
460        Ok(Self)
461    }
462}
463
464#[deprecated(since = "0.9.0", note = "Java MQHelper depends on DefaultMQPullConsumer")]
465#[derive(Debug, Clone, Default)]
466pub struct MQHelper;
467
468impl MQHelper {
469    /// Retains the Java-shaped helper signature without creating a hidden runtime.
470    ///
471    /// Use [`Self::reset_offset_by_timestamp_with_client_runtime`] for the runnable equivalent.
472    ///
473    /// # Errors
474    ///
475    /// Always returns an initialization error because this compatibility signature has no
476    /// application-owned [`ClientRuntime`].
477    pub fn reset_offset_by_timestamp(
478        _message_model: impl Into<CheetahString>,
479        _consumer_group: impl Into<CheetahString>,
480        _topic: impl Into<CheetahString>,
481        _timestamp: u64,
482    ) -> ClientResult<()> {
483        Err(ClientError::not_initialized(
484            "MQHelper requires an application-owned ClientRuntime; use reset_offset_by_timestamp_with_client_runtime",
485        ))
486    }
487
488    /// Resets every queue offset for a topic to the position nearest the timestamp.
489    ///
490    /// The temporary Classic Pull consumer persists updated offsets during its bounded shutdown.
491    ///
492    /// # Errors
493    ///
494    /// Returns a configuration, startup, route, offset-query, offset-store, or shutdown error.
495    pub async fn reset_offset_by_timestamp_with_client_runtime(
496        client_runtime: Arc<ClientRuntime>,
497        message_model: MessageModel,
498        consumer_group: impl Into<CheetahString>,
499        topic: impl Into<CheetahString>,
500        timestamp: u64,
501    ) -> ClientResult<()> {
502        let consumer = DefaultMQPullConsumer::builder(client_runtime)
503            .consumer_group(consumer_group)
504            .message_model(message_model)
505            .build()?;
506        let topic = topic.into();
507        consumer.start().await?;
508
509        let operation = async {
510            let mut queues = consumer.fetch_subscribe_message_queues(topic.as_str()).await?;
511            queues.sort();
512            for queue in queues {
513                let offset = consumer.search_offset(&queue, timestamp).await?;
514                if offset >= 0 {
515                    consumer.update_consume_offset(&queue, offset).await?;
516                }
517            }
518            Ok(())
519        }
520        .await;
521        let shutdown = consumer.shutdown().await;
522
523        match operation {
524            Ok(()) => shutdown,
525            Err(error) => {
526                if let Err(shutdown_error) = shutdown {
527                    tracing::warn!(%shutdown_error, "MQHelper consumer shutdown failed after offset reset error");
528                }
529                Err(error)
530            }
531        }
532    }
533}
534
535#[deprecated(
536    since = "0.9.0",
537    note = "Java TransactionCheckListener is deprecated; use TransactionListener"
538)]
539pub trait TransactionCheckListener {
540    fn check_local_transaction_state(&self, _message: &MessageExt) -> ClientResult<LocalTransactionState> {
541        Err(unsupported_legacy_api(
542            "TransactionCheckListener",
543            MODERN_TRANSACTION_LISTENER,
544        ))
545    }
546}
547
548#[deprecated(
549    since = "0.9.0",
550    note = "Java OpenTracing hooks are not part of the modern RocketMQ client trace API"
551)]
552#[derive(Debug, Clone, Default)]
553pub struct SendMessageOpenTracingHookImpl;
554
555impl SendMessageOpenTracingHookImpl {
556    pub fn new<T>(_tracer: T) -> Self {
557        Self
558    }
559
560    pub fn hook_name(&self) -> &'static str {
561        "SendMessageOpenTracingHook"
562    }
563
564    pub fn unsupported(&self) -> ClientResult<()> {
565        Err(unsupported_legacy_api(
566            "SendMessageOpenTracingHookImpl",
567            MODERN_TRACE_HOOKS,
568        ))
569    }
570}
571
572#[deprecated(
573    since = "0.9.0",
574    note = "Java OpenTracing hooks are not part of the modern RocketMQ client trace API"
575)]
576#[derive(Debug, Clone, Default)]
577pub struct ConsumeMessageOpenTracingHookImpl;
578
579impl ConsumeMessageOpenTracingHookImpl {
580    pub fn new<T>(_tracer: T) -> Self {
581        Self
582    }
583
584    pub fn hook_name(&self) -> &'static str {
585        "ConsumeMessageOpenTracingHook"
586    }
587
588    pub fn unsupported(&self) -> ClientResult<()> {
589        Err(unsupported_legacy_api(
590            "ConsumeMessageOpenTracingHookImpl",
591            MODERN_TRACE_HOOKS,
592        ))
593    }
594}
595
596#[deprecated(
597    since = "0.9.0",
598    note = "Java OpenTracing hooks are not part of the modern RocketMQ client trace API"
599)]
600#[derive(Debug, Clone, Default)]
601pub struct EndTransactionOpenTracingHookImpl;
602
603impl EndTransactionOpenTracingHookImpl {
604    pub fn new<T>(_tracer: T) -> Self {
605        Self
606    }
607
608    pub fn hook_name(&self) -> &'static str {
609        "EndTransactionOpenTracingHook"
610    }
611
612    pub fn unsupported(&self) -> ClientResult<()> {
613        Err(unsupported_legacy_api(
614            "EndTransactionOpenTracingHookImpl",
615            MODERN_TRACE_HOOKS,
616        ))
617    }
618}
619
620#[derive(Debug, Clone, Default)]
621pub struct DoNothingClientRemotingProcessor;
622
623impl DoNothingClientRemotingProcessor {
624    pub fn new() -> Self {
625        Self
626    }
627
628    pub fn process_request(&self) -> Option<()> {
629        None
630    }
631}
632
633#[derive(Debug, Clone, Default)]
634pub struct RebalanceImpl;
635
636impl RebalanceImpl {
637    pub fn new() -> ClientResult<Self> {
638        Err(unsupported_impl_api("RebalanceImpl"))
639    }
640
641    pub fn do_rebalance(&self) -> ClientResult<()> {
642        Err(unsupported_impl_api("RebalanceImpl"))
643    }
644}
645
646#[derive(Debug, Clone, Default)]
647pub struct ConsumeRequest;
648
649impl ConsumeRequest {
650    pub fn new() -> Self {
651        Self
652    }
653
654    pub fn run(&self) -> ClientResult<()> {
655        Err(unsupported_impl_api("ConsumeRequest"))
656    }
657}
658
659#[cfg(test)]
660#[allow(deprecated)]
661mod tests {
662    use super::*;
663    use crate::test_support::error_assertions::assert_invalid_argument;
664    use crate::test_support::error_assertions::assert_not_initialized;
665
666    #[tokio::test]
667    async fn legacy_pull_consumer_fails_closed_without_runtime() {
668        let consumer = DefaultMQPullConsumer::with_consumer_group("LegacyGroup");
669
670        assert_eq!(
671            consumer.consumer_group().map(|group| group.as_str()),
672            Some("LegacyGroup")
673        );
674        let error = consumer
675            .start()
676            .await
677            .expect_err("detached pull consumer should not start");
678
679        assert_not_initialized(
680            &error,
681            "DefaultMQPullConsumer has no ClientRuntime; create it with DefaultMQPullConsumer::builder",
682        );
683    }
684
685    #[test]
686    fn pull_task_context_preserves_java_default_delay() {
687        let mut context = PullTaskContext::new();
688
689        assert_eq!(context.get_pull_next_delay_time_millis(), 200);
690        context.set_pull_next_delay_time_millis(500);
691        assert_eq!(context.pull_next_delay_time_millis(), 500);
692    }
693
694    #[test]
695    fn detached_schedule_service_requires_runtime_for_callbacks() {
696        struct Callback;
697        impl PullTaskCallback for Callback {}
698
699        let service = MQPullConsumerScheduleService::new("LegacyGroup");
700        let error = service
701            .register_pull_task_callback("TopicA", Callback)
702            .expect_err("detached schedule service should require runtime");
703
704        assert_not_initialized(
705            &error,
706            "MQPullConsumerScheduleService has no ClientRuntime; use with_client_runtime",
707        );
708    }
709
710    #[test]
711    fn legacy_open_tracing_hooks_keep_java_hook_names() {
712        assert_eq!(
713            SendMessageOpenTracingHookImpl::new(()).hook_name(),
714            "SendMessageOpenTracingHook"
715        );
716        assert_eq!(
717            ConsumeMessageOpenTracingHookImpl::new(()).hook_name(),
718            "ConsumeMessageOpenTracingHook"
719        );
720        assert_eq!(
721            EndTransactionOpenTracingHookImpl::new(()).hook_name(),
722            "EndTransactionOpenTracingHook"
723        );
724    }
725
726    #[test]
727    fn transaction_check_listener_returns_typed_unsupported_error() {
728        struct Listener;
729        impl TransactionCheckListener for Listener {}
730
731        let error = Listener
732            .check_local_transaction_state(&MessageExt::default())
733            .expect_err("deprecated transaction listener should reject checks");
734
735        assert_invalid_argument(&error);
736    }
737
738    #[test]
739    fn impl_package_markers_do_not_panic() {
740        assert!(DoNothingClientRemotingProcessor::new().process_request().is_none());
741
742        let rebalance_error = RebalanceImpl::new().expect_err("impl-package RebalanceImpl should be unsupported");
743        assert_invalid_argument(&rebalance_error);
744
745        let consume_error = ConsumeRequest::new()
746            .run()
747            .expect_err("impl-package ConsumeRequest should be unsupported");
748        assert_invalid_argument(&consume_error);
749    }
750}
rocketmq_client_rust/legacy.rs

rocketmq_client_rust/
legacy.rs
