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
25const U8_FIELD_LEN: usize = 1;
29
30pub 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
44pub 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
85pub 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 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 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(®istration.namespaces)?,
318 string_field_len(®istration.task_queue)?,
319 option_string_len(registration.node.as_deref())?,
320 string_vec_field_len(®istration.activity_types)?,
321 string_field_len(®istration.identity)?,
322 activity_descriptors_field_len(®istration.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 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
481fn 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
492fn 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(®istration.namespaces)?;
535 writer.write_string_field(®istration.task_queue)?;
536 writer.write_optional_string(registration.node.as_deref())?;
539 writer.write_string_vec_field(®istration.activity_types)?;
540 writer.write_string_field(®istration.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 ®istration.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
565const 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 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 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;