Struct HeaderMap 

pub struct HeaderMap { /* private fields */ }

Case-insensitive map of broker-message headers.

Keys are normalized to ASCII lowercase on insertion. Both halves of an entry are shared buffers, Str and Bytes, so cloning a map is a reference count per entry and a broker hands over a key its own read buffer already holds. Values are bytes to support arbitrary binary metadata. Typed accessors are provided for well-known fields commonly carried by message brokers; unknown headers are read through HeaderMap::get, or through HeaderMap::get_shared where the value is to outlive the borrow.

Two maps are equal when they hold the same entries, whatever order the entries were inserted in. A lookup scans the entries, so its cost grows linearly with the number of headers; a broker’s frame limit bounds that number, and over the handful of headers a message carries the scan is cheaper than hashing the name.

Examples

use ruststream::prelude::*;
use serde::Deserialize;

/// The tenant a delivery belongs to. Header names compare without regard to case, so the
/// producer may have written `X-Tenant-Id`.
struct Tenant(String);

impl<C: Send, S: Sync> FromContext<C, S> for Tenant {
    type Rejection = HandlerOutcome;
    async fn from_context(ctx: &mut Context<'_, C, S>) -> Result<Self, HandlerOutcome> {
        match ctx.headers().get_str("x-tenant-id") {
            Some(tenant) => Ok(Tenant(tenant.to_owned())),
            None => Err(HandlerOutcome::drop()),
        }
    }
}

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber("orders")]
async fn bill(order: &Order, Tenant(tenant): Tenant) -> HandlerOutcome {
    tracing::info!(order.id, tenant, "billed");
    HandlerOutcome::ack()
}

Implementations

impl HeaderMap

pub const fn new() -> HeaderMap

Returns an empty header map.

pub fn with_capacity(cap: usize) -> HeaderMap

Returns an empty header map with capacity for at least cap entries.

pub fn insert( &mut self, name: impl Into<StrInner<Bytes>>, value: impl Into<Bytes>, ) -> Option<Bytes>

Inserts a header value, returning the previous value under that key if any.

The key is normalized to ASCII lowercase. A constant key is written Str::from_static("content-type"), which costs nothing; a String moves in without a copy; a &str is copied once, and a key that is not already lowercase is copied again.

Examples

use ruststream::memory::prelude::*;
use ruststream::runtime::{Outgoing, SlotContext};
use serde::{Deserialize, Serialize};

/// Stamps everything a slot sends with the region this instance runs in.
struct Region(&'static str);

impl<Options> PublishTransform<ForSlot, Options> for Region {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        _cx: &SlotContext<'_>,
    ) {
        out.headers_mut()
            .insert(Str::from_static("x-region"), self.0);
    }
}

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "orders.accepted")]
struct Accepted {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Accepted)]
struct Events;

#[subscriber("orders")]
async fn accept(order: &Order, Out(events): Out<impl Publisher, Events>) -> HandlerOutcome {
    match events.message(&Accepted { id: order.id }).publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept)
            .out(Events, Publish)
            .transform(Region("eu-west"))
            .build();
    })
}

pub fn get(&self, name: &str) -> Option<&[u8]>

Returns the raw bytes of a header value, or None if the header is absent.

pub fn get_shared(&self, name: &str) -> Option<Bytes>

Returns a header value as the shared buffer the map stores, or None if the header is absent.

get borrows the bytes for as long as the map lives; this hands over a counted handle on them instead, which outlives the borrow and costs a reference count rather than a copy. It is how a value becomes something that travels: a reply address read off a delivery becomes an outgoing destination through Str::try_from, a UTF-8 check over the same bytes that yields a Str.

Examples

use ruststream::memory::prelude::*;
use ruststream::runtime::{Outgoing, PublishContext};
use serde::{Deserialize, Serialize};

/// Sends each reply to the inbox its request named. The header's bytes become the
/// destination as they are: one check that they are UTF-8, no copy.
struct ToInbox;

impl<C, Options> PublishTransform<ForReply<C>, Options> for ToInbox {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(inbox) = cx.headers().get_shared("reply-to") {
            if let Ok(name) = Str::try_from(inbox) {
                out.set_name(name);
            }
        }
    }
}

#[derive(Deserialize)]
struct Request {
    id: u64,
}

