photon_backend/instrumentation/
backend.rs

1//! L0 [`PhotonBackend`] decorator for publish telemetry.
2
3use std::pin::Pin;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use futures::stream::Stream;
8use serde_json::Value;
9
10use crate::backend::PhotonBackend;
11use crate::error::Result;
12use crate::models::Event;
13use crate::registry::TopicRegistry;
14
15use super::metrics;
16
17/// Wraps an inner backend and emits `photon_publishes` on successful publish.
18pub struct InstrumentedPhotonBackend {
19    inner: Arc<dyn PhotonBackend>,
20    backend_label: &'static str,
21}
22
23impl InstrumentedPhotonBackend {
24    /// Wrap `inner`, using its [`PhotonBackend::telemetry_label`] for metrics.
25    pub fn new(inner: Arc<dyn PhotonBackend>) -> Self {
26        let backend_label = inner.telemetry_label();
27        Self {
28            inner,
29            backend_label,
30        }
31    }
32}
33
34/// Wrap `inner` with [`InstrumentedPhotonBackend`].
35pub fn wrap_backend(inner: Arc<dyn PhotonBackend>) -> Arc<dyn PhotonBackend> {
36    Arc::new(InstrumentedPhotonBackend::new(inner))
37}
38
39#[async_trait]
40impl PhotonBackend for InstrumentedPhotonBackend {
41    fn telemetry_label(&self) -> &'static str {
42        self.backend_label
43    }
44
45    async fn publish(
46        &self,
47        topic_name: &str,
48        topic_key: Option<&str>,
49        actor_json: Value,
50        payload_json: Value,
51    ) -> Result<String> {
52        match self
53            .inner
54            .publish(topic_name, topic_key, actor_json, payload_json)
55            .await
56        {
57            Ok(id) => {
58                metrics::record_publish(topic_name, self.backend_label);
59                Ok(id)
60            }
61            Err(e) => {
62                metrics::record_publish_error(topic_name, self.backend_label);
63                Err(e)
64            }
65        }
66    }
67
68    fn subscribe(
69        &self,
70        topic_name: String,
71        topic_key_filter: Option<String>,
72        after_seq: Option<i64>,
73    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
74        self.inner
75            .subscribe(topic_name, topic_key_filter, after_seq)
76    }
77
78    async fn get_event(&self, event_id: &str) -> Result<Option<Event>> {
79        self.inner.get_event(event_id).await
80    }
81
82    fn registry(&self) -> &TopicRegistry {
83        self.inner.registry()
84    }
85
86    async fn get_checkpoint_seq(
87        &self,
88        subscription_name: &str,
89        topic_name: &str,
90        topic_key: Option<&str>,
91    ) -> Result<Option<i64>> {
92        self.inner
93            .get_checkpoint_seq(subscription_name, topic_name, topic_key)
94            .await
95    }
96
97    async fn set_checkpoint(
98        &self,
99        subscription_name: &str,
100        topic_name: &str,
101        topic_key: Option<&str>,
102        last_seq: i64,
103    ) -> Result<()> {
104        self.inner
105            .set_checkpoint(subscription_name, topic_name, topic_key, last_seq)
106            .await
107    }
108}
photon_backend/instrumentation/backend.rs

photon_backend/instrumentation/
backend.rs
