reqreply

package
v0.11.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 15, 2026 License: MIT Imports: 9 Imported by: 0

Documentation

Overview

Package reqreply provides a transport-agnostic request-reply API layer for async transports (ZeroMQ, MQTT 5, AMQP RPC, etc.).

It follows the same declare → register → handle pattern as api/events and api/rest. Route is the reqreply analogue of [rest.Route]: a typed request-reply declaration with a topic/address instead of an HTTP method+path.

The protocol is just a server string in Builder.AddServer — the same Route declaration works for any transport. Adapters accept *RouteHandle directly.

Usage 

// Declare once — no HTTP method, just a topic.
var ComputeRoute = reqreply.NewRoute[ComputeReq, ComputeResp](
    "compute/add",
    computeReqCodec, computeRespCodec,
    reqreply.RouteMeta{OperationID: "computeAdd", Summary: "Add two integers."},
)

// Register with a Builder to get a RouteHandle and an AsyncAPI 3.0 spec.
builder := reqreply.NewBuilder(reqreply.Info{Title: "Compute API", Version: "1.0.0"})
builder.AddServer("zmq", reqreply.Server{URL: "tcp://localhost:5556", Protocol: "zmq"})
// OR: builder.AddServer("mqtt5", reqreply.Server{URL: "mqtt://broker:1883", Protocol: "mqtt5"})
handle, err := ComputeRoute.Register(builder)

// Same handle — works with any request-reply adapter:
zmqadapter.Serve(ctx, sock, handle, fn, zmqadapter.ServeOptions{Observer: obs})
mqtt5adapter.ServeRequestReply(ctx, client, router, handle, fn, mqtt5.ServeOptions{Observer: obs})

// AsyncAPI 3.0 spec with request-reply reply: block:
doc, _ := builder.AsyncAPISpec()
yaml, _ := doc.MarshalYAML()

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Builder 

type Builder struct {
	// contains filtered or unexported fields
}

Builder accumulates Route registrations and produces an AsyncAPI 3.0 document with request-reply operations.

Create a Builder with NewBuilder, add servers via [AddServer], register routes via Route.Register, and call [AsyncAPISpec] to produce the document.

func NewBuilder 

func NewBuilder(info Info) *Builder

NewBuilder returns a Builder initialised with the given Info.

func (*Builder) AddServer 

func (b *Builder) AddServer(name string, s Server) *Builder

AddServer registers a named server in the AsyncAPI document. Servers appear in output in registration order.

Use Protocol: "zmq" for ZeroMQ servers, "mqtt5" for MQTT 5.0, etc.

func (*Builder) AppendTo 

func (b *Builder) AppendTo(db *asyncapi.DocumentBuilder) error

AppendTo writes all request-reply channels registered on this Builder into db. Servers and schemas owned by this Builder are NOT written — the caller is responsible for configuring those on db.

Use AppendTo to combine request-reply channels with pub/sub channels from api/events.Builder in a single AsyncAPI 3.0 document: 

import asyncapi "github.com/DaniDeer/go-codex/render/asyncapi/v3"

doc := asyncapi.NewDocumentBuilder(info)
doc.AddServer("mqtt5", asyncapi.Server{URL: "mqtts://...", Protocol: "mqtt5"})

eventsB.AppendTo(doc)    // pub/sub channels
reqreplyB.AppendTo(doc)  // request-reply channels

spec, err := doc.Build()

func (*Builder) AsyncAPISpec 

func (b *Builder) AsyncAPISpec() (asyncapi.Document, error)

AsyncAPISpec builds and returns the accumulated AsyncAPI 3.0 document. Returns an error if any registered channel is invalid.

type DuplicateRouteError 

type DuplicateRouteError struct {
	// Topic is the topic that was registered more than once.
	Topic string
}

DuplicateRouteError is returned by Route.Register when a route with the same topic has already been registered with the Builder. 