#[derive(Serialize, Outgoing)]
struct Answer {
    id: u64,
}

#[subscriber("requests", reply("answers"))]
async fn answer(request: &Request) -> Answer {
    Answer { id: request.id }
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("answers", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(answer).out_reply(Publish).transform(ToInbox);
        },
    )
}

pub fn get_str(&self, name: &str) -> Option<&str>

Returns the value of a header decoded as UTF-8, or None if absent or not valid UTF-8.

pub fn remove(&mut self, name: &str) -> Option<Bytes>

Removes a header by name and returns its previous value, if any.

pub fn contains(&self, name: &str) -> bool

Returns true if the given header is present.

pub fn len(&self) -> usize

Returns the number of headers.

pub fn is_empty(&self) -> bool

Returns true if no headers are present.

pub fn iter(&self) -> impl Iterator<Item = (&str, &[u8])>

Iterates over (name, value) pairs. Names are returned in their normalized lowercase form.

pub fn content_type(&self) -> Option<&str>

Returns the value of the content-type header decoded as UTF-8.

pub fn correlation_id(&self) -> Option<&str>

Returns the value of the correlation-id header decoded as UTF-8.

pub fn reply_to(&self) -> Option<&str>

Returns the value of the reply-to header decoded as UTF-8.

pub fn message_id(&self) -> Option<&str>

Returns the value of the message-id header decoded as UTF-8.

impl HeaderMap

pub fn insert_typed<T>( &mut self, value: &T, ) -> Result<(), SerializeHeadersError>

where T: Serialize + ?Sized,

Serializes a typed contract into the header map, one entry per field.

The mirror of the crate-internal to_typed parse: every field of the flat struct (or string-keyed map) becomes a header named after the field, with the value string-encoded (numbers and booleans via their display form, unit-only enums as the variant name, byte fields as the raw value). An Option field that is None inserts nothing. Existing headers under other names are kept; a colliding name is overwritten. Header names normalize to lowercase on insertion, and #[serde(flatten)] is rejected (it could not be parsed back).

Errors

Returns SerializeHeadersError when the value is not a struct/map or a field holds a non-scalar (nested struct, sequence, map, data-carrying enum variant).

Examples

use ruststream::memory::prelude::*;
use ruststream::runtime::{Outgoing, SlotContext};
use serde::{Deserialize, Serialize};

/// The deployment every message of the slot came from, as a header contract.
#[derive(Serialize)]
struct Origin {
    region: &'static str,
    canary: bool,
}

/// Stamps the contract onto everything the slot sends.
struct Stamp(Origin);

impl<Options> PublishTransform<ForSlot, Options> for Stamp {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        _cx: &SlotContext<'_>,
    ) {
        if let Err(error) = out.headers_mut().insert_typed(&self.0) {
            tracing::warn!(%error, "origin headers left out");
        }
    }
}

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "orders.accepted")]
struct Accepted {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Accepted)]
struct Events;

#[subscriber("orders")]
async fn accept(order: &Order, Out(events): Out<impl Publisher, Events>) -> HandlerOutcome {
    match events.message(&Accepted { id: order.id }).publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

fn app() -> RustStream {
    let origin = Origin {
        region: "eu-west",
        canary: false,
    };
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept)
            .out(Events, Publish)
            .transform(Stamp(origin))
            .build();
    })
}

Trait Implementations

impl Clone for HeaderMap

fn clone(&self) -> HeaderMap

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for HeaderMap

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

Formats the value using the given formatter.

impl Default for HeaderMap

fn default() -> HeaderMap

Returns the “default value” for a type.

impl Eq for HeaderMap

impl<K, V> FromIterator<(K, V)> for HeaderMap

where K: Into<StrInner<Bytes>>, V: Into<Bytes>,

fn from_iter<I>(iter: I) -> HeaderMap

where I: IntoIterator<Item = (K, V)>,

Creates a value from an iterator.

impl HeaderSource for HeaderMap

type Position = MapHeaders

The headers position this source resolves to: TypedHeaders or MapHeaders.

fn into_position(self) -> <HeaderMap as HeaderSource>::Position

Moves the source into its position.

impl PartialEq for HeaderMap

Equal when both hold the same entries, in whatever order they were inserted.

fn eq(&self, other: &HeaderMap) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.