81224d4
ZhenYi
feat(runtime): introduce QueueRuntime for managing RocketMQ client lifecycle and runtime configuration
81224d4
use std::sync::Arc;
use rocketmq_client_rust::{ClientRuntime, ClientRuntimeConfig};
use rocketmq_observability::TelemetryRuntimeGuard;
use rocketmq_runtime::{RuntimeConfig, RuntimeOwner};
use crate::error::{QueueError, QueueResult};
pub struct QueueRuntime {
owner: RuntimeOwner,
client: Arc<ClientRuntime>,
_telemetry: TelemetryRuntimeGuard,
}
impl QueueRuntime {
pub fn connect(scope: &str) -> QueueResult<Self> {
let owner = RuntimeOwner::plan(RuntimeConfig {
thread_name: format!("rocketmq-{scope}"),
..RuntimeConfig::default()
})
.map_err(|error| QueueError::config(format!("invalid rocketmq runtime config: {error}")))?
.build()
.map_err(|error| QueueError::connect(format!("creating the rocketmq runtime: {error}")))?;
let telemetry = TelemetryRuntimeGuard::noop();
let client = ClientRuntime::try_new(
owner.root_context().component("client"),
ClientRuntimeConfig::default(),
telemetry.handle(),
)
.map_err(|error| {
QueueError::connect(format!("creating the rocketmq client runtime: {error}"))
})?;
Ok(Self {
owner,
client,
_telemetry: telemetry,
})
}
pub fn client(&self) -> Arc<ClientRuntime> {
Arc::clone(&self.client)
}
pub async fn shutdown(self) {
let report = self.client.shutdown().await;
if !report.is_healthy() {
tracing::warn!(report = %report.to_json(), "rocketmq client runtime shut down uncleanly");
}
let report = self.owner.shutdown_tasks().await;
if !report.is_healthy() {
tracing::warn!(report = %report.to_json(), "rocketmq runtime tasks shut down uncleanly");
}
let report = self.owner.shutdown_background();
if !report.is_healthy() {
tracing::warn!(report = %report.to_json(), "rocketmq runtime did not shut down cleanly");
}
}
}