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. Explicitly shut down the dispatcher for graceful cleanup. Async
61/// implementations should expose and document a non-blocking async shutdown
62/// entry point for callers running on a Tokio worker.
63pub trait TraceDispatcher: Any {
64 /// Starts the trace dispatcher and establishes connection to the trace backend.
65 ///
66 /// This method initializes the dispatcher's internal state and establishes
67 /// any necessary connections to name servers or trace storage systems.
68 ///
69 /// # Parameters
70 ///
71 /// * `name_srv_addr` - The name server address for service discovery
72 /// * `access_channel` - The access channel configuration for network communication
73 ///
74 /// # Returns
75 ///
76 /// Returns `Ok(())` if the dispatcher started successfully, or an error if
77 /// initialization or connection failed.
78 ///
79 /// # Notes
80 ///
81 /// This method should be called once before any trace data is appended.
82 /// Calling it multiple times may result in undefined behavior.
83 fn start(&self, name_srv_addr: &str, access_channel: AccessChannel) -> crate::ClientResult<()>;
84
85 /// Appends a trace context to the dispatcher's queue for later transmission.
86 ///
87 /// This method adds trace data to an internal buffer. The data is typically
88 /// sent in batches to minimize network overhead and performance impact.
89 ///
90 /// # Parameters
91 ///
92 /// * `ctx` - The trace context containing message metadata. The actual type depends on whether
93 /// this is producer or consumer tracing (e.g., `TraceContext` for producers, `ConsumeContext`
94 /// for consumers).
95 ///
96 /// # Returns
97 ///
98 /// Returns `true` if the context was successfully queued, `false` if the
99 /// queue is full or the dispatcher is not ready.
100 ///
101 /// # Notes
102 ///
103 /// This method should be non-blocking and fast to avoid impacting message
104 /// production or consumption performance.
105 fn append(&self, ctx: &dyn std::any::Any) -> bool;
106
107 /// Flushes all pending trace data to the trace backend.
108 ///
109 /// This method forces immediate transmission of all buffered trace contexts,
110 /// bypassing any batching delay. It's typically called during shutdown or
111 /// at regular intervals to ensure trace data is not lost.
112 ///
113 /// # Returns
114 ///
115 /// Returns `Ok(())` if all pending data was successfully flushed, or an error
116 /// if transmission failed.
117 ///
118 /// # Notes
119 ///
120 /// This operation may block until all data is sent. For production systems,
121 /// consider implementing a timeout mechanism.
122 fn flush(&self) -> crate::ClientResult<()>;
123
124 /// Shuts down the trace dispatcher and releases all resources.
125 ///
126 /// This method performs graceful shutdown by flushing any remaining trace data,
127 /// closing network connections, and cleaning up internal state.
128 ///
129 /// # Notes
130 ///
131 /// After calling this method, the dispatcher should not be used again.
132 /// Any subsequent calls to other methods may result in errors or panics.
133 /// Dropping a dispatcher is only a cancellation fallback and is not
134 /// required to flush pending trace data.
135 fn shutdown(&self);
136
137 /// Returns a reference to this dispatcher as an `Any` trait object.
138 ///
139 /// This method enables runtime type inspection and downcasting to concrete
140 /// dispatcher implementations when needed.
141 ///
142 /// # Returns
143 ///
144 /// A reference to `self` as `&dyn Any`.
145 fn as_any(&self) -> &dyn Any;
146
147 /// Returns a mutable reference to this dispatcher as an `Any` trait object.
148 ///
149 /// This method enables runtime type inspection and mutable downcasting to
150 /// concrete dispatcher implementations when needed.
151 ///
152 /// # Returns
153 ///
154 /// A mutable reference to `self` as `&mut dyn Any`.
155 fn as_mut_any(&mut self) -> &mut dyn Any;
156
157 fn trace_topic_name(&self) -> Option<&str> {
158 None
159 }
160
161 fn trace_client_host(&self) -> Option<CheetahString> {
162 None
163 }
164}
165
166/// Type alias for an atomically reference-counted trace dispatcher.
167///
168/// This is the standard way to share a [`TraceDispatcher`] implementation across multiple
169/// threads or components. The `Arc` provides thread-safe reference counting.
170///
171/// # Thread Safety
172///
173/// `ArcTraceDispatcher` is both `Send` and `Sync`, allowing it to be safely shared
174/// across thread boundaries and accessed concurrently from multiple threads.
175///
176/// # Examples
177///
178/// ```ignore
179/// use std::sync::Arc;
180/// use crate::trace::trace_dispatcher::{TraceDispatcher, ArcTraceDispatcher};
181/// use crate::trace::async_trace_dispatcher::AsyncTraceDispatcher;
182///
183/// // Create a dispatcher
184/// let dispatcher = AsyncTraceDispatcher::new("ProducerGroup", Type::Produce, "TraceTopic", None);
185/// let arc_dispatcher: ArcTraceDispatcher = Arc::new(dispatcher);
186///
187/// // Share across threads
188/// let dispatcher_clone = arc_dispatcher.clone();
189/// ```
190pub type ArcTraceDispatcher = Arc<dyn TraceDispatcher + Send + Sync>;