rocketmq_observability/runtime_diagnostics/
sources.rs

1// Copyright 2026 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::sync::Arc;
16
17use parking_lot::Mutex;
18use rocketmq_runtime::{MetadataIoObserver, RuntimeDiagnosticsInputs, ScheduledTaskObserver, ShutdownReport};
19
20use super::RuntimeDiagnosticsDataProvider;
21
22#[derive(Default)]
23struct Sources {
24    schedules: Option<ScheduledTaskObserver>,
25    metadata: Option<MetadataIoObserver>,
26    shutdown: Option<ShutdownReport>,
27}
28
29/// Caller-owned, read-only component sources and a retained shutdown summary.
30///
31/// Each source slot replaces its previous value. Schedule observers select fixed
32/// component jobs, and the metadata registry bounds its resource population.
33/// Neither observer keeps execution or write authority alive. Keep a clone at
34/// the composition root to read the final summary after the endpoint has stopped.
35#[derive(Clone, Default)]
36pub struct RuntimeDiagnosticsSources {
37    inner: Arc<Mutex<Sources>>,
38}
39
40impl RuntimeDiagnosticsSources {
41    pub fn set_schedules(&self, observer: ScheduledTaskObserver) {
42        self.inner.lock().schedules = Some(observer);
43    }
44
45    pub fn set_metadata(&self, observer: MetadataIoObserver) {
46        self.inner.lock().metadata = Some(observer);
47    }
48
49    /// Retains only local numeric fields; task names, children and details are discarded.
50    ///
51    /// V2 continues to report the supplied report's local counters. This method
52    /// does not flatten a subtree or reinterpret wrapper completion as success.
53    pub fn retain_shutdown(&self, report: &ShutdownReport) {
54        let mut summary = ShutdownReport::new("", report.elapsed);
55        summary.completed = report.completed;
56        summary.cancelled = report.cancelled;
57        summary.aborted = report.aborted;
58        summary.failed = report.failed;
59        summary.panicked = report.panicked;
60        summary.timed_out = report.timed_out;
61        summary.leaked = report.leaked;
62        summary.blocking_still_running = report.blocking_still_running;
63        self.inner.lock().shutdown = Some(summary);
64    }
65}
66
67impl RuntimeDiagnosticsDataProvider for RuntimeDiagnosticsSources {
68    fn snapshot(&self) -> RuntimeDiagnosticsInputs {
69        let (schedules, metadata, shutdown) = {
70            let sources = self.inner.lock();
71            (
72                sources.schedules.clone(),
73                sources.metadata.clone(),
74                sources.shutdown.clone(),
75            )
76        };
77        RuntimeDiagnosticsInputs {
78            schedule: schedules.map_or_else(Vec::new, |observer| observer.snapshot()),
79            metadata: metadata.and_then(|observer| observer.snapshot()),
80            shutdown,
81        }
82    }
83}
rocketmq_observability/runtime_diagnostics/sources.rs

rocketmq_observability/runtime_diagnostics/
sources.rs
