Struct ClusterDispatcher 

pub struct ClusterDispatcher { /* private fields */ }

Cluster-aware dispatcher.

Implementations

impl ClusterDispatcher

pub fn new(pool: Arc<ServerPool>) -> Self

Wrap a ServerPool in a dispatcher.

Examples

let pool = Arc::new(ServerPool::new(cfg, vec![local]));
let disp = ClusterDispatcher::new(pool);
let _ = disp.pool();

pub fn with_mbuf_pool(self, pool: MbufPool) -> Self

Override the dispatcher’s mbuf pool. Useful when the embedding wants every synthetic error payload to come from the same recycled buffers as the rest of the engine.

Examples

let pool = Arc::new(ServerPool::new(cfg, vec![local]));
let _disp = ClusterDispatcher::new(pool).with_mbuf_pool(MbufPool::default());

pub fn mbuf_pool(&self) -> &MbufPool

Borrow the dispatcher’s mbuf pool. Exposed so embedders can reuse the same pool when building synthetic responses outside the dispatcher’s own code paths.

pub fn with_backend(self, backend: Sender<OutboundRequest>) -> Self

Attach a backend request channel. Calls to Self::dispatch that produce a DispatchPlan::LocalDatastore plan will forward the request bytes onto this channel for the local datastore driver to write to the backend.

The channel sender must be the request side of a crate::net::ServerConn task; multiple senders cloned from the same channel are fine.

pub fn with_peer_backend( self, peer_idx: u32, sender: Sender<OutboundRequest>, ) -> Self

Attach an outbound channel for a single peer (by Peer::idx). The supplied sender feeds a crate::net::DnodeServerConn task that writes dnode-framed requests to the peer’s dyn_listen and routes the response back through the per-request responder channel.

Wiring is additive: call this once per non-local peer. Calling it again with the same peer_idx replaces the previous sender (used by reconnect supervisors that rebuild channels on restart).

pub fn has_backend(&self) -> bool

Whether a backend channel is wired.

pub fn peer_backend_count(&self) -> usize

Number of peer-backend channels wired.

pub fn pool(&self) -> &Arc<ServerPool>

Borrow the underlying pool.

pub fn with_hint_store(self, store: Arc<HintStore>) -> Self

Attach a crate::cluster::hints::HintStore.

When set AND the pool’s enable_hinted_handoff flag is true, write requests targeted at peers in crate::cluster::peer::PeerState::Down (or at peers whose outbound channel is closed / full) are stored as hints and counted toward the consistency threshold. The background drainer task in dynomited is responsible for shipping the hints back to the peer once it returns to crate::cluster::peer::PeerState::Normal. Without this builder call (or with enable_hinted_handoff: false) the dispatcher behaviour is unchanged: a Down or unreachable target is silently skipped.

Examples

let cfg = PoolConfig { enable_hinted_handoff: true, ..PoolConfig::default() };
let local = Peer::new(
    0, PeerEndpoint::tcp("h".into(), 1), "r".into(), "d".into(),
    vec![DynToken::from_u32(0)], true, true, false,
);
let pool = Arc::new(ServerPool::new(cfg, vec![local]));
let store = Arc::new(HintStore::new(64 * 1024 * 1024));
let _disp = ClusterDispatcher::new(pool).with_hint_store(store);

pub fn hint_store(&self) -> Option<&Arc<HintStore>>

Borrow the wired hint store, if any.

pub fn with_failure_metrics(self, metrics: Arc<FailureMetrics>) -> Self

Attach a crate::stats::FailureMetrics handle.

When wired, each error-producing branch in the dispatcher (no-targets, peer-channel-full, peer-channel-closed, backend-channel-full, backend-channel-closed, response-timeout) increments the matching counter so an operator can pull the per-cause histogram off the /stats and /metrics endpoints. The default behaviour is unchanged when no metrics handle is supplied.

Examples

let cfg = PoolConfig::default();
let local = Peer::new(
    0, PeerEndpoint::tcp("h".into(), 1), "r".into(), "d".into(),
    vec![DynToken::from_u32(0)], true, true, false,
);
let pool = Arc::new(ServerPool::new(cfg, vec![local]));
let metrics = Arc::new(FailureMetrics::new());
let _disp = ClusterDispatcher::new(pool).with_failure_metrics(metrics);

pub fn failure_metrics(&self) -> Option<&Arc<FailureMetrics>>

Borrow the wired failure-metrics handle, if any.

pub fn with_vector_registry(self, registry: Arc<VectorRegistry>) -> Self

Attach a crate::vector::registry::VectorRegistry.

When wired, parsed RediSearch FT.* requests (FT.CREATE / FT.SEARCH / FT.INFO / FT.LIST / FT.DROPINDEX) short-circuit to the registry-backed executor in crate::proto::redis::ft and the synthesised RESP bytes are returned to the client as crate::net::DispatchOutcome::Inline. HSET requests whose key matches a registered prefix are routed through crate::proto::redis::ft::maybe_index_hset before they fall through to the standard backend path; a malformed or missing vector field surfaces as crate::net::DispatchOutcome::Error with a -ERR ... reply rather than reaching the backend.

Without this builder call, the dispatcher’s behaviour is unchanged: FT.* keywords are forwarded to the local datastore (which typically rejects them with -ERR unknown command).

Examples

let cfg = PoolConfig::default();
let local = Peer::new(
    0, PeerEndpoint::tcp("h".into(), 1), "r".into(), "d".into(),
    vec![DynToken::from_u32(0)], true, true, false,
);
let pool = Arc::new(ServerPool::new(cfg, vec![local]));
let registry = Arc::new(VectorRegistry::new());
let _disp = ClusterDispatcher::new(pool).with_vector_registry(registry);

pub fn vector_registry(&self) -> Option<&Arc<VectorRegistry>>

Borrow the wired vector-index registry, if any.

pub fn hinted_handoff_active(&self) -> bool

True when both the hint store is wired AND the pool has enable_hinted_handoff: true. Hot-path predicate.

pub fn plan(&self, req: &Msg, key: &[u8]) -> DispatchPlan

Compute the routing plan for req with the supplied key.

key is the primary key of the request (the first key returned by Msg::keys for parsed redis / memcache commands, or an empty slice for argument-less commands).

The function never panics; it consults the live peer table behind the pool’s RwLock and returns DispatchPlan::NoTargets when the topology cannot satisfy the request.

Examples

See the module-level example.

Trait Implementations

impl Clone for ClusterDispatcher

fn clone(&self) -> ClusterDispatcher

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ClusterDispatcher

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Dispatcher for ClusterDispatcher

fn dispatch(&self, req: Msg, responder: ServerSink) -> DispatchOutcome

Hand a parsed request to the dispatcher.