Struct FleetFeed 

pub struct FleetFeed {
    epoch: String,
    hostname: String,
    channel: String,
    version: String,
    os: String,
    install_id: Option<String>,
    tx: Sender<FleetSnapshot>,
    viewer_port: Arc<AtomicU16>,
    streams: Arc<AtomicUsize>,
    heartbeat: Duration,
}

Fields

epoch: String

16 lowercase hex chars, new on every srv launch: a client that sees a different epoch knows the instance restarted and takes the snapshot as-is rather than comparing revs.

hostname: String

channel: String

version: String

os: String

install_id: Option<String>

This install’s WAN instance id; None without a WAN identity store.

tx: Sender<FleetSnapshot>

viewer_port: Arc<AtomicU16>

Where the viewer port is read from on each look; 0 = not up.

streams: Arc<AtomicUsize>

heartbeat: Duration

Implementations

impl FleetFeed

pub fn new(hostname: String, channel: String, version: String) -> Self

pub fn with_viewer_port(self, port: Arc<AtomicU16>) -> Self

Read the viewer port from port (ViewerService::advertised_port_handle).

pub fn with_install_id(self, install_id: Option<String>) -> Self

Set the install id the body reports (fleet_source::install_id).

pub fn hostname(&self) -> &str

pub fn channel(&self) -> &str

pub fn version(&self) -> &str

pub fn os(&self) -> &str

pub fn snapshot(&self) -> FleetSnapshot

pub fn subscribe(&self) -> Receiver<FleetSnapshot>

A receiver that wakes on every change, for the presence publisher.

pub fn event_id(&self, snapshot: &FleetSnapshot) -> String

<epoch>:<rev>: the SSE id, and (quoted) the snapshot’s ETag.

pub fn body_json(&self, snapshot: &FleetSnapshot) -> String

The /agentmux/fleet body, on one line, with now_ms read as it is served, so a peer can tell how long a state has held without trusting its own clock against this one. now_ms is not part of the snapshot: it never bumps rev or changes the ETag.

fn body_json_at(&self, snapshot: &FleetSnapshot, now_ms: u64) -> String

pub fn observe_fleet(&self, observation: FleetObservation) -> bool

Record the current fleet; bumps rev and wakes every stream only if it differs from the last one (names, kinds, channel count, any agent’s state or line, or the viewer port). Returns whether it did.

pub fn spawn_change_detection<F>( self: Arc<Self>, observe: F, token: CancellationToken, )

where F: Fn() -> FleetObservation + Send + Sync + 'static,

Start the change-detection task: observe is read now, then every CHANGE_POLL_INTERVAL on a blocking thread (it reads the store and the shared registry) until token is cancelled (srv shutdown), the same stop signal the WAL checkpoint loop watches.

pub fn open_stream( self: &Arc<Self>, last_event_id: Option<&str>, ) -> Option<impl Stream<Item = Result<Bytes, Infallible>> + Send + 'static>

Open one event stream, or None when MAX_STREAMS are already open.

The slot is held by a guard moved into the stream, so it is released whenever the stream is dropped: when hyper drops the response body after the client goes away (noticed at the latest on the next heartbeat write), or if the response is never sent at all.

With last_event_id equal to the current <epoch>:<rev> the client is already up to date and gets no initial event, only later ones.

fn event_bytes(&self, snapshot: &FleetSnapshot) -> Bytes

One fleet event. body_json is serde output, so it holds no raw newline that could break the data: line.