Add LpFrameKind::Stream variant with StreamFrameAttributes

- Define LP wire format for stream multiplexing
- Handle new variant in entry gateway match arm
This commit is contained in:
mfahampshire
2026-03-17 12:02:12 +00:00
parent 6384467526
commit 0b6166d20e
2 changed files with 64 additions and 1 deletions
+63
View File
@@ -103,6 +103,13 @@ impl LpFrame {
Self::new(LpFrameKind::Forward, data)
}
pub fn new_stream(attrs: StreamFrameAttributes, content: impl Into<Bytes>) -> Self {
Self {
header: LpFrameHeader::new(LpFrameKind::Stream, attrs.encode()),
content: content.into(),
}
}
pub(crate) fn len(&self) -> usize {
LpFrameHeader::SIZE + self.content.len()
}
@@ -115,6 +122,62 @@ pub enum LpFrameKind {
Opaque = 0,
Registration = 1,
Forward = 2,
Stream = 3,
}
/// Message type within a `LpFrameKind::Stream` frame.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum StreamMsgType {
/// Open a new stream. Content is optional initial data.
Open = 0,
/// Data on an existing stream.
Data = 1,
}
/// Parsed form of the 14-byte `frame_attributes` for `LpFrameKind::Stream`.
///
/// Wire layout (big-endian):
/// ```text
/// [0..8 ) stream_id : u64
/// [8 ) msg_type : u8 (0 = Open, 1 = Data)
/// [9..13) sequence_num : u32
/// [13 ) reserved : u8
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StreamFrameAttributes {
pub stream_id: u64,
pub msg_type: StreamMsgType,
pub sequence_num: u32,
}
impl StreamFrameAttributes {
pub fn encode(&self) -> [u8; 14] {
let mut buf = [0u8; 14];
buf[0..8].copy_from_slice(&self.stream_id.to_be_bytes());
buf[8] = self.msg_type as u8;
buf[9..13].copy_from_slice(&self.sequence_num.to_be_bytes());
buf
}
pub fn parse(attrs: &[u8; 14]) -> Result<Self, MalformedLpPacketError> {
let stream_id = u64::from_be_bytes(attrs[0..8].try_into().unwrap());
let msg_type = match attrs[8] {
0 => StreamMsgType::Open,
1 => StreamMsgType::Data,
other => {
return Err(MalformedLpPacketError::DeserialisationFailure(format!(
"invalid stream msg_type: {other}"
)));
}
};
let sequence_num = u32::from_be_bytes(attrs[9..13].try_into().unwrap());
Ok(Self {
stream_id,
msg_type,
sequence_num,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -261,7 +261,7 @@ where
self.handle_forwarding_request(receiver_idx, forward_data)
.await
}
typ @ LpFrameKind::Opaque => {
typ @ (LpFrameKind::Opaque | LpFrameKind::Stream) => {
// Neither registration nor forwarding - unknown payload type
warn!(
"Unknown transport payload type from {remote} (receiver_idx={receiver_idx}). dropping {} bytes",