pub struct StreamingManager { /* private fields */ }streaming only.Expand description
Streaming cache manager backed by redb (metadata) + tokio::fs (bodies).
This implementation provides:
- Persistence across restarts: metadata lives in an on-disk redb database, not just in-memory moka — cached entries survive process restarts.
- True streaming reads: Cached responses are streamed from disk in 64KB chunks, not loaded fully into memory.
- Single-instance enforcement: redb’s file lock prevents multiple
StreamingManagers from operating on the same cache_dir concurrently. - Crash-safe writes: atomic rename + 16-byte nonce header detect overwrite-crash corruption; orphan tmp files are swept on startup.
- Body size limits: Configurable max body size to prevent memory exhaustion.
§Only one instance per cache directory
Only one StreamingManager may point at a given cache_dir at a time
(enforced by redb’s internal file lock on metadata.redb). Cloning an
existing StreamingManager is fine — construction via [new], [with_max_body_size],
or [with_temp_dir] against a directory already in use will fail. This
guarantee is reliable on local filesystems; on NFS or container overlay
filesystems it is best-effort. Do not share a cache directory across hosts.
§Example
use http_cache::StreamingManager;
use std::path::PathBuf;
let manager = StreamingManager::new(PathBuf::from("./cache"), 10_000).await?;Implementations§
Source§impl StreamingManager
impl StreamingManager
Sourcepub async fn new(
cache_dir: PathBuf,
capacity: u64,
) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
pub async fn new( cache_dir: PathBuf, capacity: u64, ) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
Creates a new StreamingManager with disk-backed storage.
Uses the default maximum body size of 100MB. For custom limits,
use StreamingManager::with_max_body_size.
§Single-instance invariant
Only one StreamingManager may operate on a given cache_dir at a
time. Construction fails if another instance in any process currently
holds the metadata.redb file lock.
§Arguments
cache_dir- Directory to store cached response bodies and metadatacapacity- Maximum number of metadata entries in the in-memory hot cache
§Example
use http_cache::StreamingManager;
use std::path::PathBuf;
let manager = StreamingManager::new(PathBuf::from("./cache"), 10_000).await?;Sourcepub async fn with_max_body_size(
cache_dir: PathBuf,
capacity: u64,
max_body_size: u64,
) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
pub async fn with_max_body_size( cache_dir: PathBuf, capacity: u64, max_body_size: u64, ) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
Creates a new StreamingManager with a custom maximum body size.
See StreamingManager::new for details on the single-instance
invariant.
§Arguments
cache_dir- Directory to store cached response bodies and metadatacapacity- Maximum number of metadata entries in the in-memory hot cachemax_body_size- Maximum body size in bytes (responses larger than this are not cached — caching is declined and the body still streams through to the caller)
§Example
use http_cache::StreamingManager;
use std::path::PathBuf;
let manager = StreamingManager::with_max_body_size(
PathBuf::from("./cache"),
10_000,
50 * 1024 * 1024,
).await?;Sourcepub async fn in_memory(
capacity: u64,
) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
👎Deprecated since 1.1.0: renamed to with_temp_dir() for clarity
pub async fn in_memory( capacity: u64, ) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
renamed to with_temp_dir() for clarity
Creates a new StreamingManager using a temporary directory.
Note: Despite the historical name, this still uses disk storage in a temp directory. Only metadata is kept in memory; response bodies are stored on disk and streamed.
Use StreamingManager::new with a persistent directory for production deployments.
§Arguments
capacity- Maximum number of entries in the cache
Sourcepub async fn with_temp_dir(
capacity: u64,
) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
pub async fn with_temp_dir( capacity: u64, ) -> Result<StreamingManager, Box<dyn Error + Send + Sync>>
Creates a new StreamingManager using a temporary directory.
This is useful for testing or when persistence is not needed. The cache directory is created in the system’s temporary directory with a unique name including process ID and random component for security.
Note: This still uses disk storage in a temp directory. Only metadata is kept in memory; response bodies are stored on disk and streamed.
§Arguments
capacity- Maximum number of entries in the cache
§Example
use http_cache::StreamingManager;
let manager = StreamingManager::with_temp_dir(1000).await?;Sourcepub fn entry_count(&self) -> u64
pub fn entry_count(&self) -> u64
Returns the current number of entries in the in-memory hot cache.
Note: this is not the total number of persisted entries. Hydration
is lazy, so this reads 0 after a restart until keys are accessed,
and once capacity is exceeded, cold entries remain on disk
(reachable via get) but are not counted here.
Sourcepub fn max_body_size(&self) -> u64
pub fn max_body_size(&self) -> u64
Returns the maximum body size for cached responses.
Sourcepub async fn clear(&self) -> Result<(), Box<dyn Error + Send + Sync>>
pub async fn clear(&self) -> Result<(), Box<dyn Error + Send + Sync>>
Clears all entries from the cache — moka, redb, and on-disk bodies.
Sourcepub async fn run_pending_tasks(&self)
pub async fn run_pending_tasks(&self)
Runs pending maintenance tasks (eviction, etc).
This is called automatically but can be invoked manually to force immediate cleanup.
Trait Implementations§
Source§impl Clone for StreamingManager
impl Clone for StreamingManager
Source§fn clone(&self) -> StreamingManager
fn clone(&self) -> StreamingManager
1.0.0 (const: unstable) · Source§fn clone_from(&mut self, source: &Self)
fn clone_from(&mut self, source: &Self)
source. Read moreSource§impl Debug for StreamingManager
impl Debug for StreamingManager
Source§impl StreamingCacheManager for StreamingManager
impl StreamingCacheManager for StreamingManager
Source§type Body = StreamingBody<UnsyncBoxBody<Bytes, StreamingError>>
type Body = StreamingBody<UnsyncBoxBody<Bytes, StreamingError>>
Source§async fn get(
&self,
cache_key: &str,
) -> Result<Option<(Response<<StreamingManager as StreamingCacheManager>::Body>, CachePolicy)>, Box<dyn Error + Send + Sync>>where
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Into<StreamingError> + Send + Sync + 'static,
async fn get(
&self,
cache_key: &str,
) -> Result<Option<(Response<<StreamingManager as StreamingCacheManager>::Body>, CachePolicy)>, Box<dyn Error + Send + Sync>>where
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Into<StreamingError> + Send + Sync + 'static,
Source§async fn put<B>(
&self,
cache_key: String,
response: Response<B>,
policy: CachePolicy,
_request_url: Url,
user_metadata: Option<Vec<u8>>,
) -> Result<Response<<StreamingManager as StreamingCacheManager>::Body>, Box<dyn Error + Send + Sync>>where
B: Body + Send + 'static,
<B as Body>::Data: Send,
<B as Body>::Error: Into<StreamingError>,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Into<StreamingError> + Send + Sync + 'static,
async fn put<B>(
&self,
cache_key: String,
response: Response<B>,
policy: CachePolicy,
_request_url: Url,
user_metadata: Option<Vec<u8>>,
) -> Result<Response<<StreamingManager as StreamingCacheManager>::Body>, Box<dyn Error + Send + Sync>>where
B: Body + Send + 'static,
<B as Body>::Data: Send,
<B as Body>::Error: Into<StreamingError>,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Into<StreamingError> + Send + Sync + 'static,
Source§async fn update_metadata(
&self,
cache_key: &str,
headers: &HeaderMap,
policy: CachePolicy,
user_metadata: Option<Vec<u8>>,
token: Option<&CacheEntryToken>,
) -> Result<bool, Box<dyn Error + Send + Sync>>
async fn update_metadata( &self, cache_key: &str, headers: &HeaderMap, policy: CachePolicy, user_metadata: Option<Vec<u8>>, token: Option<&CacheEntryToken>, ) -> Result<bool, Box<dyn Error + Send + Sync>>
Source§async fn convert_body<B>(
&self,
response: Response<B>,
) -> Result<Response<<StreamingManager as StreamingCacheManager>::Body>, Box<dyn Error + Send + Sync>>where
B: Body + Send + 'static,
<B as Body>::Data: Send,
<B as Body>::Error: Into<StreamingError>,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Into<StreamingError> + Send + Sync + 'static,
async fn convert_body<B>(
&self,
response: Response<B>,
) -> Result<Response<<StreamingManager as StreamingCacheManager>::Body>, Box<dyn Error + Send + Sync>>where
B: Body + Send + 'static,
<B as Body>::Data: Send,
<B as Body>::Error: Into<StreamingError>,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Into<StreamingError> + Send + Sync + 'static,
Source§async fn delete(
&self,
cache_key: &str,
) -> Result<(), Box<dyn Error + Send + Sync>>
async fn delete( &self, cache_key: &str, ) -> Result<(), Box<dyn Error + Send + Sync>>
Source§fn empty_body(&self) -> <StreamingManager as StreamingCacheManager>::Body
fn empty_body(&self) -> <StreamingManager as StreamingCacheManager>::Body
Source§fn body_to_bytes_stream(
body: <StreamingManager as StreamingCacheManager>::Body,
) -> impl Stream<Item = Result<Bytes, Box<dyn Error + Send + Sync>>> + Sendwhere
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Send + Sync + 'static,
fn body_to_bytes_stream(
body: <StreamingManager as StreamingCacheManager>::Body,
) -> impl Stream<Item = Result<Bytes, Box<dyn Error + Send + Sync>>> + Sendwhere
<<StreamingManager as StreamingCacheManager>::Body as Body>::Data: Send,
<<StreamingManager as StreamingCacheManager>::Body as Body>::Error: Send + Sync + 'static,
streaming only.Auto Trait Implementations§
impl !RefUnwindSafe for StreamingManager
impl !UnwindSafe for StreamingManager
impl Freeze for StreamingManager
impl Send for StreamingManager
impl Sync for StreamingManager
impl Unpin for StreamingManager
impl UnsafeUnpin for StreamingManager
Blanket Implementations§
Source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
impl<T> BorrowMut<T> for Twhere
T: ?Sized,
Source§fn borrow_mut(&mut self) -> &mut T
fn borrow_mut(&mut self) -> &mut T
impl<ST, DT> CastableFrom<ST, Initialized, Initialized> for DT
impl<ST, DT> CastableFrom<ST, Uninit, Uninit> for DT
Source§impl<T> CloneToUninit for Twhere
T: Clone,
impl<T> CloneToUninit for Twhere
T: Clone,
Source§impl<T> Instrument for T
impl<T> Instrument for T
Source§fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
Source§fn in_current_span(self) -> Instrumented<Self> ⓘ
fn in_current_span(self) -> Instrumented<Self> ⓘ
Source§impl<T> IntoEither for T
impl<T> IntoEither for T
Source§fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
self into a Left variant of Either<Self, Self>
if into_left is true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read moreSource§fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
self into a Left variant of Either<Self, Self>
if into_left(&self) returns true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read more