Build: The MessageStream Codec

Maps to: Phase 6 (cluster). Kind: Build.

Objective

Add the wire to control-core. You already have the Message enum from Part II — defined, tested, but never yet sent anywhere. Here you build MessageStream: a thin wrapper over Framed<TcpStream, LengthDelimitedCodec> that offers send(&Message) and recv() -> Option<Message>, and prove it with a single test that ships a Message across a real loopback socket and gets it back byte-identical. This is the first code in the entire course that touches the network — and it is small on purpose, because LengthDelimitedCodec does the hard part.

Scaffold

Create:

• crates/control-core/src/codec.rs — the MessageStream struct, its new/send/recv, and its one test.

Edit:

• crates/control-core/src/lib.rs — add pub mod codec; and re-export MessageStream alongside the other public types.
• crates/control-core/Cargo.toml — four new dependencies.

New deps and why:

• tokio-util (features ["codec"]) — supplies Framed and LengthDelimitedCodec, the framing layer. This is the new crate the concept chapter demonstrated. ([dependencies].)
• futures — supplies the SinkExt/StreamExt extension traits whose .send() and .next() you call on a Framed. ([dependencies].)
• bytes — supplies Bytes, the type the codec's Sink accepts for an outgoing frame. ([dependencies].)
• tokio — add the net feature (for TcpStream/TcpListener) to the existing entry; macros/rt are already there from the seam build. ([dependencies], plus #[tokio::test] from dev.)

serde_json (encode/decode each frame) and pretty_assertions (the test's assert_eq!) are already in the manifest from earlier builds.

Expected result: cargo test -p control-core → 12 tests pass. The eleven from the Core arc are untouched; the new one is a_message_roundtrips_over_a_real_socket in codec.rs. The Core-arc build told you a twelfth test would arrive "when you build the framed codec that actually ships these frames" — this is that test, and the count reaches 12 now.

The spec (givens)

MessageStream

/// A bidirectional channel of `Message`s over one TCP connection.
pub struct MessageStream {
    framed: Framed<TcpStream, LengthDelimitedCodec>,
}

impl MessageStream {
    pub fn new(stream: TcpStream) -> Self { /* wrap in Framed + LengthDelimitedCodec */ }

    /// Encode a message as one length-prefixed JSON frame and send it.
    pub async fn send(&mut self, msg: &Message) -> Result<(), ControlError>;

    /// Read the next frame and decode it. `Ok(None)` means the peer closed cleanly.
    pub async fn recv(&mut self) -> Result<Option<Message>, ControlError>;
}

• new wraps the TcpStream with Framed::new(stream, LengthDelimitedCodec::new()). Nothing else — the codec is the whole configuration.
• send serializes msg with serde_json::to_vec, wraps the Vec<u8> in Bytes::from(...), and pushes it with self.framed.send(...).await. Both the serialize step and the sink step map any error to ControlError::Protocol(e.to_string()).
• recv pulls self.framed.next().await and matches three arms: Some(Ok(frame)) deserializes the frame bytes with serde_json::from_slice, mapping a serde error to ControlError::Protocol and returning Ok(Some(msg)); Some(Err(e)) (a framing/IO error) maps to Err(ControlError::Protocol(...)); None (clean end of stream) returns Ok(None).

The error mapping is the whole policy: a garbled frame or a broken link is retryable (Protocol), and a clean close is not an error at all (Ok(None)). Hold the two apart — the connection actor in the next chapter branches on exactly this difference.

→ Answer key

Concepts exercised

• Framed + LengthDelimitedCodec as the framing layer over a raw TcpStream.
• The futures::SinkExt/StreamExt extension traits behind .send()/.next().
• One JSON document per length-delimited frame via serde_json::to_vec/from_slice.
• Mapping both serde and framing failures onto the retryable ControlError::Protocol, and a clean close onto Ok(None).
• Driving two MessageStreams over a loopback TcpListener bound to 127.0.0.1:0.

The build loop (you drive)

Test — a_message_roundtrips_over_a_real_socket (in codec.rs, #[tokio::test])

1. Write the failing test. Bind a TcpListener to 127.0.0.1:0 (the OS picks a free port; read it back with local_addr()). tokio::spawn a server task that accept()s one connection, wraps it in a MessageStream, and echoes: recv().await one message and send it straight back. In the main task, TcpStream::connect(addr), wrap it in a MessageStream, send a Message::Register { worker_id: "w1".into(), capacity: 4 }, then recv and assert the echoed message equals the one you sent.
2. Predict: before you implement recv, what does the server task's recv().await return the instant the client task ends and its socket drops — Ok(Some(..)), Ok(None), or an Err? Tie your answer to the three match arms in the spec.
3. Run — it fails to compile (no MessageStream yet).
4. Implement MessageStream — new, send, recv — exactly as specified. Do not forget use futures::{SinkExt, StreamExt};, or the compiler will claim Framed has no send/next method.
5. Run green. Then run the whole suite — cargo test -p control-core — and confirm you are at 12 with the eleven Core-arc tests still green.
6. Commit.

Done when

cargo test -p control-core shows 12 green (the eleven Core-arc tests plus a_message_roundtrips_over_a_real_socket); a Message::Register survives the trip across a loopback socket and compares equal; you can say why send wraps its bytes in Bytes and why recv's clean-close arm returns Ok(None) rather than an error. Commit. control-core now has a wire — the next chapter puts a worker on the other end of it.