var dup reqreply.DuplicateRouteError
if errors.As(err, &dup) {
    slog.Error("duplicate route", "topic", dup.Topic)
}

func (DuplicateRouteError) Error 

func (e DuplicateRouteError) Error() string

func (DuplicateRouteError) LogValue 

func (e DuplicateRouteError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

type Info 

type Info = asyncapi.Info

Info is an alias for asyncapi.Info. Using the alias avoids duplicating fields and keeps the two in sync automatically.

type MissingRouteParamError 

type MissingRouteParamError struct {
	// Name is the {varName} placeholder that was missing from vars.
	Name string
}

MissingRouteParamError is returned by RouteHandle.BuildTopic and RouteHandle.ValidateTopicVars when a required topic variable is absent from the vars map. It mirrors [events.MissingTopicVarError]. 

var missing reqreply.MissingRouteParamError
if errors.As(err, &missing) {
    slog.Warn("missing topic var", "name", missing.Name)
}

func (MissingRouteParamError) Error 

func (e MissingRouteParamError) Error() string

func (MissingRouteParamError) LogValue 

func (e MissingRouteParamError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

type Route 

type Route[Req, Resp any] struct {
	// contains filtered or unexported fields
}

Route[Req,Resp] is a typed request-reply route for async transports (ZeroMQ, MQTT 5, AMQP, etc.). It is the api/reqreply analogue of [rest.Route], which is for HTTP. The key difference is that a Route has a topic/address instead of an HTTP method and path.

NewRoute is infallible — it only captures the spec. Validation runs at Route.Register time.

Typical usage: 

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

// Register with a builder to get an AsyncAPI spec + a RouteHandle:
builder := reqreply.NewBuilder(reqreply.Info{Title: "API", Version: "1.0.0"})
builder.AddServer("zmq", reqreply.Server{URL: "tcp://...", Protocol: "zmq"})
handle, err := ComputeRoute.Register(builder)

// Adapters accept *reqreply.RouteHandle:
zmqadapter.Serve(ctx, sock, handle, fn, zmqadapter.ServeOptions{Observer: obs})
mqtt5adapter.ServeRequestReply(ctx, client, router, handle, fn, mqtt5.ServeOptions{Observer: obs})

func NewRoute 

func NewRoute[Req, Resp any](
	topic string,
	reqCodec codex.Codec[Req],
	respCodec codex.Codec[Resp],
	opts ...RouteOpt,
) Route[Req, Resp]

NewRoute creates a Route spec from a topic, codecs, and variadic opts. NewRoute is infallible — validation runs at Route.Register time.

NewRoute is a free function (not a method) because Go requires type parameters on free functions, not on method receivers.

func (Route[Req, Resp]) ClientHandle 

func (r Route[Req, Resp]) ClientHandle() *RouteHandle[Req, Resp]

ClientHandle returns a RouteHandle for client-side use without registering with a Builder. No spec registration occurs.

Use ClientHandle when only the client side needs codec and route definitions (no AsyncAPI spec, no server), or when sharing a Route definition between server and client in the same binary without a second builder registration.

The returned handle has the same Decode / Encode / EncodeRequest / DecodeResponse codec helpers and BuildTopic / ValidateTopicVars methods as a handle returned by Route.Register.

Example — client-only usage (no builder required): 

var ComputeRoute = reqreply.NewRoute[ComputeReq, ComputeResp](
    "compute/add", computeReqCodec, computeRespCodec,
)

// Client side — no builder needed.
handle := ComputeRoute.ClientHandle()
resp, err := mqtt5adapter.Call(ctx, client, router, handle, req, mqtt5adapter.CallOptions{})

Mirrors [rest.Route.ClientHandle].

func (Route[Req, Resp]) Register 

func (r Route[Req, Resp]) Register(b *Builder) (*RouteHandle[Req, Resp], error)

Register registers the route with b and returns a RouteHandle.

Returns DuplicateRouteError if a route with the same topic has already been registered with b.

Use RouteHandle.WithRequestFormats and RouteHandle.WithFormats after Register to configure multi-format request/response handling.

type RouteHandle 

type RouteHandle[Req, Resp any] struct {
	// Topic is the request address (e.g. "compute/add").
	Topic string

	// Decode deserialises and validates a JSON request payload into Req.
	// All Refine constraints on the request codec run automatically.
	Decode func(payload []byte) (Req, error)

	// Encode serialises Resp to JSON bytes.
	Encode func(resp Resp) ([]byte, error)

	// EncodeRequest serialises Req to JSON bytes for use as an outgoing request
	// payload. It is the client-side complement of Decode.
	EncodeRequest func(req Req) ([]byte, error)

	// DecodeResponse deserialises and validates a JSON reply payload into Resp.
	// It is the client-side complement of Encode.
	DecodeResponse func(payload []byte) (Resp, error)

	// RequestFormats, when non-empty, overrides the default JSON format for
	// decoding incoming request payloads. The adapter uses RequestFormats[0]
	// instead of Decode when present.
	// Configure via [RouteHandle.WithRequestFormats].
	RequestFormats []format.Format[Req]

	// Formats, when non-empty, overrides the default JSON format for encoding
	// reply payloads. The adapter uses Formats[0] instead of Encode when present.
	// Configure via [RouteHandle.WithFormats].
	Formats []format.Format[Resp]
	// contains filtered or unexported fields
}

RouteHandle is returned by Route.Register. It holds the codec-backed Decode/Encode helpers and is passed directly to request-reply adapters (adapters/zeromq, adapters/mqtt5).

RouteHandle mirrors [rest.RouteHandle] and [events.ChannelHandle]: it is a value that callers pass around and store. No magic, no global state.

func (*RouteHandle[Req, Resp]) BuildTopic 

func (h *RouteHandle[Req, Resp]) BuildTopic(vars map[string]string) (string, error)

BuildTopic substitutes {varName} placeholders in the route's topic template with the values provided in vars, validating each against its registered TopicParam codec (if any).

All template variables must be present in vars; missing variables return a MissingRouteParamError. Values are validated before substitution; codec failures return a RouteParamError identifying the variable name and value. Keys in vars that do not appear in the template are silently ignored.

Mirrors [events.ChannelHandle.BuildTopic]. 

topic, err := computeRoute.BuildTopic(map[string]string{"tenantID": "acme"})
// topic = "compute/acme/add"

func (*RouteHandle[Req, Resp]) ValidateTopicVars 

func (h *RouteHandle[Req, Resp]) ValidateTopicVars(vars map[string]string) error

ValidateTopicVars validates extracted topic variable values against the registered TopicParam codecs. Call this after extracting vars from an incoming request topic to ensure each variable satisfies its codec constraints.

Returns RouteParamError for the first variable that fails its codec. Variables without a registered codec are skipped. Missing required variables return MissingRouteParamError.

Mirrors [events.ChannelHandle.ValidateTopicVars].

func (*RouteHandle[Req, Resp]) WithFormats 

func (h *RouteHandle[Req, Resp]) WithFormats(fmts ...format.Format[Resp]) *RouteHandle[Req, Resp]

WithFormats sets the formats used for encoding reply payloads and returns the updated handle. Adapters use Formats[0] for encoding when non-empty, falling back to RouteHandle.Encode (JSON) otherwise.

Mirrors [rest.RouteHandle.WithFormats].

func (*RouteHandle[Req, Resp]) WithRequestFormats 

func (h *RouteHandle[Req, Resp]) WithRequestFormats(fmts ...format.Format[Req]) *RouteHandle[Req, Resp]

WithRequestFormats sets the formats the route accepts for request body decoding and returns the updated handle. Adapters use RequestFormats[0] for decoding when non-empty, falling back to RouteHandle.Decode (JSON) otherwise.

Mirrors [rest.RouteHandle.WithRequestFormats].

type RouteMeta 

type RouteMeta struct {
	// OperationID is the base name for the two generated operations.
	// The send operation is named "send<OperationID>" and the receive operation
	// is named "receive<OperationID>Reply". When empty, the topic is used
	// (e.g. "compute/add" → "sendComputeAdd" / "receiveComputeAddReply").
	OperationID string

	// Summary is a short human-readable summary of the route.
	Summary string

	// Description is a longer human-readable description for the send operation.
	Description string

	// Tags attach arbitrary labels to the operations in the AsyncAPI spec.
	Tags []string

	// ReqSchemaName, when non-empty, registers the request payload schema in
	// components/schemas and emits a $ref. Use to share schemas across routes.
	ReqSchemaName string

	// RespSchemaName, when non-empty, registers the response payload schema in
	// components/schemas and emits a $ref.
	RespSchemaName string
}

RouteMeta holds metadata for a Route registration. It controls the generated AsyncAPI operation IDs, summary, description, and schema refs.

RouteMeta implements RouteOpt: pass it directly to NewRoute.

type RouteOpt 

type RouteOpt interface {
	// contains filtered or unexported methods
}

RouteOpt is the sealed interface for variadic NewRoute options.

The following types implement RouteOpt:

  • RouteMeta — operation metadata (OperationID, Summary, Description, Tags, schema names)
  • TopicParam — topic template variable with optional codec and description

type RouteParamError 

type RouteParamError struct {
	Name  string // the {varName} that failed
	Value string // the value that was rejected
	Err   error  // the underlying codec error
}

RouteParamError is returned by RouteHandle.BuildTopic and RouteHandle.ValidateTopicVars when a topic variable fails its registered codec check. It mirrors [events.TopicParamError]. 

var paramErr reqreply.RouteParamError
if errors.As(err, &paramErr) {
    slog.Warn("bad topic var", "error", paramErr)
}

func (RouteParamError) Error 

func (e RouteParamError) Error() string

func (RouteParamError) LogValue 

func (e RouteParamError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (RouteParamError) Unwrap 

func (e RouteParamError) Unwrap() error

Unwrap allows errors.Is and errors.As to traverse the underlying error.

type Server 

type Server = asyncapi.Server

Server is an alias for asyncapi.Server.

type TopicParam 

type TopicParam struct {
	// Name is the variable name (without braces) as it appears in the topic template.
	Name string
	// Description is shown in the AsyncAPI spec for this parameter.
	Description string
	// Codec validates topic parameter values at [RouteHandle.ValidateTopicVars] and
	// [RouteHandle.BuildTopic] time.
	// When non-nil, the codec's schema is also emitted in the AsyncAPI spec.
	// Nil means no runtime validation.
	Codec *codex.Codec[string]
}

TopicParam describes a {varName} placeholder in a topic template. It is the api/reqreply analogue of [events.TopicParam].

TopicParam is optional: RouteHandle.BuildTopic and RouteHandle.ValidateTopicVars use registered params to validate variable values. Use TopicParam when you want runtime codec validation on a specific variable.

Note: all topic variables are always required — a template cannot be resolved without every {varName} placeholder present. There is no Required field.

TopicParam implements RouteOpt: pass it directly to NewRoute.

Entry names must correspond to {varName} placeholders in the topic template. 

var ComputeRoute = reqreply.NewRoute[ComputeReq, ComputeResp](
    "compute/{tenantID}/add",
    computeReqCodec, computeRespCodec,
    reqreply.RouteMeta{OperationID: "computeAdd"},
    reqreply.TopicParam{
        Name:        "tenantID",
        Description: "Tenant namespace for this computation.",
    }.WithCodec(codex.String().Refine(validate.NonEmptyString)),
)

func (TopicParam) WithCodec 

func (p TopicParam) WithCodec(c codex.Codec[string]) TopicParam

WithCodec sets the validation codec and returns the updated TopicParam.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.