adapters/tcp — Raw TCP Adapter¶

// FramedConn abstracts a framed TCP connection.
// Each ReadFrame call returns exactly one complete message;
// each WriteFrame call sends exactly one complete message.
//
// Implement this interface to support custom wire framing
// (newline-delimited, TLV, fixed-size, etc.).
// Use [NewLengthPrefixConn] for the built-in 4-byte length-prefix framing.
type FramedConn interface {
    // ReadFrame reads the next complete message frame.
    // Blocks until a frame arrives or an error occurs.
    // Returns (nil, io.EOF) when the remote end closes the connection.
    ReadFrame() ([]byte, error)

    // WriteFrame sends one complete message frame.
    WriteFrame(data []byte) error

    // SetDeadline sets the read and write deadlines (used for ctx cancellation polling).
    SetDeadline(t time.Time) error

    // Close closes the underlying connection.
    Close() error
}

// NewLengthPrefixConn wraps a net.Conn with 4-byte big-endian length-prefix framing.
// Wire format: [uint32 big-endian length][payload bytes...]
// This is the recommended default framing for new TCP services.
func NewLengthPrefixConn(conn net.Conn) FramedConn

// Serve accepts connections from listener and handles each in its own goroutine.
// For each connection:
//   1. ReadFrame → decode as Req
//   2. Call fn(ctx, req) → Resp
//   3. Encode Resp → WriteFrame
//   4. Close the connection (one request per connection model)
//
// The loop runs until ctx is cancelled (returns nil) or a listener error occurs.
// Run Serve in a dedicated goroutine.
func Serve[Req, Resp any](
    ctx context.Context,
    ln net.Listener,
    framer func(net.Conn) FramedConn, // e.g. NewLengthPrefixConn
    handle *reqreply.RouteHandle[Req, Resp],
    fn func(context.Context, Req) (Resp, error),
    opts ServeOptions,
) error

// Call dials addr, creates a FramedConn, sends the request, and reads the reply.
// The connection is closed after each call (stateless).
func Call[Req, Resp any](
    ctx context.Context,
    addr string,
    framer func(net.Conn) FramedConn,
    handle *reqreply.RouteHandle[Req, Resp],
    req Req,
    opts CallOptions,
) (Resp, error)

type ServeOptions struct {
    OnError func(ServeError)
    Observer stats.Observer
}

type CallOptions struct {
    Observer stats.Observer
    Timeout  time.Duration    // default 30s; covers dial + write + read
    Vars     map[string]string
}

// Push sends a stream of messages to addr.
// Each call to Push opens a connection, sends one message, and closes.
// For a persistent stream (many messages on one connection), see Phase 2.
func Push[T any](
    ctx context.Context,
    addr string,
    framer func(net.Conn) FramedConn,
    handle *events.ChannelHandle[T],
    msg T,
    opts PushOptions,
    formats ...format.Format[T],
) error

// Pull connects to addr and blocks, receiving frames and calling fn for each.
// The loop runs until ctx is cancelled or the connection closes.
// Run Pull in a dedicated goroutine.
func Pull[T any](
    ctx context.Context,
    addr string,
    framer func(net.Conn) FramedConn,
    handle *events.ChannelHandle[T],
    fn func(context.Context, T) error,
    opts PullOptions,
    formats ...format.Format[T],
) error

type PushOptions struct {
    Observer stats.Observer
    Timeout  time.Duration
}

type PullOptions struct {
    OnError  func(PullError)
    Observer stats.Observer
}

type ErrorKind int
const (
    KindDecode  ErrorKind = iota // payload could not be decoded
    KindHandler                  // application handler returned error
    KindEncode                   // response encoding failed
    KindTimeout                  // no reply within deadline
)

// ServeError — per-connection failure on the server side.
type ServeError struct {
    Kind       ErrorKind
    RemoteAddr string  // conn.RemoteAddr().String()
    Err        error
}
// LogValue emits: {kind, remote_addr, err}

// CallError — caller-side failure: dial, write, read, decode, timeout.
type CallError struct {
    Kind ErrorKind
    Addr string
    Err  error
}
// LogValue emits: {kind, addr, err}

// PullError — per-message failure in the Pull loop.
type PullError struct {
    Kind       ErrorKind
    RemoteAddr string
    Err        error
}
// LogValue emits: {kind, remote_addr, err}

// ConnError — connection-level failure: dial failure, accept failure, frame read/write.
// Distinct from codec-level errors (ServeError, CallError).
type ConnError struct {
    Op  string // "dial" | "accept" | "read_frame" | "write_frame"
    Err error
}
// LogValue emits: {op, err}

import (
    "net"
    tcpadapter "github.com/DaniDeer/go-codex/adapters/tcp"
    "github.com/DaniDeer/go-codex/api/reqreply"
)

var ComputeRoute = reqreply.NewRoute[ComputeReq, ComputeResp](
    "compute.add", computeReqCodec, computeRespCodec,
    reqreply.RouteMeta{OperationID: "computeAdd"},
)

b := reqreply.NewBuilder(reqreply.Info{Title: "Compute API", Version: "1.0.0"})
b.AddServer("tcp", reqreply.Server{URL: "tcp://localhost:5555", Protocol: "tcp"})
handle, _ := ComputeRoute.Register(b)

// Server
ln, _ := net.Listen("tcp", ":5555")
go tcpadapter.Serve(ctx, ln, tcpadapter.NewLengthPrefixConn, handle,
    func(ctx context.Context, req ComputeReq) (ComputeResp, error) {
        return ComputeResp{Sum: req.X + req.Y}, nil
    },
    tcpadapter.ServeOptions{Observer: obs},
)

// Client
resp, err := tcpadapter.Call(ctx, "localhost:5555",
    tcpadapter.NewLengthPrefixConn, handle,
    ComputeReq{X: 3, Y: 4},
    tcpadapter.CallOptions{Timeout: 5 * time.Second, Observer: obs},
)
// resp.Sum == 7

var SensorChannel, _ = events.NewChannel[SensorReading](
    "sensor.readings", sensorCodec,
    events.Subscribe{Summary: "Sensor readings pushed from a device."},
).Register(eventsBuilder)

// Pull — device connects and pushes; service receives
go tcpadapter.Pull(ctx, "device-gateway:6000",
    tcpadapter.NewLengthPrefixConn, SensorChannel,
    func(ctx context.Context, r SensorReading) error {
        return store.Save(ctx, r)
    },
    tcpadapter.PullOptions{Observer: obs},
)