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/// Status byte prefixing a [`Frame::WorkerRegisterAck`] payload that signals the
50/// registration was accepted (no further payload follows).
51pub(crate) const WORKER_REGISTER_ACK_ACCEPTED: u8 = 0x00;
52
53/// Status byte prefixing a [`Frame::WorkerRegisterAck`] payload that signals the
54/// registration was rejected (a length-prefixed reason string follows).
55pub(crate) const WORKER_REGISTER_ACK_REJECTED: u8 = 0x01;
56
57/// Protocol frame categories and their stable wire discriminants.
58#[derive(Clone, Copy, Debug, PartialEq, Eq)]
59pub enum FrameType {
60 /// Connection request.
61 Connect,
62 /// Successful connection response.
63 ConnectAck,
64 /// Failed connection response.
65 ConnectError,
66 /// Connection close notification.
67 Disconnect,
68 /// Channel subscription request.
69 Subscribe,
70 /// Successful subscription response.
71 SubscribeAck,
72 /// Failed subscription response.
73 SubscribeError,
74 /// Channel unsubscription request.
75 Unsubscribe,
76 /// Channel publish request.
77 Publish,
78 /// Successful publish response.
79 PublishAck,
80 /// Failed publish response.
81 PublishError,
82 /// Conversation lifecycle open.
83 ConversationOpen,
84 /// Conversation message delivery.
85 ConversationMessage,
86 /// Conversation lifecycle close.
87 ConversationClose,
88 /// Conversation processing error.
89 ConversationError,
90 /// In-band backpressure acceptance.
91 Accept,
92 /// In-band backpressure deferral.
93 Defer,
94 /// In-band backpressure rejection.
95 Reject,
96 /// Connection keepalive ping.
97 Ping,
98 /// Connection keepalive pong.
99 Pong,
100 /// Server-initiated push of an opaque payload to a connected client.
101 Push,
102 /// Client-initiated correlated reply to a server push.
103 PushReply,
104 /// Worker self-registration announcing identity and routing dimensions.
105 WorkerRegister,
106 /// Server acknowledgement of a worker registration (accepted or rejected).
107 WorkerRegisterAck,
108 /// Server-to-client delivery of a subscribed channel message.
109 Deliver,
110 /// Forward-compatible frame type not known to this implementation.
111 Unknown(u8),
112}
113
114impl FrameType {
115 /// Return true when this frame type must appear on stream 0.
116 #[must_use]
117 pub const fn is_control(self) -> bool {
118 matches!(
119 self,
120 Self::Connect
121 | Self::ConnectAck
122 | Self::ConnectError
123 | Self::Disconnect
124 | Self::Ping
125 | Self::Pong
126 | Self::WorkerRegister
127 | Self::WorkerRegisterAck
128 )
129 }
130}
131
132impl From<u8> for FrameType {
133 fn from(value: u8) -> Self {
134 match value {
135 0x01 => Self::Connect,
136 0x02 => Self::ConnectAck,
137 0x03 => Self::ConnectError,
138 0x04 => Self::Disconnect,
139 0x05 => Self::Subscribe,
140 0x06 => Self::SubscribeAck,
141 0x07 => Self::SubscribeError,
142 0x08 => Self::Unsubscribe,
143 0x09 => Self::Publish,
144 0x0A => Self::PublishAck,
145 0x0B => Self::PublishError,
146 0x0C => Self::ConversationOpen,
147 0x0D => Self::ConversationMessage,
148 0x0E => Self::ConversationClose,
149 0x0F => Self::ConversationError,
150 0x10 => Self::Accept,
151 0x11 => Self::Defer,
152 0x12 => Self::Reject,
153 0x13 => Self::Ping,
154 0x14 => Self::Pong,
155 0x15 => Self::Push,
156 0x16 => Self::PushReply,
157 0x17 => Self::WorkerRegister,
158 0x18 => Self::WorkerRegisterAck,
159 0x19 => Self::Deliver,
160 unknown => Self::Unknown(unknown),
161 }
162 }
163}
164
165impl From<FrameType> for u8 {
166 fn from(value: FrameType) -> Self {
167 match value {
168 FrameType::Connect => 0x01,
169 FrameType::ConnectAck => 0x02,
170 FrameType::ConnectError => 0x03,
171 FrameType::Disconnect => 0x04,
172 FrameType::Subscribe => 0x05,
173 FrameType::SubscribeAck => 0x06,
174 FrameType::SubscribeError => 0x07,
175 FrameType::Unsubscribe => 0x08,
176 FrameType::Publish => 0x09,
177 FrameType::PublishAck => 0x0A,
178 FrameType::PublishError => 0x0B,
179 FrameType::ConversationOpen => 0x0C,
180 FrameType::ConversationMessage => 0x0D,
181 FrameType::ConversationClose => 0x0E,
182 FrameType::ConversationError => 0x0F,
183 FrameType::Accept => 0x10,
184 FrameType::Defer => 0x11,
185 FrameType::Reject => 0x12,
186 FrameType::Ping => 0x13,
187 FrameType::Pong => 0x14,
188 FrameType::Push => 0x15,
189 FrameType::PushReply => 0x16,
190 FrameType::WorkerRegister => 0x17,
191 FrameType::WorkerRegisterAck => 0x18,
192 FrameType::Deliver => 0x19,
193 FrameType::Unknown(type_id) => type_id,
194 }
195 }
196}
197
198/// Fixed-size frame prefix: type, flags, stream identifier, and payload length.
199#[derive(Clone, Copy, Debug, PartialEq, Eq)]
200pub struct FrameHeader {
201 /// Frame type read from or written to byte 0.
202 pub frame_type: FrameType,
203 /// Frame flags read from or written to byte 1.
204 pub flags: u8,
205 /// Stream identifier read from or written to bytes 2..6.
206 pub stream_id: u32,
207 /// Payload length read from or written to bytes 6..10.
208 pub payload_length: u32,
209}
210
211impl FrameHeader {
212 /// Serialized header length in bytes.
213 pub const WIRE_LEN: usize = HEADER_LEN;
214}
215
216/// Self-describing worker registration carried by [`Frame::WorkerRegister`].
217///
218/// A worker announces the routing dimensions it serves plus a stable identity so
219/// the server can associate the worker with its connection and the application can
220/// route work to it. `node` is optional locality (the routing model treats node as
221/// an optional dimension); the codec encodes it with a presence byte rather than
222/// flattening `None` to an empty string.
223#[derive(Clone, Debug, PartialEq, Eq)]
224pub struct WorkerRegistration {
225 /// Namespaces this worker serves.
226 pub namespaces: Vec<String>,
227 /// Task queue this worker pulls from.
228 pub task_queue: String,
229 /// Optional node locality; `None` when the worker declares no node affinity.
230 pub node: Option<String>,
231 /// Activity types this worker can execute.
232 pub activity_types: Vec<String>,
233 /// Stable worker identity.
234 pub identity: String,
235}
236
237/// Outcome of a worker registration, carried by [`Frame::WorkerRegisterAck`].
238#[derive(Clone, Debug, PartialEq, Eq)]
239pub enum WorkerRegisterOutcome {
240 /// The server accepted the registration.
241 Accepted,
242 /// The server rejected the registration, with a human-readable reason.
243 Rejected {
244 /// Human-readable rejection reason surfaced to the worker.
245 reason: String,
246 },
247}
248
249/// A typed protocol frame body plus the header metadata required to encode it.
250#[derive(Clone, Debug, PartialEq, Eq)]
251pub enum Frame {
252 /// Connection request carrying a supported version range and opaque auth token.
253 Connect {
254 flags: u8,
255 min_version: ProtocolVersion,
256 max_version: ProtocolVersion,
257 auth_token: Vec<u8>,
258 },
259 /// Connection success carrying the negotiated protocol version and server capabilities.
260 ConnectAck {
261 flags: u8,
262 selected_version: ProtocolVersion,
263 capabilities: u32,
264 },
265 /// Connection failure carrying a numeric reason and optional message.
266 ConnectError {
267 flags: u8,
268 reason_code: u16,
269 message: Option<String>,
270 },
271 /// Connection close notification with no payload.
272 Disconnect { flags: u8 },
273 /// Channel subscription request carrying a channel and accepted schema hashes.
274 Subscribe {
275 flags: u8,
276 stream_id: u32,
277 channel: String,
278 accepted_schemas: Vec<SchemaId>,
279 max_in_flight: u32,
280 },
281 /// Channel subscription success carrying a subscription id and selected schema.
282 SubscribeAck {
283 flags: u8,
284 stream_id: u32,
285 subscription_id: u64,
286 selected_schema: SchemaId,
287 },
288 /// Channel subscription failure carrying a numeric reason and optional message.
289 SubscribeError {
290 flags: u8,
291 stream_id: u32,
292 reason_code: u16,
293 message: Option<String>,
294 },
295 /// Channel unsubscription request carrying the subscription id.
296 Unsubscribe {
297 flags: u8,
298 stream_id: u32,
299 subscription_id: u64,
300 },
301 /// Publish request carrying a channel and typed message envelope.
302 ///
303 /// `idempotency_key` is `Some` only when the [`PUBLISH_IDEMPOTENCY_KEY_FLAG`]
304 /// flag bit is set; it is the dedup-on-delivery key the server feeds to its
305 /// dedup cache. When `None` (and the flag clear) the frame is byte-identical
306 /// on the wire to a pre-13-L1 publish.
307 Publish {
308 flags: u8,
309 stream_id: u32,
310 channel: String,
311 envelope: MessageEnvelope,
312 idempotency_key: Option<String>,
313 },
314 /// Publish success carrying the accepted message id.
315 PublishAck {
316 flags: u8,
317 stream_id: u32,
318 message_id: u64,
319 },
320 /// Publish failure carrying a numeric reason and optional message.
321 PublishError {
322 flags: u8,
323 stream_id: u32,
324 reason_code: u16,
325 message: Option<String>,
326 },
327 /// Conversation open carrying a conversation id and subject.
328 ConversationOpen {
329 flags: u8,
330 stream_id: u32,
331 conversation_id: u64,
332 subject: String,
333 },
334 /// Conversation message carrying a conversation id and typed message envelope.
335 ConversationMessage {
336 flags: u8,
337 stream_id: u32,
338 conversation_id: u64,
339 envelope: MessageEnvelope,
340 },
341 /// Conversation close carrying a conversation id and optional reason.
342 ConversationClose {
343 flags: u8,
344 stream_id: u32,
345 conversation_id: u64,
346 reason_code: Option<u16>,
347 message: Option<String>,
348 },
349 /// Conversation failure carrying a conversation id, numeric reason, and optional message.
350 ConversationError {
351 flags: u8,
352 stream_id: u32,
353 conversation_id: u64,
354 reason_code: u16,
355 message: Option<String>,
356 },
357 /// Backpressure acceptance for a delivered message.
358 Accept {
359 flags: u8,
360 stream_id: u32,
361 referenced_message_id: MessageId,
362 },
363 /// Backpressure deferral for a buffered message.
364 Defer {
365 flags: u8,
366 stream_id: u32,
367 referenced_message_id: MessageId,
368 reason: Option<String>,
369 },
370 /// Backpressure rejection for a shed message.
371 Reject {
372 flags: u8,
373 stream_id: u32,
374 referenced_message_id: MessageId,
375 reason: Option<String>,
376 },
377 /// Connection keepalive ping.
378 Ping { flags: u8 },
379 /// Connection keepalive pong.
380 Pong { flags: u8 },
381 /// Server-initiated push carrying a correlation id and an opaque payload.
382 ///
383 /// A server writes this frame to a connected client over the client's existing
384 /// connection (server-to-client, the inverse of every other request frame). The
385 /// `correlation_id` is the key the server uses to match the client's later
386 /// [`Frame::PushReply`] back to this push; the `payload` is opaque application
387 /// bytes the server hands the client. This is an application-stream frame, so
388 /// `stream_id` is non-zero like a publish or conversation message.
389 Push {
390 flags: u8,
391 stream_id: u32,
392 correlation_id: u64,
393 payload: Vec<u8>,
394 },
395 /// Client-initiated correlated reply to a [`Frame::Push`].
396 ///
397 /// After handling a pushed frame the client writes this back on the same
398 /// connection, echoing the push's `correlation_id` so the server can match the
399 /// reply to the originating push. The `payload` is the client's opaque answer.
400 PushReply {
401 flags: u8,
402 stream_id: u32,
403 correlation_id: u64,
404 payload: Vec<u8>,
405 },
406 /// Worker self-registration over an established connection.
407 ///
408 /// A worker sends this control frame (stream 0) after the connection
409 /// handshake to announce its identity and routing dimensions. The server
410 /// associates the registration with the connection's process id and surfaces
411 /// it to the application via the connection-notifier hook, then answers with a
412 /// [`Frame::WorkerRegisterAck`]. `node` is optional locality and is encoded
413 /// with a presence byte, never flattened to an empty string.
414 WorkerRegister {
415 flags: u8,
416 registration: WorkerRegistration,
417 },
418 /// Server acknowledgement of a [`Frame::WorkerRegister`].
419 ///
420 /// Carries the registration outcome: [`WorkerRegisterOutcome::Accepted`] when
421 /// the server (and any configured notifier) accepted the worker, or
422 /// [`WorkerRegisterOutcome::Rejected`] carrying a human-readable reason when it
423 /// did not. The acknowledgement is synchronous so a worker never believes it is
424 /// registered when the application rejected it.
425 WorkerRegisterAck {
426 flags: u8,
427 outcome: WorkerRegisterOutcome,
428 },
429 /// Server-to-client delivery of a message on a subscription the client opened.
430 ///
431 /// The server writes this frame on the subscription's own `stream_id` (the
432 /// non-zero application stream negotiated by the client's `Subscribe`). It is
433 /// the only server-originated delivery frame, and it is only ever sent on a
434 /// subscription the client itself opened, so the forward-compatibility rules
435 /// for unknown frames never surprise an old client with one. `delivery_seq` is
436 /// a per-subscription monotonic counter starting at 1 — the anchor the future
437 /// ack/resume (A1 v2 credit) protocol builds on. The `envelope` reuses the same
438 /// [`MessageEnvelope`] serialization as publish and conversation frames.
439 Deliver {
440 flags: u8,
441 stream_id: u32,
442 delivery_seq: u64,
443 envelope: MessageEnvelope,
444 },
445 /// Forward-compatible frame preserved after length-delimited skipping.
446 Unknown {
447 type_id: u8,
448 flags: u8,
449 stream_id: u32,
450 payload: Vec<u8>,
451 },
452}
453
454impl Frame {
455 /// Construct a ping frame, enforcing the stream-0 control-frame invariant.
456 ///
457 /// # Errors
458 ///
459 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is not zero.
460 pub fn new_ping(stream_id: u32) -> Result<Self, ProtocolError> {
461 validate_stream(FrameType::Ping, stream_id)?;
462 Ok(Self::Ping { flags: 0 })
463 }
464
465 /// Construct a publish frame, enforcing the non-zero application-stream invariant.
466 ///
467 /// # Errors
468 ///
469 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
470 pub fn new_publish(
471 stream_id: u32,
472 channel: impl Into<String>,
473 envelope: MessageEnvelope,
474 ) -> Result<Self, ProtocolError> {
475 validate_stream(FrameType::Publish, stream_id)?;
476 Ok(Self::Publish {
477 flags: 0,
478 stream_id,
479 channel: channel.into(),
480 envelope,
481 idempotency_key: None,
482 })
483 }
484
485 /// Construct a publish frame carrying an idempotency key for dedup-on-delivery.
486 ///
487 /// The returned frame has [`PUBLISH_IDEMPOTENCY_KEY_FLAG`] set and serializes
488 /// the trailing key field, so the server consults its dedup cache for this key.
489 ///
490 /// # Errors
491 ///
492 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
493 pub fn new_publish_with_idempotency_key(
494 stream_id: u32,
495 channel: impl Into<String>,
496 envelope: MessageEnvelope,
497 idempotency_key: impl Into<String>,
498 ) -> Result<Self, ProtocolError> {
499 validate_stream(FrameType::Publish, stream_id)?;
500 Ok(Self::Publish {
501 flags: PUBLISH_IDEMPOTENCY_KEY_FLAG,
502 stream_id,
503 channel: channel.into(),
504 envelope,
505 idempotency_key: Some(idempotency_key.into()),
506 })
507 }
508
509 /// Construct a server-to-client push frame, enforcing the non-zero
510 /// application-stream invariant.
511 ///
512 /// # Errors
513 ///
514 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
515 pub fn new_push(
516 stream_id: u32,
517 correlation_id: u64,
518 payload: Vec<u8>,
519 ) -> Result<Self, ProtocolError> {
520 validate_stream(FrameType::Push, stream_id)?;
521 Ok(Self::Push {
522 flags: 0,
523 stream_id,
524 correlation_id,
525 payload,
526 })
527 }
528
529 /// Construct a server-to-client delivery frame on a subscription's stream,
530 /// enforcing the non-zero application-stream invariant.
531 ///
532 /// # Errors
533 ///
534 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
535 pub fn new_deliver(
536 stream_id: u32,
537 delivery_seq: u64,
538 envelope: MessageEnvelope,
539 ) -> Result<Self, ProtocolError> {
540 validate_stream(FrameType::Deliver, stream_id)?;
541 Ok(Self::Deliver {
542 flags: 0,
543 stream_id,
544 delivery_seq,
545 envelope,
546 })
547 }
548
549 /// Construct a client-to-server push reply frame, echoing the correlation id of
550 /// the originating push, and enforcing the non-zero application-stream invariant.
551 ///
552 /// # Errors
553 ///
554 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
555 pub fn new_push_reply(
556 stream_id: u32,
557 correlation_id: u64,
558 payload: Vec<u8>,
559 ) -> Result<Self, ProtocolError> {
560 validate_stream(FrameType::PushReply, stream_id)?;
561 Ok(Self::PushReply {
562 flags: 0,
563 stream_id,
564 correlation_id,
565 payload,
566 })
567 }
568
569 /// Return the frame type represented by this frame body.
570 #[must_use]
571 pub const fn frame_type(&self) -> FrameType {
572 match self {
573 Self::Connect { .. } => FrameType::Connect,
574 Self::ConnectAck { .. } => FrameType::ConnectAck,
575 Self::ConnectError { .. } => FrameType::ConnectError,
576 Self::Disconnect { .. } => FrameType::Disconnect,
577 Self::Subscribe { .. } => FrameType::Subscribe,
578 Self::SubscribeAck { .. } => FrameType::SubscribeAck,
579 Self::SubscribeError { .. } => FrameType::SubscribeError,
580 Self::Unsubscribe { .. } => FrameType::Unsubscribe,
581 Self::Publish { .. } => FrameType::Publish,
582 Self::PublishAck { .. } => FrameType::PublishAck,
583 Self::PublishError { .. } => FrameType::PublishError,
584 Self::ConversationOpen { .. } => FrameType::ConversationOpen,
585 Self::ConversationMessage { .. } => FrameType::ConversationMessage,
586 Self::ConversationClose { .. } => FrameType::ConversationClose,
587 Self::ConversationError { .. } => FrameType::ConversationError,
588 Self::Accept { .. } => FrameType::Accept,
589 Self::Defer { .. } => FrameType::Defer,
590 Self::Reject { .. } => FrameType::Reject,
591 Self::Ping { .. } => FrameType::Ping,
592 Self::Pong { .. } => FrameType::Pong,
593 Self::Push { .. } => FrameType::Push,
594 Self::PushReply { .. } => FrameType::PushReply,
595 Self::WorkerRegister { .. } => FrameType::WorkerRegister,
596 Self::WorkerRegisterAck { .. } => FrameType::WorkerRegisterAck,
597 Self::Deliver { .. } => FrameType::Deliver,
598 Self::Unknown { type_id, .. } => FrameType::Unknown(*type_id),
599 }
600 }
601
602 /// Return the frame flags stored in the fixed header.
603 #[must_use]
604 pub const fn flags(&self) -> u8 {
605 match self {
606 Self::Connect { flags, .. }
607 | Self::ConnectAck { flags, .. }
608 | Self::ConnectError { flags, .. }
609 | Self::Disconnect { flags, .. }
610 | Self::Subscribe { flags, .. }
611 | Self::SubscribeAck { flags, .. }
612 | Self::SubscribeError { flags, .. }
613 | Self::Unsubscribe { flags, .. }
614 | Self::Publish { flags, .. }
615 | Self::PublishAck { flags, .. }
616 | Self::PublishError { flags, .. }
617 | Self::ConversationOpen { flags, .. }
618 | Self::ConversationMessage { flags, .. }
619 | Self::ConversationClose { flags, .. }
620 | Self::ConversationError { flags, .. }
621 | Self::Accept { flags, .. }
622 | Self::Defer { flags, .. }
623 | Self::Reject { flags, .. }
624 | Self::Ping { flags }
625 | Self::Pong { flags }
626 | Self::Push { flags, .. }
627 | Self::PushReply { flags, .. }
628 | Self::WorkerRegister { flags, .. }
629 | Self::WorkerRegisterAck { flags, .. }
630 | Self::Deliver { flags, .. }
631 | Self::Unknown { flags, .. } => *flags,
632 }
633 }
634
635 /// Return the stream id stored in the fixed header.
636 #[must_use]
637 pub const fn stream_id(&self) -> u32 {
638 match self {
639 Self::Connect { .. }
640 | Self::ConnectAck { .. }
641 | Self::ConnectError { .. }
642 | Self::Disconnect { .. }
643 | Self::Ping { .. }
644 | Self::Pong { .. }
645 | Self::WorkerRegister { .. }
646 | Self::WorkerRegisterAck { .. } => 0,
647 Self::Subscribe { stream_id, .. }
648 | Self::SubscribeAck { stream_id, .. }
649 | Self::SubscribeError { stream_id, .. }
650 | Self::Unsubscribe { stream_id, .. }
651 | Self::Publish { stream_id, .. }
652 | Self::PublishAck { stream_id, .. }
653 | Self::PublishError { stream_id, .. }
654 | Self::ConversationOpen { stream_id, .. }
655 | Self::ConversationMessage { stream_id, .. }
656 | Self::ConversationClose { stream_id, .. }
657 | Self::ConversationError { stream_id, .. }
658 | Self::Accept { stream_id, .. }
659 | Self::Defer { stream_id, .. }
660 | Self::Reject { stream_id, .. }
661 | Self::Push { stream_id, .. }
662 | Self::PushReply { stream_id, .. }
663 | Self::Deliver { stream_id, .. }
664 | Self::Unknown { stream_id, .. } => *stream_id,
665 }
666 }
667
668 /// Validate the stream invariant for this frame.
669 pub(crate) fn validate(&self) -> Result<(), ProtocolError> {
670 validate_stream(self.frame_type(), self.stream_id())?;
671
672 if let Self::Subscribe { max_in_flight, .. } = self {
673 if *max_in_flight == 0 {
674 return Err(ProtocolError::codec(
675 "max_in_flight must be greater than zero",
676 ));
677 }
678 }
679
680 Ok(())
681 }
682}
683
684/// Validate stream placement for a known frame type.
685///
686/// # Errors
687///
688/// Returns [`ProtocolError::InvalidStream`] when `stream_id` is invalid for
689/// `frame_type`.
690pub fn validate_stream(frame_type: FrameType, stream_id: u32) -> Result<(), ProtocolError> {
691 if matches!(frame_type, FrameType::Unknown(_)) {
692 return Ok(());
693 }
694
695 let valid = if frame_type.is_control() {
696 stream_id == 0
697 } else {
698 stream_id >= 1
699 };
700
701 if valid {
702 Ok(())
703 } else {
704 Err(ProtocolError::invalid_stream(frame_type, stream_id))
705 }
706}
707
708#[cfg(test)]
709mod tests;