08c499f8ca
The MoQT transport that joule-code-collab's network rung needs lived in the private joulesperbit repo (as inv-moq + jwp), so a public consumer couldn't depend on it. Port both into the public open-standards joule-code workspace, self-contained (concrete dep versions, not workspace-inherited) so a bare git clone builds them: - crates/jwp: Joule Wire Protocol — energy-aware binary framing (CBOR, zstd/lz4, TCP + optional QUIC). Self-contained, external deps only. - crates/joule-moq: MoQT (Media-over-QUIC) wire protocol + JWP signed-receipt extension for CommunicationOS — Publisher/Subscriber/Relay traits, MockFabric (in-process) + a production QUIC path (quic feature). Renamed from inv-moq to drop the invisible-* namespace and match the joule-code branding; depends only on jwp. Both build (default + quic) and pass their tests (joule-moq 29, jwp 152) in the new home. The transport-moqt example now path-deps the public joule-moq instead of reaching into joulesperbit, and its end-to-end BFT-ops-over-MoQT-with-signed- receipts tests pass. This unblocks public consumers (agent-zed) from git-depping the transport for cross-machine fleet coordination. Co-Authored-By: Claude Opus 4.8 <[email protected]>
104 lines
3.2 KiB
Rust
104 lines
3.2 KiB
Rust
use crate::error::JwpError;
|
|
use crate::frame::{FrameHeader, HEADER_LEN, JwpFrame};
|
|
use crate::transport::{Transport, TransportStats};
|
|
|
|
/// JWP transport over QUIC bidirectional streams (quinn).
|
|
///
|
|
/// Wraps a `quinn::SendStream` + `quinn::RecvStream` pair with JWP
|
|
/// frame I/O. Each QUIC bidirectional stream maps to one JWP session.
|
|
pub struct QuicTransport {
|
|
send: quinn::SendStream,
|
|
recv: quinn::RecvStream,
|
|
stats: TransportStats,
|
|
}
|
|
|
|
impl QuicTransport {
|
|
/// Create a transport from a QUIC bidirectional stream pair.
|
|
pub fn new(send: quinn::SendStream, recv: quinn::RecvStream) -> Self {
|
|
Self {
|
|
send,
|
|
recv,
|
|
stats: TransportStats::default(),
|
|
}
|
|
}
|
|
|
|
/// Access the underlying send stream.
|
|
pub fn send_stream(&self) -> &quinn::SendStream {
|
|
&self.send
|
|
}
|
|
|
|
/// Access the underlying recv stream.
|
|
pub fn recv_stream(&self) -> &quinn::RecvStream {
|
|
&self.recv
|
|
}
|
|
}
|
|
|
|
impl Transport for QuicTransport {
|
|
async fn send_frame(&mut self, frame: JwpFrame) -> Result<(), JwpError> {
|
|
let mut header_buf = [0u8; HEADER_LEN];
|
|
frame.header.encode(&mut header_buf);
|
|
|
|
self.send
|
|
.write_all(&header_buf)
|
|
.await
|
|
.map_err(|e| JwpError::Io(std::io::Error::other(e.to_string())))?;
|
|
|
|
if !frame.payload.is_empty() {
|
|
self.send
|
|
.write_all(&frame.payload)
|
|
.await
|
|
.map_err(|e| JwpError::Io(std::io::Error::other(e.to_string())))?;
|
|
}
|
|
|
|
self.stats.bytes_sent += HEADER_LEN as u64 + frame.payload.len() as u64;
|
|
self.stats.frames_sent += 1;
|
|
Ok(())
|
|
}
|
|
|
|
async fn recv_frame(&mut self) -> Result<Option<JwpFrame>, JwpError> {
|
|
let mut header_buf = [0u8; HEADER_LEN];
|
|
match self.recv.read_exact(&mut header_buf).await {
|
|
Ok(()) => {}
|
|
Err(quinn::ReadExactError::FinishedEarly(_)) => return Ok(None),
|
|
Err(quinn::ReadExactError::ReadError(e)) => {
|
|
return Err(JwpError::Io(std::io::Error::other(e.to_string())));
|
|
}
|
|
}
|
|
|
|
let header = FrameHeader::decode(&header_buf)?;
|
|
let payload_len = header.payload_length as usize;
|
|
|
|
let mut payload = vec![0u8; payload_len];
|
|
if payload_len > 0 {
|
|
match self.recv.read_exact(&mut payload).await {
|
|
Ok(()) => {}
|
|
Err(quinn::ReadExactError::FinishedEarly(n)) => {
|
|
return Err(JwpError::IncompleteFrame {
|
|
needed: payload_len,
|
|
available: n,
|
|
});
|
|
}
|
|
Err(quinn::ReadExactError::ReadError(e)) => {
|
|
return Err(JwpError::Io(std::io::Error::other(e.to_string())));
|
|
}
|
|
}
|
|
}
|
|
|
|
self.stats.bytes_received += HEADER_LEN as u64 + payload_len as u64;
|
|
self.stats.frames_received += 1;
|
|
|
|
Ok(Some(JwpFrame { header, payload }))
|
|
}
|
|
|
|
fn transport_id(&self) -> &str {
|
|
"quic"
|
|
}
|
|
|
|
fn stats(&self) -> &TransportStats {
|
|
&self.stats
|
|
}
|
|
|
|
fn stats_mut(&mut self) -> &mut TransportStats {
|
|
&mut self.stats
|
|
}
|
|
}
|