tansu_storage/service/
get_telemetry_subscriptions.rs

1// Copyright ⓒ 2024-2025 Peter Morgan <peter.james.morgan@gmail.com>
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 rama::{Context, Service};
16use tansu_sans_io::{
17    ApiKey, ErrorCode, GetTelemetrySubscriptionsRequest, GetTelemetrySubscriptionsResponse,
18};
19use tracing::instrument;
20use uuid::Uuid;
21
22use crate::{Error, Result, Storage};
23
24/// A [`Service`] using [`Storage`] as [`Context`] taking [`GetTelemetrySubscriptionsRequest`] returning [`GetTelemetrySubscriptionsResponse`].
25/// ```
26/// use rama::{Context, Layer as _, Service, layer::MapStateLayer};
27/// use tansu_sans_io::{ErrorCode, GetTelemetrySubscriptionsRequest};
28/// use tansu_storage::{Error, GetTelemetrySubscriptionsService, StorageContainer};
29/// use url::Url;
30///
31/// # #[tokio::main]
32/// # async fn main() -> Result<(), Error> {
33///
34/// const HOST: &str = "localhost";
35/// const PORT: i32 = 9092;
36/// const NODE_ID: i32 = 111;
37///
38/// let storage = StorageContainer::builder()
39///     .cluster_id("tansu")
40///     .node_id(NODE_ID)
41///     .advertised_listener(Url::parse(&format!("tcp://{HOST}:{PORT}"))?)
42///     .storage(Url::parse("memory://tansu/")?)
43///     .build()
44///     .await?;
45///
46/// let service = MapStateLayer::new(|_| storage).into_layer(GetTelemetrySubscriptionsService);
47///
48/// let client_instance_id = [0; 16];
49///
50/// let response = service
51///     .serve(
52///         Context::default(),
53///         GetTelemetrySubscriptionsRequest::default().client_instance_id(client_instance_id),
54///     )
55///     .await?;
56///
57/// assert_eq!(ErrorCode::None, ErrorCode::try_from(response.error_code)?);
58/// # Ok(())
59/// # }
60/// ```
61#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
62pub struct GetTelemetrySubscriptionsService;
63
64impl ApiKey for GetTelemetrySubscriptionsService {
65    const KEY: i16 = GetTelemetrySubscriptionsRequest::KEY;
66}
67
68impl<G> Service<G, GetTelemetrySubscriptionsRequest> for GetTelemetrySubscriptionsService
69where
70    G: Storage,
71{
72    type Response = GetTelemetrySubscriptionsResponse;
73    type Error = Error;
74
75    #[instrument(skip(ctx, req))]
76    async fn serve(
77        &self,
78        ctx: Context<G>,
79        req: GetTelemetrySubscriptionsRequest,
80    ) -> Result<Self::Response, Self::Error> {
81        let _ = (ctx, req);
82
83        let client_instance_id = *Uuid::new_v4().as_bytes();
84
85        Ok(GetTelemetrySubscriptionsResponse::default()
86            .throttle_time_ms(0)
87            .error_code(ErrorCode::None.into())
88            .client_instance_id(client_instance_id)
89            .subscription_id(0)
90            .accepted_compression_types(Some([0].into()))
91            .push_interval_ms(5_000)
92            .telemetry_max_bytes(1_024)
93            .delta_temporality(false)
94            .requested_metrics(Some([].into())))
95    }
96}
tansu_storage/service/get_telemetry_subscriptions.rs

tansu_storage/service/
get_telemetry_subscriptions.rs
