liminal/protocol/frame.rs
1use super::{
2 causal::MessageId,
3 envelope::{MessageEnvelope, SchemaId},
4 error::ProtocolError,
5 version::ProtocolVersion,
6};
7
8/// Number of bytes in every serialized frame header.
9pub const HEADER_LEN: usize = 10;
10
11/// Frame-flag bit set on a [`Frame::ConversationMessage`] to request a correlated
12/// reply.
13///
14/// A client sets this bit on the request frame of a request-reply round trip. The
15/// server, after delivering the message to the conversation participant, drains
16/// the participant's reply and sends it back as a `ConversationMessage` carrying
17/// the same `conversation_id` (the correlation key) and this same flag bit. A
18/// `ConversationMessage` WITHOUT this bit keeps the pre-existing fire-and-forget
19/// semantics: the server stays silent on success. The bit travels in the frame
20/// header's `flags` byte, which the codec already round-trips, so no wire-format
21/// change is required.
22pub const CONVERSATION_REPLY_REQUESTED_FLAG: u8 = 0x01;
23
24/// Frame-flag bit set on a [`Frame::Publish`] to declare that the frame body
25/// carries a trailing idempotency-key string field (the dedup-on-delivery key).
26///
27/// A publisher sets this bit when it wants the server to consult its dedup cache
28/// keyed by the trailing idempotency key, delivering the message to subscribers
29/// AT MOST ONCE across re-publishes of the same key. A `Publish` frame WITHOUT
30/// this bit keeps the pre-existing wire layout EXACTLY: no trailing field is
31/// written and none is read, so a no-key publish is byte-identical to before.
32/// The bit travels in the frame header's `flags` byte, which the codec already
33/// round-trips, so no header-format change is required (the 13-L0 precedent).
34pub const PUBLISH_IDEMPOTENCY_KEY_FLAG: u8 = 0x02;
35
36/// Frame-flag bit set on a [`Frame::PublishAck`] to report a GENUINE delivery
37/// ack: the published message was accepted by at least one live subscriber on
38/// this publish.
39///
40/// This is distinct from the backpressure `Accept`/`Defer`/`Reject` signal: a
41/// `PublishAck` always means the publish was processed without error, but only a
42/// `PublishAck` carrying this bit means a subscriber actually received the
43/// message. An ack WITHOUT this bit means the publish succeeded but reached no
44/// subscriber (an empty channel, or a duplicate suppressed by dedup-on-delivery),
45/// so a caller that needs a true delivery ack can observe the difference. The bit
46/// rides the existing `flags` byte, so no wire-format change is required.
47pub const PUBLISH_DELIVERED_FLAG: u8 = 0x01;
48
49/// Frame-flag bit set on a [`Frame::PublishAck`] to declare that the frame body
50/// carries a trailing durable-position field after `message_id`.
51///
52/// The durable position is the store-assigned position of the row the publish
53/// answers for, written as a `u64` big-endian (the width and byte order of
54/// `message_id`). An ack WITHOUT this bit keeps the older layout EXACTLY: no
55/// trailing field is written and none is read, so an ack that carries no
56/// position is byte-identical to before (the [`PUBLISH_IDEMPOTENCY_KEY_FLAG`]
57/// precedent). The codec sets or clears the bit from the frame's
58/// `durable_position`, so a caller never sets it by hand.
59pub const PUBLISH_DURABLE_POSITION_FLAG: u8 = 0x02;
60
61/// Protocol version that carries the [`Frame::PublishAck`] durable position.
62///
63/// A server writes an ack carrying a position only on a connection that
64/// negotiated this version or later; a connection at 1.0 always receives the
65/// older ack layout.
66pub const PUBLISH_ACK_POSITION_VERSION: ProtocolVersion = ProtocolVersion::new(1, 1);
67
68/// Frame-flag bit set on a [`Frame::Deliver`] to declare that the frame body
69/// carries a trailing [`DeliveredId`] after the envelope.
70///
71/// The pair is written as two `u64`s big-endian, the boot then the id. A
72/// Deliver WITHOUT this bit keeps the older layout EXACTLY: nothing trailing is
73/// written and nothing is read, so a 1.0 or 1.1 peer receives the bytes it
74/// always did. The codec sets or clears the bit from the frame's `message`, so a
75/// caller never sets it by hand (the [`PUBLISH_DURABLE_POSITION_FLAG`]
76/// precedent).
77pub const DELIVER_MESSAGE_ID_FLAG: u8 = 0x01;
78
79/// Protocol version that carries the [`Frame::Deliver`] message id.
80///
81/// A server writes a Deliver carrying a [`DeliveredId`] only on a connection
82/// that negotiated this version or later; a connection at 1.0 or 1.1 always
83/// receives the older Deliver layout.
84pub const DELIVER_MESSAGE_ID_VERSION: ProtocolVersion = ProtocolVersion::new(1, 2);
85
86/// The server's id for a delivered message, with the boot it was drawn in.
87///
88/// `id` is the number the publisher's `PublishAck` carried and the console's
89/// tap id names; it is drawn from a counter that restarts on every boot, so
90/// `boot` (the server incarnation's start, in wall-clock milliseconds since
91/// the epoch) says which run of the server the number belongs to. A reply that
92/// names a request does so by both, and a tap pairs it only within its own boot.
93#[derive(Clone, Copy, Debug, PartialEq, Eq)]
94pub struct DeliveredId {
95 /// The server incarnation's start, in wall-clock milliseconds since the epoch.
96 pub boot: u64,
97 /// The server's message id within that boot.
98 pub id: u64,
99}
100
101impl DeliveredId {
102 /// The causal parent a reply to this message names: `<boot>:<id>`, both in
103 /// decimal. It is the tap id's number with the boot in place of the device
104 /// (Hermes 7 Oct 08:12), so a tap pairs a reply only within its own boot.
105 #[must_use]
106 pub fn reply_parent(self) -> MessageId {
107 MessageId::new(format!("{}:{}", self.boot, self.id))
108 }
109
110 /// The message a reply's parent names, when the parent is exactly the form
111 /// [`DeliveredId::reply_parent`] writes; any other form names none. Reads
112 /// without allocating, because a tap reads it once per reply.
113 #[must_use]
114 pub fn from_parent(parent: &MessageId) -> Option<Self> {
115 let (boot, id) = parent.as_str().split_once(':')?;
116 Some(Self {
117 boot: canonical_decimal(boot)?,
118 id: canonical_decimal(id)?,
119 })
120 }
121}
122
123/// `digits` as a `u64` when it is the decimal [`u64`]'s `Display` writes: one
124/// or more ASCII digits, no sign, no leading zero, within range.
125fn canonical_decimal(digits: &str) -> Option<u64> {
126 let bytes = digits.as_bytes();
127 let canonical = !bytes.is_empty()
128 && bytes.iter().all(u8::is_ascii_digit)
129 && (bytes.len() == 1 || bytes.first() != Some(&b'0'));
130 if canonical { digits.parse().ok() } else { None }
131}
132
133/// Status byte prefixing a [`Frame::WorkerRegisterAck`] payload that signals the
134/// registration was accepted (no further payload follows).
135pub(crate) const WORKER_REGISTER_ACK_ACCEPTED: u8 = 0x00;
136
137/// Status byte prefixing a [`Frame::WorkerRegisterAck`] payload that signals the
138/// registration was rejected (a length-prefixed reason string follows).
139pub(crate) const WORKER_REGISTER_ACK_REJECTED: u8 = 0x01;
140
141/// Protocol frame categories and their stable wire discriminants.
142#[derive(Clone, Copy, Debug, PartialEq, Eq)]
143pub enum FrameType {
144 /// Connection request.
145 Connect,
146 /// Successful connection response.
147 ConnectAck,
148 /// Failed connection response.
149 ConnectError,
150 /// Connection close notification.
151 Disconnect,
152 /// Channel subscription request.
153 Subscribe,
154 /// Successful subscription response.
155 SubscribeAck,
156 /// Failed subscription response.
157 SubscribeError,
158 /// Channel unsubscription request.
159 Unsubscribe,
160 /// Channel publish request.
161 Publish,
162 /// Successful publish response.
163 PublishAck,
164 /// Failed publish response.
165 PublishError,
166 /// Conversation lifecycle open.
167 ConversationOpen,
168 /// Conversation message delivery.
169 ConversationMessage,
170 /// Conversation lifecycle close.
171 ConversationClose,
172 /// Conversation processing error.
173 ConversationError,
174 /// In-band backpressure acceptance.
175 Accept,
176 /// In-band backpressure deferral.
177 Defer,
178 /// In-band backpressure rejection.
179 Reject,
180 /// Connection keepalive ping.
181 Ping,
182 /// Connection keepalive pong.
183 Pong,
184 /// Server-initiated push of an opaque payload to a connected client.
185 Push,
186 /// Client-initiated correlated reply to a server push.
187 PushReply,
188 /// Worker self-registration announcing identity and routing dimensions.
189 WorkerRegister,
190 /// Server acknowledgement of a worker registration (accepted or rejected).
191 WorkerRegisterAck,
192 /// Server-to-client delivery of a subscribed channel message.
193 Deliver,
194 /// Forward-compatible frame type not known to this implementation.
195 Unknown(u8),
196}
197
198impl FrameType {
199 /// Return true when this frame type must appear on stream 0.
200 #[must_use]
201 pub const fn is_control(self) -> bool {
202 matches!(
203 self,
204 Self::Connect
205 | Self::ConnectAck
206 | Self::ConnectError
207 | Self::Disconnect
208 | Self::Ping
209 | Self::Pong
210 | Self::WorkerRegister
211 | Self::WorkerRegisterAck
212 )
213 }
214}
215
216impl From<u8> for FrameType {
217 fn from(value: u8) -> Self {
218 match value {
219 0x01 => Self::Connect,
220 0x02 => Self::ConnectAck,
221 0x03 => Self::ConnectError,
222 0x04 => Self::Disconnect,
223 0x05 => Self::Subscribe,
224 0x06 => Self::SubscribeAck,
225 0x07 => Self::SubscribeError,
226 0x08 => Self::Unsubscribe,
227 0x09 => Self::Publish,
228 0x0A => Self::PublishAck,
229 0x0B => Self::PublishError,
230 0x0C => Self::ConversationOpen,
231 0x0D => Self::ConversationMessage,
232 0x0E => Self::ConversationClose,
233 0x0F => Self::ConversationError,
234 0x10 => Self::Accept,
235 0x11 => Self::Defer,
236 0x12 => Self::Reject,
237 0x13 => Self::Ping,
238 0x14 => Self::Pong,
239 0x15 => Self::Push,
240 0x16 => Self::PushReply,
241 0x17 => Self::WorkerRegister,
242 0x18 => Self::WorkerRegisterAck,
243 0x19 => Self::Deliver,
244 unknown => Self::Unknown(unknown),
245 }
246 }
247}
248
249impl From<FrameType> for u8 {
250 fn from(value: FrameType) -> Self {
251 match value {
252 FrameType::Connect => 0x01,
253 FrameType::ConnectAck => 0x02,
254 FrameType::ConnectError => 0x03,
255 FrameType::Disconnect => 0x04,
256 FrameType::Subscribe => 0x05,
257 FrameType::SubscribeAck => 0x06,
258 FrameType::SubscribeError => 0x07,
259 FrameType::Unsubscribe => 0x08,
260 FrameType::Publish => 0x09,
261 FrameType::PublishAck => 0x0A,
262 FrameType::PublishError => 0x0B,
263 FrameType::ConversationOpen => 0x0C,
264 FrameType::ConversationMessage => 0x0D,
265 FrameType::ConversationClose => 0x0E,
266 FrameType::ConversationError => 0x0F,
267 FrameType::Accept => 0x10,
268 FrameType::Defer => 0x11,
269 FrameType::Reject => 0x12,
270 FrameType::Ping => 0x13,
271 FrameType::Pong => 0x14,
272 FrameType::Push => 0x15,
273 FrameType::PushReply => 0x16,
274 FrameType::WorkerRegister => 0x17,
275 FrameType::WorkerRegisterAck => 0x18,
276 FrameType::Deliver => 0x19,
277 FrameType::Unknown(type_id) => type_id,
278 }
279 }
280}
281
282/// Fixed-size frame prefix: type, flags, stream identifier, and payload length.
283#[derive(Clone, Copy, Debug, PartialEq, Eq)]
284pub struct FrameHeader {
285 /// Frame type read from or written to byte 0.
286 pub frame_type: FrameType,
287 /// Frame flags read from or written to byte 1.
288 pub flags: u8,
289 /// Stream identifier read from or written to bytes 2..6.
290 pub stream_id: u32,
291 /// Payload length read from or written to bytes 6..10.
292 pub payload_length: u32,
293}
294
295impl FrameHeader {
296 /// Serialized header length in bytes.
297 pub const WIRE_LEN: usize = HEADER_LEN;
298}
299
300/// Typed activity surface committed by a worker build.
301#[derive(Clone, Debug, PartialEq, Eq)]
302pub struct WorkerActivityDescriptor {
303 /// Activity type.
304 pub name: String,
305 /// Canonical JSON Schema accepted by the worker.
306 pub input_schema_json: String,
307 /// Canonical JSON Schema produced by the worker.
308 pub output_schema_json: String,
309}
310
311/// Self-describing worker registration carried by [`Frame::WorkerRegister`].
312///
313/// A worker announces the routing dimensions it serves plus a stable identity so
314/// the server can associate the worker with its connection and the application can
315/// route work to it. `node` is optional locality (the routing model treats node as
316/// an optional dimension); the codec encodes it with a presence byte rather than
317/// flattening `None` to an empty string.
318#[derive(Clone, Debug, PartialEq, Eq)]
319pub struct WorkerRegistration {
320 /// Namespaces this worker serves.
321 pub namespaces: Vec<String>,
322 /// Task queue this worker pulls from.
323 pub task_queue: String,
324 /// Optional node locality; `None` when the worker declares no node affinity.
325 pub node: Option<String>,
326 /// Activity types this worker can execute.
327 pub activity_types: Vec<String>,
328 /// Stable worker identity.
329 pub identity: String,
330 /// Typed action descriptors; empty identifies a pre-contract worker.
331 pub activities: Vec<WorkerActivityDescriptor>,
332}
333
334/// Outcome of a worker registration, carried by [`Frame::WorkerRegisterAck`].
335#[derive(Clone, Debug, PartialEq, Eq)]
336pub enum WorkerRegisterOutcome {
337 /// The server accepted the registration.
338 Accepted,
339 /// The server rejected the registration, with a human-readable reason.
340 Rejected {
341 /// Human-readable rejection reason surfaced to the worker.
342 reason: String,
343 },
344}
345
346/// A typed protocol frame body plus the header metadata required to encode it.
347#[derive(Clone, PartialEq, Eq)]
348pub enum Frame {
349 /// Connection request carrying a supported version range and opaque auth token.
350 Connect {
351 flags: u8,
352 min_version: ProtocolVersion,
353 max_version: ProtocolVersion,
354 auth_token: Vec<u8>,
355 },
356 /// Connection success carrying the negotiated protocol version and server capabilities.
357 ConnectAck {
358 flags: u8,
359 selected_version: ProtocolVersion,
360 capabilities: u32,
361 },
362 /// Connection failure carrying a numeric reason and optional message.
363 ConnectError {
364 flags: u8,
365 reason_code: u16,
366 message: Option<String>,
367 },
368 /// Connection close notification with no payload.
369 Disconnect { flags: u8 },
370 /// Channel subscription request carrying a channel and accepted schema hashes.
371 Subscribe {
372 flags: u8,
373 stream_id: u32,
374 channel: String,
375 accepted_schemas: Vec<SchemaId>,
376 max_in_flight: u32,
377 },
378 /// Channel subscription success carrying a subscription id and selected schema.
379 SubscribeAck {
380 flags: u8,
381 stream_id: u32,
382 subscription_id: u64,
383 selected_schema: SchemaId,
384 },
385 /// Channel subscription failure carrying a numeric reason and optional message.
386 SubscribeError {
387 flags: u8,
388 stream_id: u32,
389 reason_code: u16,
390 message: Option<String>,
391 },
392 /// Channel unsubscription request carrying the subscription id.
393 Unsubscribe {
394 flags: u8,
395 stream_id: u32,
396 subscription_id: u64,
397 },
398 /// Publish request carrying a channel and typed message envelope.
399 ///
400 /// `idempotency_key` is `Some` only when the [`PUBLISH_IDEMPOTENCY_KEY_FLAG`]
401 /// flag bit is set; it is the dedup-on-delivery key the server feeds to its
402 /// dedup cache. When `None` (and the flag clear) the frame is byte-identical
403 /// on the wire to a pre-13-L1 publish.
404 Publish {
405 flags: u8,
406 stream_id: u32,
407 channel: String,
408 envelope: MessageEnvelope,
409 idempotency_key: Option<String>,
410 },
411 /// Publish success carrying the accepted message id.
412 ///
413 /// `durable_position` is `Some` only when the ack answers for a stored row
414 /// and the connection negotiated [`PUBLISH_ACK_POSITION_VERSION`]; the codec
415 /// then sets [`PUBLISH_DURABLE_POSITION_FLAG`] and writes the position after
416 /// `message_id`. When `None` the frame is byte-identical to the older ack.
417 PublishAck {
418 flags: u8,
419 stream_id: u32,
420 message_id: u64,
421 durable_position: Option<u64>,
422 },
423 /// Publish failure carrying a numeric reason and optional message.
424 PublishError {
425 flags: u8,
426 stream_id: u32,
427 reason_code: u16,
428 message: Option<String>,
429 },
430 /// Conversation open carrying a conversation id and subject.
431 ConversationOpen {
432 flags: u8,
433 stream_id: u32,
434 conversation_id: u64,
435 subject: String,
436 },
437 /// Conversation message carrying a conversation id and typed message envelope.
438 ConversationMessage {
439 flags: u8,
440 stream_id: u32,
441 conversation_id: u64,
442 envelope: MessageEnvelope,
443 },
444 /// Conversation close carrying a conversation id and optional reason.
445 ConversationClose {
446 flags: u8,
447 stream_id: u32,
448 conversation_id: u64,
449 reason_code: Option<u16>,
450 message: Option<String>,
451 },
452 /// Conversation failure carrying a conversation id, numeric reason, and optional message.
453 ConversationError {
454 flags: u8,
455 stream_id: u32,
456 conversation_id: u64,
457 reason_code: u16,
458 message: Option<String>,
459 },
460 /// Backpressure acceptance for a delivered message.
461 Accept {
462 flags: u8,
463 stream_id: u32,
464 referenced_message_id: MessageId,
465 },
466 /// Backpressure deferral for a buffered message.
467 Defer {
468 flags: u8,
469 stream_id: u32,
470 referenced_message_id: MessageId,
471 reason: Option<String>,
472 },
473 /// Backpressure rejection for a shed message.
474 Reject {
475 flags: u8,
476 stream_id: u32,
477 referenced_message_id: MessageId,
478 reason: Option<String>,
479 },
480 /// Connection keepalive ping.
481 Ping { flags: u8 },
482 /// Connection keepalive pong.
483 Pong { flags: u8 },
484 /// Server-initiated push carrying a correlation id and an opaque payload.
485 ///
486 /// A server writes this frame to a connected client over the client's existing
487 /// connection (server-to-client, the inverse of every other request frame). The
488 /// `correlation_id` is the key the server uses to match the client's later
489 /// [`Frame::PushReply`] back to this push; the `payload` is opaque application
490 /// bytes the server hands the client. This is an application-stream frame, so
491 /// `stream_id` is non-zero like a publish or conversation message.
492 Push {
493 flags: u8,
494 stream_id: u32,
495 correlation_id: u64,
496 payload: Vec<u8>,
497 },
498 /// Client-initiated correlated reply to a [`Frame::Push`].
499 ///
500 /// After handling a pushed frame the client writes this back on the same
501 /// connection, echoing the push's `correlation_id` so the server can match the
502 /// reply to the originating push. The `payload` is the client's opaque answer.
503 PushReply {
504 flags: u8,
505 stream_id: u32,
506 correlation_id: u64,
507 payload: Vec<u8>,
508 },
509 /// Worker self-registration over an established connection.
510 ///
511 /// A worker sends this control frame (stream 0) after the connection
512 /// handshake to announce its identity and routing dimensions. The server
513 /// associates the registration with the connection's process id and surfaces
514 /// it to the application via the connection-notifier hook, then answers with a
515 /// [`Frame::WorkerRegisterAck`]. `node` is optional locality and is encoded
516 /// with a presence byte, never flattened to an empty string.
517 WorkerRegister {
518 flags: u8,
519 registration: WorkerRegistration,
520 },
521 /// Server acknowledgement of a [`Frame::WorkerRegister`].
522 ///
523 /// Carries the registration outcome: [`WorkerRegisterOutcome::Accepted`] when
524 /// the server (and any configured notifier) accepted the worker, or
525 /// [`WorkerRegisterOutcome::Rejected`] carrying a human-readable reason when it
526 /// did not. The acknowledgement is synchronous so a worker never believes it is
527 /// registered when the application rejected it.
528 WorkerRegisterAck {
529 flags: u8,
530 outcome: WorkerRegisterOutcome,
531 },
532 /// Server-to-client delivery of a message on a subscription the client opened.
533 ///
534 /// The server writes this frame on the subscription's own `stream_id` (the
535 /// non-zero application stream negotiated by the client's `Subscribe`). It is
536 /// the only server-originated delivery frame, and it is only ever sent on a
537 /// subscription the client itself opened, so the forward-compatibility rules
538 /// for unknown frames never surprise an old client with one. `delivery_seq` is
539 /// a per-subscription monotonic counter starting at 1 — the anchor the future
540 /// ack/resume (A1 v2 credit) protocol builds on. The `envelope` reuses the same
541 /// [`MessageEnvelope`] serialization as publish and conversation frames.
542 ///
543 /// `message` is `Some` only when the server knows the message's id and boot
544 /// and the connection negotiated [`DELIVER_MESSAGE_ID_VERSION`]; the codec
545 /// then sets [`DELIVER_MESSAGE_ID_FLAG`] and writes the pair after the
546 /// envelope. When `None` the frame is byte-identical to the older Deliver.
547 Deliver {
548 flags: u8,
549 stream_id: u32,
550 delivery_seq: u64,
551 envelope: MessageEnvelope,
552 message: Option<DeliveredId>,
553 },
554 /// Forward-compatible frame preserved after length-delimited skipping.
555 Unknown {
556 type_id: u8,
557 flags: u8,
558 stream_id: u32,
559 payload: Vec<u8>,
560 },
561}
562
563impl Frame {
564 /// Construct a ping frame, enforcing the stream-0 control-frame invariant.
565 ///
566 /// # Errors
567 ///
568 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is not zero.
569 pub fn new_ping(stream_id: u32) -> Result<Self, ProtocolError> {
570 validate_stream(FrameType::Ping, stream_id)?;
571 Ok(Self::Ping { flags: 0 })
572 }
573
574 /// Construct a publish frame, enforcing the non-zero application-stream invariant.
575 ///
576 /// # Errors
577 ///
578 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
579 pub fn new_publish(
580 stream_id: u32,
581 channel: impl Into<String>,
582 envelope: MessageEnvelope,
583 ) -> Result<Self, ProtocolError> {
584 validate_stream(FrameType::Publish, stream_id)?;
585 Ok(Self::Publish {
586 flags: 0,
587 stream_id,
588 channel: channel.into(),
589 envelope,
590 idempotency_key: None,
591 })
592 }
593
594 /// Construct a publish frame carrying an idempotency key for dedup-on-delivery.
595 ///
596 /// The returned frame has [`PUBLISH_IDEMPOTENCY_KEY_FLAG`] set and serializes
597 /// the trailing key field, so the server consults its dedup cache for this key.
598 ///
599 /// # Errors
600 ///
601 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
602 pub fn new_publish_with_idempotency_key(
603 stream_id: u32,
604 channel: impl Into<String>,
605 envelope: MessageEnvelope,
606 idempotency_key: impl Into<String>,
607 ) -> Result<Self, ProtocolError> {
608 validate_stream(FrameType::Publish, stream_id)?;
609 Ok(Self::Publish {
610 flags: PUBLISH_IDEMPOTENCY_KEY_FLAG,
611 stream_id,
612 channel: channel.into(),
613 envelope,
614 idempotency_key: Some(idempotency_key.into()),
615 })
616 }
617
618 /// Construct a server-to-client push frame, enforcing the non-zero
619 /// application-stream invariant.
620 ///
621 /// # Errors
622 ///
623 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
624 pub fn new_push(
625 stream_id: u32,
626 correlation_id: u64,
627 payload: Vec<u8>,
628 ) -> Result<Self, ProtocolError> {
629 validate_stream(FrameType::Push, stream_id)?;
630 Ok(Self::Push {
631 flags: 0,
632 stream_id,
633 correlation_id,
634 payload,
635 })
636 }
637
638 /// Construct a server-to-client delivery frame on a subscription's stream,
639 /// enforcing the non-zero application-stream invariant.
640 ///
641 /// # Errors
642 ///
643 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
644 pub fn new_deliver(
645 stream_id: u32,
646 delivery_seq: u64,
647 envelope: MessageEnvelope,
648 ) -> Result<Self, ProtocolError> {
649 validate_stream(FrameType::Deliver, stream_id)?;
650 Ok(Self::Deliver {
651 flags: 0,
652 stream_id,
653 delivery_seq,
654 envelope,
655 message: None,
656 })
657 }
658
659 /// Construct a client-to-server push reply frame, echoing the correlation id of
660 /// the originating push, and enforcing the non-zero application-stream invariant.
661 ///
662 /// # Errors
663 ///
664 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
665 pub fn new_push_reply(
666 stream_id: u32,
667 correlation_id: u64,
668 payload: Vec<u8>,
669 ) -> Result<Self, ProtocolError> {
670 validate_stream(FrameType::PushReply, stream_id)?;
671 Ok(Self::PushReply {
672 flags: 0,
673 stream_id,
674 correlation_id,
675 payload,
676 })
677 }
678
679 /// Return the frame type represented by this frame body.
680 #[must_use]
681 pub const fn frame_type(&self) -> FrameType {
682 match self {
683 Self::Connect { .. } => FrameType::Connect,
684 Self::ConnectAck { .. } => FrameType::ConnectAck,
685 Self::ConnectError { .. } => FrameType::ConnectError,
686 Self::Disconnect { .. } => FrameType::Disconnect,
687 Self::Subscribe { .. } => FrameType::Subscribe,
688 Self::SubscribeAck { .. } => FrameType::SubscribeAck,
689 Self::SubscribeError { .. } => FrameType::SubscribeError,
690 Self::Unsubscribe { .. } => FrameType::Unsubscribe,
691 Self::Publish { .. } => FrameType::Publish,
692 Self::PublishAck { .. } => FrameType::PublishAck,
693 Self::PublishError { .. } => FrameType::PublishError,
694 Self::ConversationOpen { .. } => FrameType::ConversationOpen,
695 Self::ConversationMessage { .. } => FrameType::ConversationMessage,
696 Self::ConversationClose { .. } => FrameType::ConversationClose,
697 Self::ConversationError { .. } => FrameType::ConversationError,
698 Self::Accept { .. } => FrameType::Accept,
699 Self::Defer { .. } => FrameType::Defer,
700 Self::Reject { .. } => FrameType::Reject,
701 Self::Ping { .. } => FrameType::Ping,
702 Self::Pong { .. } => FrameType::Pong,
703 Self::Push { .. } => FrameType::Push,
704 Self::PushReply { .. } => FrameType::PushReply,
705 Self::WorkerRegister { .. } => FrameType::WorkerRegister,
706 Self::WorkerRegisterAck { .. } => FrameType::WorkerRegisterAck,
707 Self::Deliver { .. } => FrameType::Deliver,
708 Self::Unknown { type_id, .. } => FrameType::Unknown(*type_id),
709 }
710 }
711
712 /// Return the frame flags stored in the fixed header.
713 #[must_use]
714 pub const fn flags(&self) -> u8 {
715 match self {
716 Self::Connect { flags, .. }
717 | Self::ConnectAck { flags, .. }
718 | Self::ConnectError { flags, .. }
719 | Self::Disconnect { flags, .. }
720 | Self::Subscribe { flags, .. }
721 | Self::SubscribeAck { flags, .. }
722 | Self::SubscribeError { flags, .. }
723 | Self::Unsubscribe { flags, .. }
724 | Self::Publish { flags, .. }
725 | Self::PublishAck { flags, .. }
726 | Self::PublishError { flags, .. }
727 | Self::ConversationOpen { flags, .. }
728 | Self::ConversationMessage { flags, .. }
729 | Self::ConversationClose { flags, .. }
730 | Self::ConversationError { flags, .. }
731 | Self::Accept { flags, .. }
732 | Self::Defer { flags, .. }
733 | Self::Reject { flags, .. }
734 | Self::Ping { flags }
735 | Self::Pong { flags }
736 | Self::Push { flags, .. }
737 | Self::PushReply { flags, .. }
738 | Self::WorkerRegister { flags, .. }
739 | Self::WorkerRegisterAck { flags, .. }
740 | Self::Deliver { flags, .. }
741 | Self::Unknown { flags, .. } => *flags,
742 }
743 }
744
745 /// Return the flags byte the codec writes into the fixed header.
746 ///
747 /// This is [`Frame::flags`] except on a [`Frame::PublishAck`], whose
748 /// [`PUBLISH_DURABLE_POSITION_FLAG`] bit is set exactly when the ack carries
749 /// a durable position, and a [`Frame::Deliver`], whose
750 /// [`DELIVER_MESSAGE_ID_FLAG`] bit is set exactly when it carries a
751 /// [`DeliveredId`], so the bit and the trailing field never disagree.
752 #[must_use]
753 pub const fn wire_flags(&self) -> u8 {
754 match self {
755 Self::PublishAck {
756 flags,
757 durable_position: Some(_),
758 ..
759 } => *flags | PUBLISH_DURABLE_POSITION_FLAG,
760 Self::PublishAck {
761 flags,
762 durable_position: None,
763 ..
764 } => *flags & !PUBLISH_DURABLE_POSITION_FLAG,
765 Self::Deliver {
766 flags,
767 message: Some(_),
768 ..
769 } => *flags | DELIVER_MESSAGE_ID_FLAG,
770 Self::Deliver {
771 flags,
772 message: None,
773 ..
774 } => *flags & !DELIVER_MESSAGE_ID_FLAG,
775 _ => self.flags(),
776 }
777 }
778
779 /// Return the stream id stored in the fixed header.
780 #[must_use]
781 pub const fn stream_id(&self) -> u32 {
782 match self {
783 Self::Connect { .. }
784 | Self::ConnectAck { .. }
785 | Self::ConnectError { .. }
786 | Self::Disconnect { .. }
787 | Self::Ping { .. }
788 | Self::Pong { .. }
789 | Self::WorkerRegister { .. }
790 | Self::WorkerRegisterAck { .. } => 0,
791 Self::Subscribe { stream_id, .. }
792 | Self::SubscribeAck { stream_id, .. }
793 | Self::SubscribeError { stream_id, .. }
794 | Self::Unsubscribe { stream_id, .. }
795 | Self::Publish { stream_id, .. }
796 | Self::PublishAck { stream_id, .. }
797 | Self::PublishError { stream_id, .. }
798 | Self::ConversationOpen { stream_id, .. }
799 | Self::ConversationMessage { stream_id, .. }
800 | Self::ConversationClose { stream_id, .. }
801 | Self::ConversationError { stream_id, .. }
802 | Self::Accept { stream_id, .. }
803 | Self::Defer { stream_id, .. }
804 | Self::Reject { stream_id, .. }
805 | Self::Push { stream_id, .. }
806 | Self::PushReply { stream_id, .. }
807 | Self::Deliver { stream_id, .. }
808 | Self::Unknown { stream_id, .. } => *stream_id,
809 }
810 }
811
812 /// Validate the stream invariant for this frame.
813 pub(crate) fn validate(&self) -> Result<(), ProtocolError> {
814 validate_stream(self.frame_type(), self.stream_id())?;
815
816 if let Self::Subscribe { max_in_flight, .. } = self {
817 if *max_in_flight == 0 {
818 return Err(ProtocolError::codec(
819 "max_in_flight must be greater than zero",
820 ));
821 }
822 }
823
824 Ok(())
825 }
826}
827
828/// Validate stream placement for a known frame type.
829///
830/// # Errors
831///
832/// Returns [`ProtocolError::InvalidStream`] when `stream_id` is invalid for
833/// `frame_type`.
834pub fn validate_stream(frame_type: FrameType, stream_id: u32) -> Result<(), ProtocolError> {
835 if matches!(frame_type, FrameType::Unknown(_)) {
836 return Ok(());
837 }
838
839 let valid = if frame_type.is_control() {
840 stream_id == 0
841 } else {
842 stream_id >= 1
843 };
844
845 if valid {
846 Ok(())
847 } else {
848 Err(ProtocolError::invalid_stream(frame_type, stream_id))
849 }
850}
851
852mod debug;
853
854#[cfg(test)]
855mod tests;