running_process/broker/
get_http_endpoint_dispatch.rs

1//! `GetBrokerHttpEndpoint` RPC dispatch (slice 6 of #488).
2//!
3//! The CLI calls `GetBrokerHttpEndpoint` over the v2 broker control
4//! channel to discover the broker's HTTP endpoint (per #483 §4 — the
5//! single discovery surface). This module implements the broker side
6//! of that RPC: given the broker's currently-resolved HTTP port + its
7//! own pid, build a `GetBrokerHttpEndpointResponse` and serialize it.
8//!
9//! The real plumbing (read incoming frame → dispatch on payload type →
10//! write response frame) lives in the broker's connection loop, which
11//! is filled in by later slices. This slice exposes the typed
12//! request/response handler so subsequent slices have a pinned API to
13//! call.
14
15use prost::Message;
16
17use crate::broker::protocol_v2::{GetBrokerHttpEndpointRequest, GetBrokerHttpEndpointResponse};
18
19/// In-broker resolved HTTP endpoint state (set at boot per #483 §3 via
20/// `BrokerHttpPort::resolve(config, env)`).
21#[derive(Debug, Clone, Copy)]
22pub struct BrokerHttpEndpoint {
23    /// The port the broker's own HTTP server bound. Slice 7 actually
24    /// binds it; before then the broker can stub this to its
25    /// configured-static port for early consumer testing.
26    pub port: u16,
27    /// The broker's process id. Used by consumers to disambiguate a
28    /// fresh response from a stale one mid-restart (#483 §4 rationale).
29    pub pid: u32,
30}
31
32impl BrokerHttpEndpoint {
33    /// Build a `GetBrokerHttpEndpointResponse` carrying this endpoint.
34    pub fn to_response(self) -> GetBrokerHttpEndpointResponse {
35        GetBrokerHttpEndpointResponse {
36            port: self.port as u32,
37            pid: self.pid,
38        }
39    }
40}
41
42/// Errors from [`decode_request_and_dispatch`].
43#[derive(Debug, thiserror::Error)]
44pub enum GetHttpEndpointError {
45    /// The incoming frame body did not decode as `GetBrokerHttpEndpointRequest`.
46    #[error("decode GetBrokerHttpEndpointRequest: {0}")]
47    Decode(#[from] prost::DecodeError),
48
49    /// Encoding the response failed.
50    #[error("encode GetBrokerHttpEndpointResponse: {0}")]
51    Encode(#[from] prost::EncodeError),
52}
53
54/// Decode an incoming `GetBrokerHttpEndpointRequest` frame body and
55/// produce a serialized `GetBrokerHttpEndpointResponse` body the
56/// connection loop can write back via `protocol::write_frame`.
57///
58/// The request currently has no fields (`GetBrokerHttpEndpointRequest`
59/// is an empty marker per #483 §4) — decoding is purely validation
60/// that the peer sent a structurally well-formed proto message of the
61/// expected type.
62pub fn decode_request_and_dispatch(
63    request_body: &[u8],
64    endpoint: BrokerHttpEndpoint,
65) -> Result<Vec<u8>, GetHttpEndpointError> {
66    let _request = GetBrokerHttpEndpointRequest::decode(request_body)?;
67    let response = endpoint.to_response();
68    let mut body = Vec::with_capacity(response.encoded_len());
69    response.encode(&mut body)?;
70    Ok(body)
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76
77    #[test]
78    fn to_response_carries_port_and_pid() {
79        let resp = BrokerHttpEndpoint {
80            port: 8765,
81            pid: 12_345,
82        }
83        .to_response();
84        assert_eq!(resp.port, 8765);
85        assert_eq!(resp.pid, 12_345);
86    }
87
88    #[test]
89    fn dispatch_round_trip_with_empty_request() {
90        let req = GetBrokerHttpEndpointRequest::default();
91        let mut body = Vec::with_capacity(req.encoded_len());
92        req.encode(&mut body).expect("encode request");
93
94        let resp_body = decode_request_and_dispatch(
95            &body,
96            BrokerHttpEndpoint {
97                port: 4242,
98                pid: 99_999,
99            },
100        )
101        .expect("dispatch succeeds");
102
103        let resp =
104            GetBrokerHttpEndpointResponse::decode(resp_body.as_slice()).expect("decode response");
105        assert_eq!(resp.port, 4242);
106        assert_eq!(resp.pid, 99_999);
107    }
108
109    #[test]
110    fn dispatch_rejects_malformed_request_body() {
111        let err = decode_request_and_dispatch(
112            &[0xFF; 4],
113            BrokerHttpEndpoint {
114                port: 4242,
115                pid: 99_999,
116            },
117        )
118        .expect_err("malformed request body should be rejected");
119        match err {
120            GetHttpEndpointError::Decode(_) => {}
121            other => panic!("expected Decode error, got: {other:?}"),
122        }
123    }
124}
running_process/broker/get_http_endpoint_dispatch.rs

running_process/broker/
get_http_endpoint_dispatch.rs
