Typed Pipeline

The typed pipeline is rs-netty’s central constraint mechanism. It puts stage order and message transitions into builder type parameters so many mistakes become compile-time errors.

Shape

TCP:

#![allow(unused)]
fn main() {
pipeline()
    .codec(...)
    .inbound(...)*
    .business(...)*
    .handler(...)
    .outbound(...)*
}

UDP:

#![allow(unused)]
fn main() {
datagram_pipeline()
    .codec(...)
    .inbound(...)*
    .business(...)*
    .handler(...)
    .outbound(...)*
}

The two builders have the same stage shape, but different codec traits and final handler traits.

Builder States

Shared state markers live in pipeline/core/state.rs:

• Start: initial state; only a codec can be added.
• InboundPhase: a codec exists; inbound, business, or final handler can be added.
• BusinessPhase: business processing has started; more business stages or the final handler can be added.
• Ready: the final handler exists; outbound stages can be added and the builder can become a runtime pipeline.

These states appear as the first type parameter of PipelineBuilder<State, C, InP, BizP, H, OutP, CurrentIn, Write, CurrentOut> and DatagramPipelineBuilder<...>. If a stage is illegal in the current state, the method is simply not implemented for that builder type.

Message Type Chain

For TCP, the constraints are:

C: Decoder<Item = A>
InboundPipe<A, Out = B>
BusinessPipe<B, Out = CIn>
H: Handler<CIn, Write = W>
OutboundPipe<W, Out = COut>
codec: Encoder<COut>

This means:

• the first inbound stage must accept the type decoded by the codec.
• each following inbound/business stage must accept the previous stage’s Out.
• the final handler input must match the inbound/business chain output.
• the first outbound stage must accept Handler::Write.
• the final outbound output must be encodable by the stream codec.

UDP uses the same type chain, but swaps in DatagramDecoder / DatagramEncoder and DatagramHandler.

Compile Failures Are Intentional API

The trybuild compile-fail tests document these constraints. This pipeline does not compile because Parse converts String into Request, but the final handler expects String:

#![allow(unused)]
fn main() {
let _ = pipeline()
    .codec(LineCodec::new())
    .inbound(Parse)
    .handler(EchoString);
}

This one also fails because LineCodec can encode String, while the handler writes Response and no outbound stage converts it:

#![allow(unused)]
fn main() {
let _ = TcpServer::bind("127.0.0.1:0")
    .pipeline(|| pipeline().codec(LineCodec::new()).handler(Router));
}

The fix is to add an outbound stage:

#![allow(unused)]
fn main() {
let _ = pipeline()
    .codec(LineCodec::new())
    .inbound(Parse)
    .handler(Router)
    .outbound(RenderResponse);
}

Flow

Inbound, Business, and Outbound return Result<Flow<T>>:

#![allow(unused)]
fn main() {
pub enum Flow<T> {
    Next(T),
    Stop,
}
}

Flow::Next continues the pipeline. Flow::Stop consumes the current message and stops processing in that direction without error. Final Handler / DatagramHandler implementations do not return Flow; they return Result<()> because they are the end of the inbound side.