Struct MyProxy 

pub struct MyProxy {

    pub(crate) cos_endpoint: String,
    pub(crate) cos_mapping: Arc<RwLock<HashMap<String, CosMapItem>>>,
    pub(crate) hmac_keystore: Arc<RwLock<HashMap<String, String>>>,
    pub(crate) secrets_cache: SecretsCache,
    pub(crate) auth_cache: AuthCache,
    pub(crate) validator: Option<PyObject>,
    pub(crate) bucket_creds_fetcher: Option<PyObject>,
    pub(crate) verify: Option<bool>,
    pub(crate) skip_signature_validation: Option<bool>,
    pub(crate) hmac_fetcher: Option<PyObject>,
    pub(crate) tracker: UrlTracker,
    pub(crate) max_presign_url_usage_attempts: Option<usize>,
    pub(crate) server_name: String,
}

The core Pingora proxy handler.

One instance is created per server and shared (via Arc) across all worker threads. It implements [ProxyHttp] and drives the full request lifecycle: signature validation → authorization → credential injection → upstream routing.

Fields

cos_endpoint: String

cos_mapping: Arc<RwLock<HashMap<String, CosMapItem>>>

hmac_keystore: Arc<RwLock<HashMap<String, String>>>

secrets_cache: SecretsCache

auth_cache: AuthCache

validator: Option<PyObject>

bucket_creds_fetcher: Option<PyObject>

verify: Option<bool>

skip_signature_validation: Option<bool>

hmac_fetcher: Option<PyObject>

tracker: UrlTracker

max_presign_url_usage_attempts: Option<usize>

server_name: String

Trait Implementations

impl ProxyHttp for MyProxy

type CTX = MyCtx

The per request object to share state across the different filters

fn new_ctx(&self) -> Self::CTX

Define how the ctx should be created.

fn upstream_peer<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, session: &'life1 mut Session, ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<Box<HttpPeer>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Define where the proxy should send the request to.

fn request_filter<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, session: &'life1 mut Session, ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Handle the incoming request.

fn upstream_request_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, upstream_request: &'life2 mut RequestHeader, ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Modify the request before it is sent to the upstream

fn response_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, resp: &'life2 mut ResponseHeader, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Modify the response header before it is send to the downstream

fn request_body_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, body: &'life2 mut Option<Bytes>, end_of_stream: bool, ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Handle the incoming request body.

fn init_downstream_modules(&self, modules: &mut HttpModules)

Set up downstream modules.

fn early_request_filter<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<(), Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Handle the incoming request before any downstream module is executed.

fn allow_spawning_subrequest( &self, _session: &Session, _ctx: &Self::CTX, ) -> bool

where Self::CTX: Send + Sync,

Returns whether this session is allowed to spawn subrequests.

fn request_cache_filter( &self, _session: &mut Session, _ctx: &mut Self::CTX, ) -> Result<(), Box<Error>>

where Self::CTX: Send + Sync,

This filter decides if the request is cacheable and what cache backend to use

fn cache_key_callback( &self, _session: &Session, _ctx: &mut Self::CTX, ) -> Result<CacheKey, Box<Error>>

This callback generates the cache key.

fn cache_miss(&self, session: &mut Session, _ctx: &mut Self::CTX)

This callback is invoked when a cacheable response is ready to be admitted to cache.

fn cache_hit_filter<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _meta: &'life2 CacheMeta, _hit_handler: &'life3 mut Box<dyn HandleHit + Sync + Send>, _is_fresh: bool, _ctx: &'life4 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<Option<ForcedFreshness>, Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

This filter is called after a successful cache lookup and before the cache asset is ready to be used.

fn proxy_upstream_filter<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<bool, Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Decide if a request should continue to upstream after not being served from cache.

fn response_cache_filter( &self, _session: &Session, _resp: &ResponseHeader, _ctx: &mut Self::CTX, ) -> Result<RespCacheable, Box<Error>>

Decide if the response is cacheable

fn cache_vary_filter( &self, _meta: &CacheMeta, _ctx: &mut Self::CTX, _req: &RequestHeader, ) -> Option<[u8; 16]>

