carriage.go RawHistoryFolder

1package brokeripc
2
3import (
4	"github.com/bnema/vev/internal/adapters/streamframe"
5	"github.com/bnema/vev/internal/protocol/wire"
6)
7
8// carriage adapts one bounded streamPipe to the wire.Transport contract the
9// sessionwire typed connection frames over: Send writes one complete envelope
10// (4-byte big-endian length then payload) into the pipe, Recv reads one
11// complete envelope back out, and Close settles the pipe. Framing, the bounded
12// egress queue, and close-unblocks-IO behavior all come from the shared
13// streamframe implementation; this adapter owns no frame format of its own.
14type carriage struct {
15	framer *streamframe.Framer
16}
17
18var _ wire.BoundedTransport = (*carriage)(nil)
19
20// newCarriage builds one envelope carriage over a stream pipe. The pipe carries
21// opaque chunks and sessionwire applies the negotiated per-category ceilings
22// above it, so the carriage ceiling is the protocol absolute limit.
23func newCarriage(pipe *streamPipe) *carriage {
24	return &carriage{
25		framer: streamframe.NewFramer(pipe, pipe, func() error { pipe.closeWith(nil); return nil }, streamframe.WithMaxEnvelope(wire.AbsoluteEnvelopeLimit)),
26	}
27}
28
29// Send queues one envelope for the sole writer and waits for its pipe write.
30func (c *carriage) Send(envelope wire.Envelope) error { return c.framer.Send(envelope.Payload) }
31
32// Recv reads one complete envelope.
33func (c *carriage) Recv() (wire.Envelope, error) {
34	payload, err := c.framer.Recv()
35	if err != nil {
36		return wire.Envelope{}, err
37	}
38	return wire.Envelope{Payload: payload}, nil
39}
40
41// RecvBounded reads one complete envelope bounded to limit before allocation.
42func (c *carriage) RecvBounded(limit uint64) (wire.Envelope, error) {
43	payload, err := c.framer.RecvBounded(limit)
44	if err != nil {
45		return wire.Envelope{}, err
46	}
47	return wire.Envelope{Payload: payload}, nil
48}
49
50// Close settles the pipe and joins the carriage writer. It is idempotent and
51// concurrent-safe.
52func (c *carriage) Close() error { return c.framer.Close() }