hitbox_actix/
runtime.rs

1//! [hitbox::runtime::RuntimeAdapter] implementation for Actix runtime.
2use actix::dev::{MessageResponse, ToEnvelope};
3use actix::{Actor, Addr, Handler, Message};
4use serde::de::DeserializeOwned;
5use serde::Serialize;
6use tracing::warn;
7
8use hitbox::response::CacheableResponse;
9use hitbox::runtime::{AdapterResult, EvictionPolicy, RuntimeAdapter, TtlSettings};
10use hitbox::{CacheError, CacheState, Cacheable, CachedValue};
11use hitbox_backend::{Backend, Get, Set};
12
13use crate::QueryCache;
14
15/// [`RuntimeAdapter`] for Actix runtime.
16pub struct ActixAdapter<A, M, B>
17where
18    A: Actor + Handler<M>,
19    M: Message + Cacheable + Send,
20    M::Result: MessageResponse<A, M> + Send,
21    B: Backend,
22{
23    message: Option<QueryCache<A, M>>,
24    cache_key: String,
25    cache_ttl: u32,
26    cache_stale_ttl: u32,
27    backend: Addr<B>,
28}
29
30impl<A, M, B> ActixAdapter<A, M, B>
31where
32    A: Actor + Handler<M>,
33    M: Message + Cacheable + Send,
34    M::Result: MessageResponse<A, M> + Send,
35    B: Backend,
36{
37    /// Creates new instance of Actix runtime adapter.
38    pub fn new(message: QueryCache<A, M>, backend: Addr<B>) -> Result<Self, CacheError> {
39        let cache_key = message.cache_key()?;
40        let cache_stale_ttl = message.message.cache_ttl();
41        let cache_ttl = message.message.cache_ttl();
42        Ok(Self {
43            message: Some(message),
44            backend,
45            cache_key,
46            cache_ttl,
47            cache_stale_ttl,
48        })
49    }
50}
51
52impl<A, M, T, B, U> RuntimeAdapter for ActixAdapter<A, M, B>
53where
54    A: Actor + Handler<M>,
55    A::Context: ToEnvelope<A, M>,
56    M: Message<Result = T> + Cacheable + Send + 'static,
57    M::Result: MessageResponse<A, M> + Send,
58    B: Backend,
59    <B as Actor>::Context: ToEnvelope<B, Get> + ToEnvelope<B, Set>,
60    T: CacheableResponse<Cached = U> + 'static,
61    U: DeserializeOwned + Serialize,
62{
63    type UpstreamResult = T;
64
65    fn poll_upstream(&mut self) -> AdapterResult<Self::UpstreamResult> {
66        let message = self.message.take();
67        Box::pin(async move {
68            let message = message.ok_or_else(|| {
69                CacheError::CacheKeyGenerationError("Message already sent to upstream".to_owned())
70            })?;
71            Ok(message.upstream.send(message.message).await?)
72        })
73    }
74
75    fn poll_cache(&self) -> AdapterResult<CacheState<Self::UpstreamResult>> {
76        let backend = self.backend.clone();
77        let cache_key = self.cache_key.clone();
78        Box::pin(async move {
79            let cached_value = backend.send(Get { key: cache_key }).await??;
80            CacheState::from_bytes(cached_value.as_ref())
81        })
82    }
83
84    fn update_cache(&self, cached_value: &CachedValue<Self::UpstreamResult>) -> AdapterResult<()> {
85        let serialized = cached_value.serialize();
86        let ttl = self.cache_ttl;
87        let backend = self.backend.clone();
88        let cache_key = self.cache_key.clone();
89        Box::pin(async move {
90            let serialized = serialized?;
91            let _ = backend
92                .send(Set {
93                    key: cache_key,
94                    value: serialized,
95                    ttl: Some(ttl),
96                })
97                .await
98                .map_err(|error| warn!("Updating Cache Error {}", error))
99                .and_then(|value| value.map_err(|error| warn!("Updating Cache Error. {}", error)));
100            Ok(())
101        })
102    }
103    fn eviction_settings(&self) -> EvictionPolicy {
104        let ttl_settings = TtlSettings {
105            ttl: self.cache_ttl,
106            stale_ttl: self.cache_stale_ttl,
107        };
108        EvictionPolicy::Ttl(ttl_settings)
109    }
110}
hitbox_actix/runtime.rs

hitbox_actix/
runtime.rs