Decide how to generate cache vary key from both request and response

fn cache_not_modified_filter( &self, session: &Session, resp: &ResponseHeader, _ctx: &mut Self::CTX, ) -> Result<bool, Box<Error>>

Decide if the incoming request’s condition fails against the cached response.

fn range_header_filter( &self, session: &mut Session, resp: &mut ResponseHeader, _ctx: &mut Self::CTX, ) -> RangeType

This filter is called when cache is enabled to determine what byte range to return (in both cache hit and miss cases) from the response body. It is only used when caching is enabled, otherwise the upstream is responsible for any filtering. It allows users to define the range this request is for via its return type range_filter::RangeType.

fn upstream_response_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _upstream_response: &'life2 mut ResponseHeader, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<(), Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Modify the response header from the upstream

fn upstream_response_body_filter( &self, _session: &mut Session, _body: &mut Option<Bytes>, _end_of_stream: bool, _ctx: &mut Self::CTX, ) -> Result<Option<Duration>, Box<Error>>

Similar to [Self::upstream_response_filter()] but for response body

fn upstream_response_trailer_filter( &self, _session: &mut Session, _upstream_trailers: &mut HeaderMap, _ctx: &mut Self::CTX, ) -> Result<(), Box<Error>>

Similar to [Self::upstream_response_filter()] but for response trailers

fn response_body_filter( &self, _session: &mut Session, _body: &mut Option<Bytes>, _end_of_stream: bool, _ctx: &mut Self::CTX, ) -> Result<Option<Duration>, Box<Error>>

where Self::CTX: Send + Sync,

Similar to [Self::response_filter()] but for response body chunks

fn response_trailer_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _upstream_trailers: &'life2 mut HeaderMap, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Similar to [Self::response_filter()] but for response trailers. Note, returning an Ok(Some(Bytes)) will result in the downstream response trailers being written to the response body.

fn logging<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _e: Option<&'life2 Error>, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

This filter is called when the entire response is sent to the downstream successfully or there is a fatal error that terminate the request.

fn suppress_error_log( &self, _session: &Session, _ctx: &Self::CTX, _error: &Error, ) -> bool

A value of true means that the log message will be suppressed. The default value is false.

fn error_while_proxy( &self, peer: &HttpPeer, session: &mut Session, e: Box<Error>, _ctx: &mut Self::CTX, client_reused: bool, ) -> Box<Error>

This filter is called when there is an error after a connection is established (or reused) to the upstream.

fn fail_to_connect( &self, _session: &mut Session, _peer: &HttpPeer, _ctx: &mut Self::CTX, e: Box<Error>, ) -> Box<Error>

This filter is called when there is an error in the process of establishing a connection to the upstream.

fn fail_to_proxy<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, session: &'life1 mut Session, e: &'life2 Error, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = FailToProxy> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

This filter is called when the request encounters a fatal error.

fn should_serve_stale( &self, _session: &mut Session, _ctx: &mut Self::CTX, error: Option<&Error>, ) -> bool

Decide whether should serve stale when encountering an error or during revalidation

fn connected_to_upstream<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _reused: bool, _peer: &'life2 HttpPeer, _fd: i32, _digest: Option<&'life3 Digest>, _ctx: &'life4 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<(), Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

This filter is called when the request just established or reused a connection to the upstream

fn request_summary(&self, session: &Session, _ctx: &Self::CTX) -> String

This callback is invoked every time request related error log needs to be generated

fn is_purge(&self, _session: &Session, _ctx: &Self::CTX) -> bool

Whether the request should be used to invalidate(delete) the HTTP cache

fn purge_response_filter( &self, _session: &Session, _ctx: &mut Self::CTX, _purge_status: PurgeStatus, _purge_response: &mut Cow<'static, ResponseHeader>, ) -> Result<(), Box<Error>>

This filter is called after the proxy cache generates the downstream response to the purge request (to invalidate or delete from the HTTP cache), based on the purge status, which indicates whether the request succeeded or failed.