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/// Typed activity surface committed by a worker build.
217#[derive(Clone, Debug, PartialEq, Eq)]
218pub struct WorkerActivityDescriptor {
219 /// Activity type.
220 pub name: String,
221 /// Canonical JSON Schema accepted by the worker.
222 pub input_schema_json: String,
223 /// Canonical JSON Schema produced by the worker.
224 pub output_schema_json: String,
225}
226
227/// Self-describing worker registration carried by [`Frame::WorkerRegister`].
228///
229/// A worker announces the routing dimensions it serves plus a stable identity so
230/// the server can associate the worker with its connection and the application can
231/// route work to it. `node` is optional locality (the routing model treats node as
232/// an optional dimension); the codec encodes it with a presence byte rather than
233/// flattening `None` to an empty string.
234#[derive(Clone, Debug, PartialEq, Eq)]
235pub struct WorkerRegistration {
236 /// Namespaces this worker serves.
237 pub namespaces: Vec<String>,
238 /// Task queue this worker pulls from.
239 pub task_queue: String,
240 /// Optional node locality; `None` when the worker declares no node affinity.
241 pub node: Option<String>,
242 /// Activity types this worker can execute.
243 pub activity_types: Vec<String>,
244 /// Stable worker identity.
245 pub identity: String,
246 /// Typed action descriptors; empty identifies a pre-contract worker.
247 pub activities: Vec<WorkerActivityDescriptor>,
248}
249
250/// Outcome of a worker registration, carried by [`Frame::WorkerRegisterAck`].
251#[derive(Clone, Debug, PartialEq, Eq)]
252pub enum WorkerRegisterOutcome {
253 /// The server accepted the registration.
254 Accepted,
255 /// The server rejected the registration, with a human-readable reason.
256 Rejected {
257 /// Human-readable rejection reason surfaced to the worker.
258 reason: String,
259 },
260}
261
262/// A typed protocol frame body plus the header metadata required to encode it.
263#[derive(Clone, Debug, PartialEq, Eq)]
264pub enum Frame {
265 /// Connection request carrying a supported version range and opaque auth token.
266 Connect {
267 flags: u8,
268 min_version: ProtocolVersion,
269 max_version: ProtocolVersion,
270 auth_token: Vec<u8>,
271 },
272 /// Connection success carrying the negotiated protocol version and server capabilities.
273 ConnectAck {
274 flags: u8,
275 selected_version: ProtocolVersion,
276 capabilities: u32,
277 },
278 /// Connection failure carrying a numeric reason and optional message.
279 ConnectError {
280 flags: u8,
281 reason_code: u16,
282 message: Option<String>,
283 },
284 /// Connection close notification with no payload.
285 Disconnect { flags: u8 },
286 /// Channel subscription request carrying a channel and accepted schema hashes.
287 Subscribe {
288 flags: u8,
289 stream_id: u32,
290 channel: String,
291 accepted_schemas: Vec<SchemaId>,
292 max_in_flight: u32,
293 },
294 /// Channel subscription success carrying a subscription id and selected schema.
295 SubscribeAck {
296 flags: u8,
297 stream_id: u32,
298 subscription_id: u64,
299 selected_schema: SchemaId,
300 },
301 /// Channel subscription failure carrying a numeric reason and optional message.
302 SubscribeError {
303 flags: u8,
304 stream_id: u32,
305 reason_code: u16,
306 message: Option<String>,
307 },
308 /// Channel unsubscription request carrying the subscription id.
309 Unsubscribe {
310 flags: u8,
311 stream_id: u32,
312 subscription_id: u64,
313 },
314 /// Publish request carrying a channel and typed message envelope.
315 ///
316 /// `idempotency_key` is `Some` only when the [`PUBLISH_IDEMPOTENCY_KEY_FLAG`]
317 /// flag bit is set; it is the dedup-on-delivery key the server feeds to its
318 /// dedup cache. When `None` (and the flag clear) the frame is byte-identical
319 /// on the wire to a pre-13-L1 publish.
320 Publish {
321 flags: u8,
322 stream_id: u32,
323 channel: String,
324 envelope: MessageEnvelope,
325 idempotency_key: Option<String>,
326 },
327 /// Publish success carrying the accepted message id.
328 PublishAck {
329 flags: u8,
330 stream_id: u32,
331 message_id: u64,
332 },
333 /// Publish failure carrying a numeric reason and optional message.
334 PublishError {
335 flags: u8,
336 stream_id: u32,
337 reason_code: u16,
338 message: Option<String>,
339 },
340 /// Conversation open carrying a conversation id and subject.
341 ConversationOpen {
342 flags: u8,
343 stream_id: u32,
344 conversation_id: u64,
345 subject: String,
346 },
347 /// Conversation message carrying a conversation id and typed message envelope.
348 ConversationMessage {
349 flags: u8,
350 stream_id: u32,
351 conversation_id: u64,
352 envelope: MessageEnvelope,
353 },
354 /// Conversation close carrying a conversation id and optional reason.
355 ConversationClose {
356 flags: u8,
357 stream_id: u32,
358 conversation_id: u64,
359 reason_code: Option<u16>,
360 message: Option<String>,
361 },
362 /// Conversation failure carrying a conversation id, numeric reason, and optional message.
363 ConversationError {
364 flags: u8,
365 stream_id: u32,
366 conversation_id: u64,
367 reason_code: u16,
368 message: Option<String>,
369 },
370 /// Backpressure acceptance for a delivered message.
371 Accept {
372 flags: u8,
373 stream_id: u32,
374 referenced_message_id: MessageId,
375 },
376 /// Backpressure deferral for a buffered message.
377 Defer {
378 flags: u8,
379 stream_id: u32,
380 referenced_message_id: MessageId,
381 reason: Option<String>,
382 },
383 /// Backpressure rejection for a shed message.
384 Reject {
385 flags: u8,
386 stream_id: u32,
387 referenced_message_id: MessageId,
388 reason: Option<String>,
389 },
390 /// Connection keepalive ping.
391 Ping { flags: u8 },
392 /// Connection keepalive pong.
393 Pong { flags: u8 },
394 /// Server-initiated push carrying a correlation id and an opaque payload.
395 ///
396 /// A server writes this frame to a connected client over the client's existing
397 /// connection (server-to-client, the inverse of every other request frame). The
398 /// `correlation_id` is the key the server uses to match the client's later
399 /// [`Frame::PushReply`] back to this push; the `payload` is opaque application
400 /// bytes the server hands the client. This is an application-stream frame, so
401 /// `stream_id` is non-zero like a publish or conversation message.
402 Push {
403 flags: u8,
404 stream_id: u32,
405 correlation_id: u64,
406 payload: Vec<u8>,
407 },
408 /// Client-initiated correlated reply to a [`Frame::Push`].
409 ///
410 /// After handling a pushed frame the client writes this back on the same
411 /// connection, echoing the push's `correlation_id` so the server can match the
412 /// reply to the originating push. The `payload` is the client's opaque answer.
413 PushReply {
414 flags: u8,
415 stream_id: u32,
416 correlation_id: u64,
417 payload: Vec<u8>,
418 },
419 /// Worker self-registration over an established connection.
420 ///
421 /// A worker sends this control frame (stream 0) after the connection
422 /// handshake to announce its identity and routing dimensions. The server
423 /// associates the registration with the connection's process id and surfaces
424 /// it to the application via the connection-notifier hook, then answers with a
425 /// [`Frame::WorkerRegisterAck`]. `node` is optional locality and is encoded
426 /// with a presence byte, never flattened to an empty string.
427 WorkerRegister {
428 flags: u8,
429 registration: WorkerRegistration,
430 },
431 /// Server acknowledgement of a [`Frame::WorkerRegister`].
432 ///
433 /// Carries the registration outcome: [`WorkerRegisterOutcome::Accepted`] when
434 /// the server (and any configured notifier) accepted the worker, or
435 /// [`WorkerRegisterOutcome::Rejected`] carrying a human-readable reason when it
436 /// did not. The acknowledgement is synchronous so a worker never believes it is
437 /// registered when the application rejected it.
438 WorkerRegisterAck {
439 flags: u8,
440 outcome: WorkerRegisterOutcome,
441 },
442 /// Server-to-client delivery of a message on a subscription the client opened.
443 ///
444 /// The server writes this frame on the subscription's own `stream_id` (the
445 /// non-zero application stream negotiated by the client's `Subscribe`). It is
446 /// the only server-originated delivery frame, and it is only ever sent on a
447 /// subscription the client itself opened, so the forward-compatibility rules
448 /// for unknown frames never surprise an old client with one. `delivery_seq` is
449 /// a per-subscription monotonic counter starting at 1 — the anchor the future
450 /// ack/resume (A1 v2 credit) protocol builds on. The `envelope` reuses the same
451 /// [`MessageEnvelope`] serialization as publish and conversation frames.
452 Deliver {
453 flags: u8,
454 stream_id: u32,
455 delivery_seq: u64,
456 envelope: MessageEnvelope,
457 },
458 /// Forward-compatible frame preserved after length-delimited skipping.
459 Unknown {
460 type_id: u8,
461 flags: u8,
462 stream_id: u32,
463 payload: Vec<u8>,
464 },
465}
466
467impl Frame {
468 /// Construct a ping frame, enforcing the stream-0 control-frame invariant.
469 ///
470 /// # Errors
471 ///
472 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is not zero.
473 pub fn new_ping(stream_id: u32) -> Result<Self, ProtocolError> {
474 validate_stream(FrameType::Ping, stream_id)?;
475 Ok(Self::Ping { flags: 0 })
476 }
477
478 /// Construct a publish frame, enforcing the non-zero application-stream invariant.
479 ///
480 /// # Errors
481 ///
482 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
483 pub fn new_publish(
484 stream_id: u32,
485 channel: impl Into<String>,
486 envelope: MessageEnvelope,
487 ) -> Result<Self, ProtocolError> {
488 validate_stream(FrameType::Publish, stream_id)?;
489 Ok(Self::Publish {
490 flags: 0,
491 stream_id,
492 channel: channel.into(),
493 envelope,
494 idempotency_key: None,
495 })
496 }
497
498 /// Construct a publish frame carrying an idempotency key for dedup-on-delivery.
499 ///
500 /// The returned frame has [`PUBLISH_IDEMPOTENCY_KEY_FLAG`] set and serializes
501 /// the trailing key field, so the server consults its dedup cache for this key.
502 ///
503 /// # Errors
504 ///
505 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
506 pub fn new_publish_with_idempotency_key(
507 stream_id: u32,
508 channel: impl Into<String>,
509 envelope: MessageEnvelope,
510 idempotency_key: impl Into<String>,
511 ) -> Result<Self, ProtocolError> {
512 validate_stream(FrameType::Publish, stream_id)?;
513 Ok(Self::Publish {
514 flags: PUBLISH_IDEMPOTENCY_KEY_FLAG,
515 stream_id,
516 channel: channel.into(),
517 envelope,
518 idempotency_key: Some(idempotency_key.into()),
519 })
520 }
521
522 /// Construct a server-to-client push frame, enforcing the non-zero
523 /// application-stream invariant.
524 ///
525 /// # Errors
526 ///
527 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
528 pub fn new_push(
529 stream_id: u32,
530 correlation_id: u64,
531 payload: Vec<u8>,
532 ) -> Result<Self, ProtocolError> {
533 validate_stream(FrameType::Push, stream_id)?;
534 Ok(Self::Push {
535 flags: 0,
536 stream_id,
537 correlation_id,
538 payload,
539 })
540 }
541
542 /// Construct a server-to-client delivery frame on a subscription's stream,
543 /// enforcing the non-zero application-stream invariant.
544 ///
545 /// # Errors
546 ///
547 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
548 pub fn new_deliver(
549 stream_id: u32,
550 delivery_seq: u64,
551 envelope: MessageEnvelope,
552 ) -> Result<Self, ProtocolError> {
553 validate_stream(FrameType::Deliver, stream_id)?;
554 Ok(Self::Deliver {
555 flags: 0,
556 stream_id,
557 delivery_seq,
558 envelope,
559 })
560 }
561
562 /// Construct a client-to-server push reply frame, echoing the correlation id of
563 /// the originating push, and enforcing the non-zero application-stream invariant.
564 ///
565 /// # Errors
566 ///
567 /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
568 pub fn new_push_reply(
569 stream_id: u32,
570 correlation_id: u64,
571 payload: Vec<u8>,
572 ) -> Result<Self, ProtocolError> {
573 validate_stream(FrameType::PushReply, stream_id)?;
574 Ok(Self::PushReply {
575 flags: 0,
576 stream_id,
577 correlation_id,
578 payload,
579 })
580 }
581
582 /// Return the frame type represented by this frame body.
583 #[must_use]
584 pub const fn frame_type(&self) -> FrameType {
585 match self {
586 Self::Connect { .. } => FrameType::Connect,
587 Self::ConnectAck { .. } => FrameType::ConnectAck,
588 Self::ConnectError { .. } => FrameType::ConnectError,
589 Self::Disconnect { .. } => FrameType::Disconnect,
590 Self::Subscribe { .. } => FrameType::Subscribe,
591 Self::SubscribeAck { .. } => FrameType::SubscribeAck,
592 Self::SubscribeError { .. } => FrameType::SubscribeError,
593 Self::Unsubscribe { .. } => FrameType::Unsubscribe,
594 Self::Publish { .. } => FrameType::Publish,
595 Self::PublishAck { .. } => FrameType::PublishAck,
596 Self::PublishError { .. } => FrameType::PublishError,
597 Self::ConversationOpen { .. } => FrameType::ConversationOpen,
598 Self::ConversationMessage { .. } => FrameType::ConversationMessage,
599 Self::ConversationClose { .. } => FrameType::ConversationClose,
600 Self::ConversationError { .. } => FrameType::ConversationError,
601 Self::Accept { .. } => FrameType::Accept,
602 Self::Defer { .. } => FrameType::Defer,
603 Self::Reject { .. } => FrameType::Reject,
604 Self::Ping { .. } => FrameType::Ping,
605 Self::Pong { .. } => FrameType::Pong,
606 Self::Push { .. } => FrameType::Push,
607 Self::PushReply { .. } => FrameType::PushReply,
608 Self::WorkerRegister { .. } => FrameType::WorkerRegister,
609 Self::WorkerRegisterAck { .. } => FrameType::WorkerRegisterAck,
610 Self::Deliver { .. } => FrameType::Deliver,
611 Self::Unknown { type_id, .. } => FrameType::Unknown(*type_id),
612 }
613 }
614
615 /// Return the frame flags stored in the fixed header.
616 #[must_use]
617 pub const fn flags(&self) -> u8 {
618 match self {
619 Self::Connect { flags, .. }
620 | Self::ConnectAck { flags, .. }
621 | Self::ConnectError { flags, .. }
622 | Self::Disconnect { flags, .. }
623 | Self::Subscribe { flags, .. }
624 | Self::SubscribeAck { flags, .. }
625 | Self::SubscribeError { flags, .. }
626 | Self::Unsubscribe { flags, .. }
627 | Self::Publish { flags, .. }
628 | Self::PublishAck { flags, .. }
629 | Self::PublishError { flags, .. }
630 | Self::ConversationOpen { flags, .. }
631 | Self::ConversationMessage { flags, .. }
632 | Self::ConversationClose { flags, .. }
633 | Self::ConversationError { flags, .. }
634 | Self::Accept { flags, .. }
635 | Self::Defer { flags, .. }
636 | Self::Reject { flags, .. }
637 | Self::Ping { flags }
638 | Self::Pong { flags }
639 | Self::Push { flags, .. }
640 | Self::PushReply { flags, .. }
641 | Self::WorkerRegister { flags, .. }
642 | Self::WorkerRegisterAck { flags, .. }
643 | Self::Deliver { flags, .. }
644 | Self::Unknown { flags, .. } => *flags,
645 }
646 }
647
648 /// Return the stream id stored in the fixed header.
649 #[must_use]
650 pub const fn stream_id(&self) -> u32 {
651 match self {
652 Self::Connect { .. }
653 | Self::ConnectAck { .. }
654 | Self::ConnectError { .. }
655 | Self::Disconnect { .. }
656 | Self::Ping { .. }
657 | Self::Pong { .. }
658 | Self::WorkerRegister { .. }
659 | Self::WorkerRegisterAck { .. } => 0,
660 Self::Subscribe { stream_id, .. }
661 | Self::SubscribeAck { stream_id, .. }
662 | Self::SubscribeError { stream_id, .. }
663 | Self::Unsubscribe { stream_id, .. }
664 | Self::Publish { stream_id, .. }
665 | Self::PublishAck { stream_id, .. }
666 | Self::PublishError { stream_id, .. }
667 | Self::ConversationOpen { stream_id, .. }
668 | Self::ConversationMessage { stream_id, .. }
669 | Self::ConversationClose { stream_id, .. }
670 | Self::ConversationError { stream_id, .. }
671 | Self::Accept { stream_id, .. }
672 | Self::Defer { stream_id, .. }
673 | Self::Reject { stream_id, .. }
674 | Self::Push { stream_id, .. }
675 | Self::PushReply { stream_id, .. }
676 | Self::Deliver { stream_id, .. }
677 | Self::Unknown { stream_id, .. } => *stream_id,
678 }
679 }
680
681 /// Validate the stream invariant for this frame.
682 pub(crate) fn validate(&self) -> Result<(), ProtocolError> {
683 validate_stream(self.frame_type(), self.stream_id())?;
684
685 if let Self::Subscribe { max_in_flight, .. } = self {
686 if *max_in_flight == 0 {
687 return Err(ProtocolError::codec(
688 "max_in_flight must be greater than zero",
689 ));
690 }
691 }
692
693 Ok(())
694 }
695}
696
697/// Validate stream placement for a known frame type.
698///
699/// # Errors
700///
701/// Returns [`ProtocolError::InvalidStream`] when `stream_id` is invalid for
702/// `frame_type`.
703pub fn validate_stream(frame_type: FrameType, stream_id: u32) -> Result<(), ProtocolError> {
704 if matches!(frame_type, FrameType::Unknown(_)) {
705 return Ok(());
706 }
707
708 let valid = if frame_type.is_control() {
709 stream_id == 0
710 } else {
711 stream_id >= 1
712 };
713
714 if valid {
715 Ok(())
716 } else {
717 Err(ProtocolError::invalid_stream(frame_type, stream_id))
718 }
719}
720
721#[cfg(test)]
722mod tests;