Skip to main content

liminal_bus_protocol/
codec.rs

1mod known;
2mod payload;
3
4#[cfg(test)]
5mod tests_support;
6
7use super::causal::MessageId;
8use super::envelope::{MessageEnvelope, SchemaId};
9use super::error::ProtocolError;
10use super::frame::{
11    DISCONNECT_REASON_BODY_VERSION, DeliveredId, Frame, FrameType, HEADER_LEN, LINK_END_EXPORT,
12    LINK_END_IMPORT, LINK_MACHINE_LEN, LINK_OPEN_ACK_HELD, LINK_OPEN_ACK_OPENED,
13    LINK_OPEN_ACK_REFUSED, LinkEnd, LinkOffer, LinkOpenOutcome, WORKER_REGISTER_ACK_ACCEPTED,
14    WORKER_REGISTER_ACK_REJECTED, WorkerActivityDescriptor, WorkerRegisterOutcome,
15    WorkerRegistration, validate_stream,
16};
17use super::version::ProtocolVersion;
18use known::decode_known_payload;
19use payload::{
20    PayloadReader, PayloadWriter, U16_LEN, U32_LEN, U64_LEN, bytes_field_len, checked_u32_len,
21    option_string_len, option_u16_len, schema_ids_field_len, string_field_len,
22    string_vec_field_len, sum_lengths,
23};
24
25/// Wire length of the one-byte fields the worker-registration frames use (the
26/// optional-node presence byte folds into [`option_string_len`]; this covers the
27/// ack's status byte).
28const U8_FIELD_LEN: usize = 1;
29
30/// Return the number of bytes needed to encode a frame.
31///
32/// # Errors
33///
34/// Returns [`ProtocolError`] when the frame violates stream invariants or its
35/// payload cannot fit in the protocol's `u32` length fields.
36pub fn encoded_len(frame: &Frame) -> Result<usize, ProtocolError> {
37    frame.validate()?;
38    let payload_len = encoded_payload_len(frame)?;
39    HEADER_LEN
40        .checked_add(payload_len)
41        .ok_or_else(|| ProtocolError::codec("encoded frame length overflowed usize"))
42}
43
44/// Encode a frame into the provided byte buffer, returning bytes written.
45///
46/// The buffer must be at least [`encoded_len`] bytes long. Encoding writes the
47/// fixed 10-byte header followed by the serialized payload and performs no heap
48/// allocation.
49///
50/// # Errors
51///
52/// Returns [`ProtocolError`] when the frame violates stream invariants, its
53/// payload cannot fit in the protocol's length fields, or the provided buffer is
54/// too small.
55pub fn encode(frame: &Frame, buffer: &mut [u8]) -> Result<usize, ProtocolError> {
56    frame.validate()?;
57    let payload_len = encoded_payload_len(frame)?;
58    let payload_length = u32::try_from(payload_len)
59        .map_err(|_| ProtocolError::codec("payload length exceeded u32::MAX"))?;
60    let total_len = HEADER_LEN
61        .checked_add(payload_len)
62        .ok_or_else(|| ProtocolError::codec("encoded frame length overflowed usize"))?;
63
64    if buffer.len() < total_len {
65        return Err(ProtocolError::codec("output buffer is too small"));
66    }
67
68    let Some(header) = buffer.get_mut(..HEADER_LEN) else {
69        return Err(ProtocolError::codec(
70            "output buffer is too small for header",
71        ));
72    };
73    write_header(frame, payload_length, header)?;
74
75    let Some(payload) = buffer.get_mut(HEADER_LEN..total_len) else {
76        return Err(ProtocolError::codec(
77            "output buffer is too small for payload",
78        ));
79    };
80    write_payload(frame, payload)?;
81
82    Ok(total_len)
83}
84
85/// Decode one complete frame from a byte buffer.
86///
87/// Returns the decoded frame and the number of bytes consumed. Unknown frame
88/// types are length-delimited and returned as [`Frame::Unknown`] without
89/// producing an error.
90///
91/// # Errors
92///
93/// Returns [`ProtocolError::IncompleteHeader`] for buffers shorter than the
94/// fixed header, [`ProtocolError::TruncatedPayload`] when the declared payload
95/// is not fully present, and [`ProtocolError`] for malformed known-frame
96/// payloads or invalid stream placement.
97pub fn decode(buffer: &[u8]) -> Result<(Frame, usize), ProtocolError> {
98    if buffer.len() < HEADER_LEN {
99        return Err(ProtocolError::IncompleteHeader {
100            message: Some("buffer shorter than fixed frame header".to_owned()),
101        });
102    }
103
104    let Some(header) = buffer.get(..HEADER_LEN) else {
105        return Err(ProtocolError::IncompleteHeader {
106            message: Some("buffer shorter than fixed frame header".to_owned()),
107        });
108    };
109    let mut header_reader = PayloadReader::new(header);
110    let type_id = header_reader.read_u8()?;
111    let flags = header_reader.read_u8()?;
112    let stream_id = header_reader.read_u32()?;
113    let payload_length = header_reader.read_u32()?;
114    header_reader.finish()?;
115
116    let payload_len = usize::try_from(payload_length)
117        .map_err(|_| ProtocolError::codec("payload length cannot fit usize"))?;
118    let total_len = HEADER_LEN
119        .checked_add(payload_len)
120        .ok_or_else(|| ProtocolError::codec("decoded frame length overflowed usize"))?;
121
122    if buffer.len() < total_len {
123        return Err(ProtocolError::TruncatedPayload {
124            message: Some("buffer shorter than declared payload length".to_owned()),
125        });
126    }
127
128    let Some(payload) = buffer.get(HEADER_LEN..total_len) else {
129        return Err(ProtocolError::TruncatedPayload {
130            message: Some("buffer shorter than declared payload length".to_owned()),
131        });
132    };
133
134    let frame_type = FrameType::from(type_id);
135    let frame = decode_payload(frame_type, flags, stream_id, payload)?;
136    Ok((frame, total_len))
137}
138
139fn write_header(
140    frame: &Frame,
141    payload_length: u32,
142    buffer: &mut [u8],
143) -> Result<(), ProtocolError> {
144    let mut writer = PayloadWriter::new(buffer);
145    writer.write_u8(u8::from(frame.frame_type()))?;
146    writer.write_u8(frame.wire_flags())?;
147    writer.write_u32(frame.stream_id())?;
148    writer.write_u32(payload_length)?;
149    writer.finish()
150}
151
152fn encoded_payload_len(frame: &Frame) -> Result<usize, ProtocolError> {
153    match frame {
154        Frame::Connect { auth_token, .. } => sum_lengths(&[
155            ProtocolVersion::WIRE_LEN,
156            ProtocolVersion::WIRE_LEN,
157            bytes_field_len(auth_token)?,
158        ]),
159        Frame::ConnectAck { .. } => sum_lengths(&[ProtocolVersion::WIRE_LEN, U32_LEN]),
160        Frame::ConnectError { message, .. }
161        | Frame::SubscribeError { message, .. }
162        | Frame::PublishError { message, .. } => {
163            sum_lengths(&[U16_LEN, option_string_len(message.as_deref())?])
164        }
165        Frame::Disconnect { .. } | Frame::Ping { .. } | Frame::Pong { .. } => Ok(0),
166        Frame::DisconnectReason { reason, .. } => {
167            sum_lengths(&[U8_FIELD_LEN, U16_LEN, string_field_len(reason.name())?])
168        }
169        Frame::Subscribe {
170            channel,
171            accepted_schemas,
172            ..
173        } => sum_lengths(&[
174            string_field_len(channel)?,
175            schema_ids_field_len(accepted_schemas)?,
176            U32_LEN,
177        ]),
178        Frame::SubscribeAck { .. } => sum_lengths(&[U64_LEN, SchemaId::WIRE_LEN]),
179        Frame::Unsubscribe { .. } | Frame::DeliverAck { .. } => Ok(U64_LEN),
180        Frame::PublishAck {
181            durable_position,
182            boot,
183            ..
184        } => sum_lengths(&[
185            U64_LEN,
186            durable_position.map_or(0, |_| U64_LEN),
187            boot.map_or(0, |_| U64_LEN),
188        ]),
189        Frame::Publish {
190            channel,
191            envelope,
192            idempotency_key,
193            ..
194        } => {
195            let mut parts = vec![
196                string_field_len(channel)?,
197                envelope_bytes_field_len(envelope.encoded_len()?)?,
198            ];
199            if let Some(key) = idempotency_key {
200                parts.push(string_field_len(key)?);
201            }
202            sum_lengths(&parts)
203        }
204        Frame::ConversationOpen { .. }
205        | Frame::ConversationMessage { .. }
206        | Frame::ConversationClose { .. }
207        | Frame::ConversationError { .. } => conversation_payload_len(frame),
208        // Conversation-message and delivery frames share a wire shape: a `u64`
209        // prefix (conversation id / delivery seq) followed by a length-prefixed
210        // envelope. A delivery carrying its message id adds the boot and the id.
211        Frame::Deliver {
212            envelope,
213            message,
214            position,
215            ..
216        } => sum_lengths(&[
217            U64_LEN,
218            envelope_bytes_field_len(envelope.encoded_len()?)?,
219            message.map_or(0, |_| U64_LEN + U64_LEN),
220            position.map_or(0, |_| U64_LEN),
221        ]),
222        Frame::LinkOpen { offer, .. } => link_open_payload_len(offer),
223        Frame::LinkOpenAck { outcome, .. } => link_open_ack_payload_len(outcome),
224        Frame::LinkMessage {
225            subject, sealed, ..
226        } => sum_lengths(&[string_field_len(subject)?, bytes_field_len(sealed)?]),
227        Frame::Accept {
228            referenced_message_id,
229            ..
230        } => message_id_field_len(referenced_message_id),
231        Frame::Defer {
232            referenced_message_id,
233            reason,
234            ..
235        }
236        | Frame::Reject {
237            referenced_message_id,
238            reason,
239            ..
240        } => sum_lengths(&[
241            message_id_field_len(referenced_message_id)?,
242            option_string_len(reason.as_deref())?,
243        ]),
244        Frame::Push { payload, .. } | Frame::PushReply { payload, .. } => {
245            sum_lengths(&[U64_LEN, bytes_field_len(payload)?])
246        }
247        Frame::WorkerRegister { registration, .. } => worker_register_payload_len(registration),
248        Frame::WorkerRegisterAck { outcome, .. } => worker_register_ack_payload_len(outcome),
249        Frame::Unknown { payload, .. } => checked_u32_len(payload.len()).map(|()| payload.len()),
250    }
251}
252
253fn envelope_bytes_field_len(envelope_len: usize) -> Result<usize, ProtocolError> {
254    checked_u32_len(envelope_len)?;
255    sum_lengths(&[U32_LEN, envelope_len])
256}
257
258fn message_id_field_len(message_id: &MessageId) -> Result<usize, ProtocolError> {
259    string_field_len(message_id.as_str())
260}
261
262fn conversation_payload_len(frame: &Frame) -> Result<usize, ProtocolError> {
263    match frame {
264        Frame::ConversationOpen { subject, .. } => {
265            sum_lengths(&[U64_LEN, string_field_len(subject)?])
266        }
267        // A conversation message shares the delivery frame's wire shape: a
268        // `u64` conversation id followed by a length-prefixed envelope.
269        Frame::ConversationMessage { envelope, .. } => {
270            sum_lengths(&[U64_LEN, envelope_bytes_field_len(envelope.encoded_len()?)?])
271        }
272        Frame::ConversationClose {
273            reason_code,
274            message,
275            ..
276        } => sum_lengths(&[
277            U64_LEN,
278            option_u16_len(*reason_code),
279            option_string_len(message.as_deref())?,
280        ]),
281        Frame::ConversationError { message, .. } => {
282            sum_lengths(&[U64_LEN, U16_LEN, option_string_len(message.as_deref())?])
283        }
284        _ => Err(ProtocolError::codec(
285            "frame type was not a conversation frame",
286        )),
287    }
288}
289
290fn link_open_payload_len(offer: &LinkOffer) -> Result<usize, ProtocolError> {
291    sum_lengths(&[
292        string_field_len(&offer.link_id)?,
293        U8_FIELD_LEN,
294        LINK_MACHINE_LEN,
295        bytes_field_len(&offer.handshake)?,
296    ])
297}
298
299fn link_open_ack_payload_len(outcome: &LinkOpenOutcome) -> Result<usize, ProtocolError> {
300    match outcome {
301        LinkOpenOutcome::Opened { handshake } => {
302            sum_lengths(&[U8_FIELD_LEN, bytes_field_len(handshake)?])
303        }
304        LinkOpenOutcome::Held { grant, draft } => sum_lengths(&[
305            U8_FIELD_LEN,
306            string_field_len(grant)?,
307            string_field_len(draft)?,
308        ]),
309        LinkOpenOutcome::Refused { reason } => {
310            sum_lengths(&[U8_FIELD_LEN, string_field_len(reason)?])
311        }
312    }
313}
314
315fn worker_register_payload_len(registration: &WorkerRegistration) -> Result<usize, ProtocolError> {
316    sum_lengths(&[
317        string_vec_field_len(&registration.namespaces)?,
318        string_field_len(&registration.task_queue)?,
319        option_string_len(registration.node.as_deref())?,
320        string_vec_field_len(&registration.activity_types)?,
321        string_field_len(&registration.identity)?,
322        activity_descriptors_field_len(&registration.activities)?,
323    ])
324}
325
326fn activity_descriptors_field_len(
327    activities: &[WorkerActivityDescriptor],
328) -> Result<usize, ProtocolError> {
329    checked_u32_len(activities.len())?;
330    let mut total = U32_LEN;
331    for activity in activities {
332        total = sum_lengths(&[
333            total,
334            string_field_len(&activity.name)?,
335            string_field_len(&activity.input_schema_json)?,
336            string_field_len(&activity.output_schema_json)?,
337        ])?;
338    }
339    Ok(total)
340}
341
342fn worker_register_ack_payload_len(
343    outcome: &WorkerRegisterOutcome,
344) -> Result<usize, ProtocolError> {
345    match outcome {
346        WorkerRegisterOutcome::Accepted => Ok(U8_FIELD_LEN),
347        WorkerRegisterOutcome::Rejected { reason } => {
348            sum_lengths(&[U8_FIELD_LEN, string_field_len(reason)?])
349        }
350    }
351}
352
353fn write_handshake_payload(
354    frame: &Frame,
355    writer: &mut PayloadWriter<'_>,
356) -> Result<(), ProtocolError> {
357    match frame {
358        Frame::Connect {
359            min_version,
360            max_version,
361            auth_token,
362            ..
363        } => {
364            writer.write_slice(&min_version.to_wire_bytes())?;
365            writer.write_slice(&max_version.to_wire_bytes())?;
366            writer.write_bytes_field(auth_token)
367        }
368        Frame::ConnectAck {
369            selected_version,
370            capabilities,
371            ..
372        } => {
373            writer.write_slice(&selected_version.to_wire_bytes())?;
374            writer.write_u32(*capabilities)
375        }
376        _ => Err(ProtocolError::codec("frame type was not a handshake frame")),
377    }
378}
379
380fn write_pressure_payload(
381    frame: &Frame,
382    writer: &mut PayloadWriter<'_>,
383) -> Result<(), ProtocolError> {
384    match frame {
385        Frame::Accept {
386            referenced_message_id,
387            ..
388        } => writer.write_string_field(referenced_message_id.as_str()),
389        Frame::Defer {
390            referenced_message_id,
391            reason,
392            ..
393        }
394        | Frame::Reject {
395            referenced_message_id,
396            reason,
397            ..
398        } => {
399            writer.write_string_field(referenced_message_id.as_str())?;
400            writer.write_optional_string(reason.as_deref())
401        }
402        _ => Err(ProtocolError::codec("frame type was not a pressure frame")),
403    }
404}
405
406fn write_publish(frame: &Frame, writer: &mut PayloadWriter<'_>) -> Result<(), ProtocolError> {
407    match frame {
408        Frame::Publish {
409            channel,
410            envelope,
411            idempotency_key,
412            ..
413        } => {
414            writer.write_string_field(channel)?;
415            writer.write_bytes_field(&envelope.serialize()?)?;
416            // The trailing key is written ONLY when present (its flag bit tells the
417            // decoder), so a no-key publish stays byte-identical to pre-13-L1.
418            if let Some(key) = idempotency_key {
419                writer.write_string_field(key)?;
420            }
421            Ok(())
422        }
423        Frame::PublishAck {
424            message_id,
425            durable_position,
426            boot,
427            ..
428        } => {
429            writer.write_u64(*message_id)?;
430            writer.write_flagged_u64(*durable_position)?;
431            writer.write_flagged_u64(*boot)
432        }
433        _ => Err(ProtocolError::codec("frame type was not a publish frame")),
434    }
435}
436
437fn write_conversation_payload(
438    frame: &Frame,
439    writer: &mut PayloadWriter<'_>,
440) -> Result<(), ProtocolError> {
441    match frame {
442        Frame::ConversationOpen {
443            conversation_id,
444            subject,
445            ..
446        } => {
447            writer.write_u64(*conversation_id)?;
448            writer.write_string_field(subject)
449        }
450        Frame::ConversationMessage {
451            conversation_id,
452            envelope,
453            ..
454        } => write_u64_prefixed_envelope(writer, *conversation_id, envelope),
455        Frame::ConversationClose {
456            conversation_id,
457            reason_code,
458            message,
459            ..
460        } => {
461            writer.write_u64(*conversation_id)?;
462            writer.write_optional_u16(*reason_code)?;
463            writer.write_optional_string(message.as_deref())
464        }
465        Frame::ConversationError {
466            conversation_id,
467            reason_code,
468            message,
469            ..
470        } => {
471            writer.write_u64(*conversation_id)?;
472            writer.write_u16(*reason_code)?;
473            writer.write_optional_string(message.as_deref())
474        }
475        _ => Err(ProtocolError::codec(
476            "frame type was not a conversation frame",
477        )),
478    }
479}
480
481/// Writes a `u64` prefix followed by a length-prefixed serialized envelope — the
482/// shared body of conversation-message and delivery frames.
483fn write_u64_prefixed_envelope(
484    writer: &mut PayloadWriter<'_>,
485    value: u64,
486    envelope: &MessageEnvelope,
487) -> Result<(), ProtocolError> {
488    writer.write_u64(value)?;
489    writer.write_bytes_field(&envelope.serialize()?)
490}
491
492/// Writes a delivery's body: the `u64` delivery sequence and the envelope, then
493/// the boot and the id ONLY when present (its flag bit tells the decoder), so a
494/// Deliver without them stays byte-identical to 1.1.
495fn write_deliver_payload(
496    writer: &mut PayloadWriter<'_>,
497    delivery_seq: u64,
498    envelope: &MessageEnvelope,
499    message: Option<DeliveredId>,
500    position: Option<u64>,
501) -> Result<(), ProtocolError> {
502    write_u64_prefixed_envelope(writer, delivery_seq, envelope)?;
503    if let Some(DeliveredId { boot, id }) = message {
504        writer.write_u64(boot)?;
505        writer.write_u64(id)?;
506    }
507    writer.write_flagged_u64(position)?;
508    Ok(())
509}
510
511fn write_push_payload(frame: &Frame, writer: &mut PayloadWriter<'_>) -> Result<(), ProtocolError> {
512    match frame {
513        Frame::Push {
514            correlation_id,
515            payload,
516            ..
517        }
518        | Frame::PushReply {
519            correlation_id,
520            payload,
521            ..
522        } => {
523            writer.write_u64(*correlation_id)?;
524            writer.write_bytes_field(payload)
525        }
526        _ => Err(ProtocolError::codec("frame type was not a push frame")),
527    }
528}
529
530fn write_worker_register_payload(
531    registration: &WorkerRegistration,
532    writer: &mut PayloadWriter<'_>,
533) -> Result<(), ProtocolError> {
534    writer.write_string_vec_field(&registration.namespaces)?;
535    writer.write_string_field(&registration.task_queue)?;
536    // `node` is optional locality: a presence byte distinguishes `None` from
537    // `Some("")` so an absent node never collapses to an empty string.
538    writer.write_optional_string(registration.node.as_deref())?;
539    writer.write_string_vec_field(&registration.activity_types)?;
540    writer.write_string_field(&registration.identity)?;
541    let count = u32::try_from(registration.activities.len())
542        .map_err(|_| ProtocolError::codec("activity descriptor count exceeded u32::MAX"))?;
543    writer.write_u32(count)?;
544    for activity in &registration.activities {
545        writer.write_string_field(&activity.name)?;
546        writer.write_string_field(&activity.input_schema_json)?;
547        writer.write_string_field(&activity.output_schema_json)?;
548    }
549    Ok(())
550}
551
552fn write_worker_register_ack_payload(
553    outcome: &WorkerRegisterOutcome,
554    writer: &mut PayloadWriter<'_>,
555) -> Result<(), ProtocolError> {
556    match outcome {
557        WorkerRegisterOutcome::Accepted => writer.write_u8(WORKER_REGISTER_ACK_ACCEPTED),
558        WorkerRegisterOutcome::Rejected { reason } => {
559            writer.write_u8(WORKER_REGISTER_ACK_REJECTED)?;
560            writer.write_string_field(reason)
561        }
562    }
563}
564
565/// The wire byte of a link end.
566const fn link_end_byte(end: LinkEnd) -> u8 {
567    match end {
568        LinkEnd::Export => LINK_END_EXPORT,
569        LinkEnd::Import => LINK_END_IMPORT,
570    }
571}
572
573fn write_link_open_payload(
574    offer: &LinkOffer,
575    writer: &mut PayloadWriter<'_>,
576) -> Result<(), ProtocolError> {
577    writer.write_string_field(&offer.link_id)?;
578    writer.write_u8(link_end_byte(offer.end))?;
579    // The machine key is fixed-length: raw, no length prefix.
580    writer.write_slice(&offer.machine)?;
581    writer.write_bytes_field(&offer.handshake)
582}
583
584fn write_link_open_ack_payload(
585    outcome: &LinkOpenOutcome,
586    writer: &mut PayloadWriter<'_>,
587) -> Result<(), ProtocolError> {
588    match outcome {
589        LinkOpenOutcome::Opened { handshake } => {
590            writer.write_u8(LINK_OPEN_ACK_OPENED)?;
591            writer.write_bytes_field(handshake)
592        }
593        LinkOpenOutcome::Held { grant, draft } => {
594            writer.write_u8(LINK_OPEN_ACK_HELD)?;
595            writer.write_string_field(grant)?;
596            writer.write_string_field(draft)
597        }
598        LinkOpenOutcome::Refused { reason } => {
599            writer.write_u8(LINK_OPEN_ACK_REFUSED)?;
600            writer.write_string_field(reason)
601        }
602    }
603}
604
605fn write_payload(frame: &Frame, buffer: &mut [u8]) -> Result<(), ProtocolError> {
606    let mut writer = PayloadWriter::new(buffer);
607    match frame {
608        Frame::Connect { .. } | Frame::ConnectAck { .. } => {
609            write_handshake_payload(frame, &mut writer)?;
610        }
611        Frame::ConnectError {
612            reason_code,
613            message,
614            ..
615        }
616        | Frame::SubscribeError {
617            reason_code,
618            message,
619            ..
620        }
621        | Frame::PublishError {
622            reason_code,
623            message,
624            ..
625        } => {
626            writer.write_u16(*reason_code)?;
627            writer.write_optional_string(message.as_deref())?;
628        }
629        Frame::Disconnect { .. } | Frame::Ping { .. } | Frame::Pong { .. } => {}
630        Frame::DisconnectReason { reason, .. } => {
631            // ACCESS-005 R5: body version, then the reason's code and name.
632            // The typed reason makes a code/name disagreement unwritable.
633            writer.write_u8(DISCONNECT_REASON_BODY_VERSION)?;
634            writer.write_u16(reason.code())?;
635            writer.write_string_field(reason.name())?;
636        }
637        Frame::Subscribe {
638            channel,
639            accepted_schemas,
640            max_in_flight,
641            ..
642        } => {
643            writer.write_string_field(channel)?;
644            writer.write_schema_ids_field(accepted_schemas)?;
645            writer.write_u32(*max_in_flight)?;
646        }
647        Frame::SubscribeAck {
648            subscription_id,
649            selected_schema,
650            ..
651        } => {
652            writer.write_u64(*subscription_id)?;
653            writer.write_schema_id(*selected_schema)?;
654        }
655        Frame::DeliverAck { position, .. } => writer.write_u64(*position)?,
656        Frame::Unsubscribe {
657            subscription_id, ..
658        } => writer.write_u64(*subscription_id)?,
659        Frame::Publish { .. } | Frame::PublishAck { .. } => write_publish(frame, &mut writer)?,
660        Frame::ConversationOpen { .. }
661        | Frame::ConversationMessage { .. }
662        | Frame::ConversationClose { .. }
663        | Frame::ConversationError { .. } => write_conversation_payload(frame, &mut writer)?,
664        Frame::Accept { .. } | Frame::Defer { .. } | Frame::Reject { .. } => {
665            write_pressure_payload(frame, &mut writer)?;
666        }
667        Frame::Push { .. } | Frame::PushReply { .. } => {
668            write_push_payload(frame, &mut writer)?;
669        }
670        Frame::Deliver {
671            delivery_seq,
672            envelope,
673            message,
674            position,
675            ..
676        } => write_deliver_payload(&mut writer, *delivery_seq, envelope, *message, *position)?,
677        Frame::WorkerRegister { registration, .. } => {
678            write_worker_register_payload(registration, &mut writer)?;
679        }
680        Frame::WorkerRegisterAck { outcome, .. } => {
681            write_worker_register_ack_payload(outcome, &mut writer)?;
682        }
683        Frame::LinkOpen { offer, .. } => write_link_open_payload(offer, &mut writer)?,
684        Frame::LinkOpenAck { outcome, .. } => write_link_open_ack_payload(outcome, &mut writer)?,
685        Frame::LinkMessage {
686            subject, sealed, ..
687        } => {
688            writer.write_string_field(subject)?;
689            writer.write_bytes_field(sealed)?;
690        }
691        Frame::Unknown { payload, .. } => writer.write_slice(payload)?,
692    }
693    writer.finish()
694}
695
696fn decode_payload(
697    frame_type: FrameType,
698    flags: u8,
699    stream_id: u32,
700    payload: &[u8],
701) -> Result<Frame, ProtocolError> {
702    if let FrameType::Unknown(type_id) = frame_type {
703        return Ok(Frame::Unknown {
704            type_id,
705            flags,
706            stream_id,
707            payload: payload.to_vec(),
708        });
709    }
710
711    validate_stream(frame_type, stream_id)?;
712    decode_known_payload(frame_type, flags, stream_id, payload)
713}
714
715#[cfg(test)]
716mod tests;