rocketmq_client_rust/trace/trace_dispatcher.rs
1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::any::Any;
16use std::sync::Arc;
17
18use cheetah_string::CheetahString;
19
20use crate::base::access_channel::AccessChannel;
21
22/// The type of tracing operation being performed.
23///
24/// This enum distinguishes between producer-side and consumer-side tracing,
25/// allowing the trace dispatcher to handle different types of message flow tracking.
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub enum Type {
28 /// Tracing for message production/sending operations.
29 Produce,
30 /// Tracing for message consumption/receiving operations.
31 Consume,
32}
33
34/// A dispatcher that manages the collection and transmission of message tracing data.
35///
36/// The `TraceDispatcher` trait defines the interface for components that collect
37/// message tracing information and send it to a trace storage backend. It supports
38/// both producer and consumer side tracing, enabling end-to-end message flow tracking
39/// in RocketMQ applications.
40///
41/// # Purpose
42///
43/// Message tracing provides visibility into:
44/// - Message production: send time, broker, topic, tags, keys
45/// - Message consumption: consume time, status, retry times
46/// - Message routing: which brokers and queues handled the message
47///
48/// # Implementation Notes
49///
50/// Implementations must be thread-safe (`Send + Sync`) as they may be accessed
51/// concurrently from multiple producer or consumer threads. The dispatcher typically
52/// batches trace data and sends it asynchronously to minimize performance impact.
53///
54/// # Lifecycle
55///
56/// 1. Create and configure the dispatcher
57/// 2. Call [`start`](TraceDispatcher::start) to initialize and connect
58/// 3. Call [`append`](TraceDispatcher::append) to add trace contexts
59/// 4. Call [`flush`](TraceDispatcher::flush) periodically or before shutdown
60/// 5. Call [`shutdown`](TraceDispatcher::shutdown) for graceful cleanup
61pub trait TraceDispatcher: Any {
62 /// Starts the trace dispatcher and establishes connection to the trace backend.
63 ///
64 /// This method initializes the dispatcher's internal state and establishes
65 /// any necessary connections to name servers or trace storage systems.
66 ///
67 /// # Parameters
68 ///
69 /// * `name_srv_addr` - The name server address for service discovery
70 /// * `access_channel` - The access channel configuration for network communication
71 ///
72 /// # Returns
73 ///
74 /// Returns `Ok(())` if the dispatcher started successfully, or an error if
75 /// initialization or connection failed.
76 ///
77 /// # Notes
78 ///
79 /// This method should be called once before any trace data is appended.
80 /// Calling it multiple times may result in undefined behavior.
81 fn start(&self, name_srv_addr: &str, access_channel: AccessChannel) -> rocketmq_error::RocketMQResult<()>;
82
83 /// Appends a trace context to the dispatcher's queue for later transmission.
84 ///
85 /// This method adds trace data to an internal buffer. The data is typically
86 /// sent in batches to minimize network overhead and performance impact.
87 ///
88 /// # Parameters
89 ///
90 /// * `ctx` - The trace context containing message metadata. The actual type depends on whether
91 /// this is producer or consumer tracing (e.g., `TraceContext` for producers, `ConsumeContext`
92 /// for consumers).
93 ///
94 /// # Returns
95 ///
96 /// Returns `true` if the context was successfully queued, `false` if the
97 /// queue is full or the dispatcher is not ready.
98 ///
99 /// # Notes
100 ///
101 /// This method should be non-blocking and fast to avoid impacting message
102 /// production or consumption performance.
103 fn append(&self, ctx: &dyn std::any::Any) -> bool;
104
105 /// Flushes all pending trace data to the trace backend.
106 ///
107 /// This method forces immediate transmission of all buffered trace contexts,
108 /// bypassing any batching delay. It's typically called during shutdown or
109 /// at regular intervals to ensure trace data is not lost.
110 ///
111 /// # Returns
112 ///
113 /// Returns `Ok(())` if all pending data was successfully flushed, or an error
114 /// if transmission failed.
115 ///
116 /// # Notes
117 ///
118 /// This operation may block until all data is sent. For production systems,
119 /// consider implementing a timeout mechanism.
120 fn flush(&self) -> rocketmq_error::RocketMQResult<()>;
121
122 /// Shuts down the trace dispatcher and releases all resources.
123 ///
124 /// This method performs graceful shutdown by flushing any remaining trace data,
125 /// closing network connections, and cleaning up internal state.
126 ///
127 /// # Notes
128 ///
129 /// After calling this method, the dispatcher should not be used again.
130 /// Any subsequent calls to other methods may result in errors or panics.
131 fn shutdown(&self);
132
133 /// Returns a reference to this dispatcher as an `Any` trait object.
134 ///
135 /// This method enables runtime type inspection and downcasting to concrete
136 /// dispatcher implementations when needed.
137 ///
138 /// # Returns
139 ///
140 /// A reference to `self` as `&dyn Any`.
141 fn as_any(&self) -> &dyn Any;
142
143 /// Returns a mutable reference to this dispatcher as an `Any` trait object.
144 ///
145 /// This method enables runtime type inspection and mutable downcasting to
146 /// concrete dispatcher implementations when needed.
147 ///
148 /// # Returns
149 ///
150 /// A mutable reference to `self` as `&mut dyn Any`.
151 fn as_mut_any(&mut self) -> &mut dyn Any;
152
153 fn trace_topic_name(&self) -> Option<&str> {
154 None
155 }
156
157 fn trace_client_host(&self) -> Option<CheetahString> {
158 None
159 }
160}
161
162/// Type alias for an atomically reference-counted trace dispatcher.
163///
164/// This is the standard way to share a [`TraceDispatcher`] implementation across multiple
165/// threads or components. The `Arc` provides thread-safe reference counting.
166///
167/// # Thread Safety
168///
169/// `ArcTraceDispatcher` is both `Send` and `Sync`, allowing it to be safely shared
170/// across thread boundaries and accessed concurrently from multiple threads.
171///
172/// # Examples
173///
174/// ```ignore
175/// use std::sync::Arc;
176/// use crate::trace::trace_dispatcher::{TraceDispatcher, ArcTraceDispatcher};
177/// use crate::trace::async_trace_dispatcher::AsyncTraceDispatcher;
178///
179/// // Create a dispatcher
180/// let dispatcher = AsyncTraceDispatcher::new("ProducerGroup", Type::Produce, "TraceTopic", None);
181/// let arc_dispatcher: ArcTraceDispatcher = Arc::new(dispatcher);
182///
183/// // Share across threads
184/// let dispatcher_clone = arc_dispatcher.clone();
185/// ```
186pub type ArcTraceDispatcher = Arc<dyn TraceDispatcher + Send + Sync>;