zeromq

package
v0.12.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrTimeout = errors.New("zeromq: receive timeout")

ErrTimeout is returned by FramedSocket.RecvFrames when no message arrived within the socket's configured receive timeout window. The adapter receive loops treat ErrTimeout as a non-fatal signal to re-check context cancellation and retry.

Functions

func AsPipelineFunc 

func AsPipelineFunc[Req, Resp any](
	fn func(context.Context, Req) gstream.Stream[Resp],
) func(context.Context, Req) (Resp, error)

AsPipelineFunc converts a pipeline handler function into the plain handler function signature accepted by Serve and ServeRouter.

Internally: wraps req as gstream.Single, calls fn to build the pipeline, then collects the result via gstream.Collect. Errors take precedence over values. If the pipeline emits no value, PipelineNoResponseError is returned.

Use AsPipelineFunc when the handler body benefits from gstream.Tap for declarative intermediate observation, gstream.Apply for multi-step forge function composition, or gstream.MapErr for per-step error recovery: 

zeromq.Serve(ctx, sock, oeeHandle,
    zeromq.AsPipelineFunc(func(ctx context.Context, req SensorReq) gstream.Stream[OEEResult] {
        s  := gstream.Single(ctx, req)
        s   = gstream.Apply(ctx, s, validateFn, gstream.ApplyOptions{Observer: obs})
        s   = gstream.Tap(ctx, s, func(v ValidatedReq) { slog.Info("valid", "id", v.ID) })
        out := gstream.Apply(ctx, s, oeeCalcFn, gstream.ApplyOptions{Observer: obs})
        return gstream.Tap(ctx, out, func(r OEEResult) { auditLog.Write(r) })
    }),
    zeromq.ServeOptions{Observer: obs})

For simple single-step handlers, use a plain fn directly with Serve.

func Call 

func Call[Req, Resp any](
	ctx context.Context,
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	req Req,
	opts CallOptions,
) (Resp, error)

Call encodes req, sends it to a REQ socket, and decodes the reply. It is the client side of a ZMQ REQ/REP contract.

Message framing:

  • Outgoing request: [payload]
  • Expected reply OK: ["ok", encoded_response]
  • Server error reply:["error", message] → returns CallError

ctx cancellation is honoured during the reply receive loop. Call blocks until a reply arrives, ctx is cancelled, or a socket error occurs.

Format overrides are applied via reqreply.RouteHandle.WithRequestFormats and reqreply.RouteHandle.WithFormats on the handle before calling Call.

Example (REQ compute client): 

result, err := zeromq.Call(ctx, sock, computeHandle.ClientHandle(), req,
    zeromq.CallOptions{Observer: obs})

func CallAdapter 

func CallAdapter[Req, Resp any](
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	opts CallStreamOptions,
) ports.IOAdapter[Req, Resp]

CallAdapter returns a ports.IOAdapter that performs ZeroMQ request-reply for each upstream item. Use with ports.IOPort.Bind: 

domain.Calibration.Bind(ctx, zeromq.CallAdapter(sock, calibHandle, zeromq.CallStreamOptions{}))

func CallDealer 

func CallDealer[Req, Resp any](
	ctx context.Context,
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	req Req,
	opts CallOptions,
) (Resp, error)

CallDealer encodes req and sends it via a DEALER socket using the ZMQ envelope format (empty delimiter + payload), then synchronously waits for one reply.

DEALER message framing (client sends): 

["", payload]

Expected reply framing (client receives): 

["", "ok", encoded_response]
["", "error", error_message]

For concurrent use, call CallDealer from multiple goroutines; each invocation manages its own independent send/recv cycle.

ctx cancellation is honoured during the reply receive loop.

Errors are wrapped in CallError, the same type used by Call.

Example (DEALER compute client): 

result, err := zeromq.CallDealer(ctx, sock, handle, ComputeReq{X: 3, Y: 4},
    zeromq.CallOptions{Observer: obs})

func CallHandle added in v0.12.0 

func CallHandle[Req, Resp any](
	ctx context.Context,
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	req Req,
	opts CallOptions,
) (Resp, error)

CallHandle is the single-call convenience wrapper around Call: it derives CallOptions.Vars from req automatically, using the route's merge-capable topic params (reqreply.RouteHandle.MergeFields + codex.EncodeVars) — mirrors [nethttp.CallHandle]/[mqtt5.CallHandle].

An explicit CallOptions.Vars takes PRECEDENCE over the derived value. Call remains the lower-level escape hatch.

Note: this convenience is CLIENT-SIDE only. ZMQ REQ/REP routing is socket-based, not topic-based — Serve's incoming messages carry no per-message topic string to extract vars FROM (unlike MQTT's broker-routed topics), so there is no server-side decode-merge equivalent for zeromq. The resolved topic here is used only for codec validation and observer reporting, matching CallOptions.Vars's existing documented behavior. 

resp, err := zeromq.CallHandle(ctx, sock, computeRoute, req, zeromq.CallOptions{})

func LatestAdapter added in v0.12.0 

func LatestAdapter[Resp any](
	sock FramedSocket,
	handle *reqreply.RouteHandle[struct{}, Resp],
	opts ServeLatestOptions,
) ports.LatestAdapter[Resp]

LatestAdapter returns a ports.LatestAdapter that serves a ports.LatestPort's cached value over a blocking REP loop — the port-based successor to ServeLatest (which owns its own cache cell; the port owns it here). Use with ports.LatestPort.Bind; the port runs the blocking Serve in a supervised goroutine: 

handle, _ := domain.Latest.PluginReqReplyPattern(domain.LatestPattern)
must(domain.Latest.Bind(ctx, zeromq.LatestAdapter(sock, handle, zeromq.ServeLatestOptions{})))
go domain.Latest.Feed(ctx, oeeStream)

When a request arrives before the first value, the REP socket sends an error reply and opts.OnError receives NoLatestValueError (same semantics as ServeLatest).

func Publish 

func Publish[T any](
	ctx context.Context,
	sock FramedSocket,
	handle *events.ChannelHandle[T],
	msg T,
	vars map[string]string,
	opts PublishOptions,
	formats ...format.Format[T],
) error

Publish encodes msg using handle's codec and sends it to a PUB or PUSH socket.

The message is framed as [topic, payload]:

  • topic: the resolved topic string (after BuildTopic if vars are provided)
  • payload: the codec-encoded message bytes

For PUB sockets the topic frame is used for ZMQ prefix-filter matching. For PUSH sockets the topic frame is sent but ignored by PULL receivers.

vars controls topic resolution:

  • nil: use handle.Topic directly (static topics).
  • non-nil: call handle.BuildTopic(vars) to resolve a template topic.

The optional formats parameter overrides the channel handle's default JSON codec. Priority: call-time formats > handle.PublishFormats > handle.Formats > handle.Encode (JSON fallback).

func PublishAdapter 

func PublishAdapter[T any](
	sock FramedSocket,
	handle *events.ChannelHandle[T],
	fmt format.Format[T],
	opts DrainPublishOptions,
) ports.SinkAdapter[T]

PublishAdapter returns a ports.SinkAdapter that publishes each item via ZeroMQ. Use with ports.SinkPort.Bind: 

domain.OEEResults.Bind(ctx, zeromq.PublishAdapter(sock, alertHandle, format.JSON(OEECodec),
    zeromq.DrainPublishOptions{}))

func PublishHandle added in v0.12.0 

func PublishHandle[T any](
	ctx context.Context,
	sock FramedSocket,
	handle *events.ChannelHandle[T],
	msg T,
	opts PublishOptions,
	formats ...format.Format[T],
) error

PublishHandle is the single-call convenience wrapper around Publish: it derives the topic vars map from msg automatically, using the channel's merge-capable topic params (events.ChannelHandle.MergeFields + codex.EncodeVars) — one struct in, no manual vars map, mirroring [mqtt5.PublishHandle]'s convenience for MQTT 5 events.

Publish remains available as the lower-level escape hatch for callers that build the vars map themselves (e.g. no merge fields declared, or vars come from a non-struct source). 

err := zeromq.PublishHandle(ctx, sock, sensorChannel, reading, zeromq.PublishOptions{})

func Serve 

func Serve[Req, Resp any](
	ctx context.Context,
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	fn func(context.Context, Req) (Resp, error),
	opts ServeOptions,
) error

Serve runs a blocking REP loop: receives requests, calls fn, sends replies. It is the server side of a ZMQ REQ/REP contract.

Message framing:

  • Incoming request: [payload]
  • Reply on success: ["ok", encoded_response]
  • Reply on failure: ["error", error_message]

The REP socket always sends a reply (even on error) to avoid leaving the REQ peer blocked. Per-error details are delivered via ServeOptions.OnError.

The loop runs until ctx is cancelled (returns nil) or a socket error occurs. Run Serve in a dedicated goroutine.

Format overrides are applied via reqreply.RouteHandle.WithRequestFormats and reqreply.RouteHandle.WithFormats on the handle before calling Serve.

Example (REP compute server): 

go func() {
    if err := zeromq.Serve(ctx, sock, computeHandle, handler, zeromq.ServeOptions{Observer: obs}); err != nil {
        log.Error("serve stopped", "err", err)
    }
}()

func ServeAdapter 

func ServeAdapter[Req, Resp any](
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	opts ServeOptions,
) ports.ToolAdapter[Req, Resp]

ServeAdapter returns a ports.ToolAdapter that registers the pipeline function as a ZeroMQ REP server via Serve. When ports.ToolPort.Bind is called, the pipeline function is wrapped as an AsPipelineFunc handler and Serve is started in a background goroutine. Use with ports.ToolPort.Bind: 

domain.OEEToolPort.Bind(ctx, zeromq.ServeAdapter(repSock, handle, zeromq.ServeOptions{}))

func ServeLatest 

func ServeLatest[Req, Resp any](
	ctx context.Context,
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	src gstream.Stream[Resp],
	opts ServeLatestOptions,
) error

ServeLatest runs a blocking REP loop that replies to every incoming request with the most recently emitted value from src.

A background goroutine reads src.Values and atomically stores each new value. When a request arrives but no value has been produced yet, the REP socket sends an error reply and opts.OnError is called with NoLatestValueError.

The Req payload is decoded and validated by handle (standard Serve behaviour) but is not used to compute the response — the response is always the latest value.

Returns nil when ctx is cancelled, or a SocketError on socket failure.

Use ServeLatest for "get current OEE", "get latest sensor reading", or any "current state" ZMQ endpoint backed by a continuous stream computation.

func ServeRouter 

func ServeRouter[Req, Resp any](
	ctx context.Context,
	sock FramedSocket,
	handle *reqreply.RouteHandle[Req, Resp],
	fn func(context.Context, Req) (Resp, error),
	opts ServeOptions,
) error

ServeRouter runs a blocking ROUTER loop. Each incoming request is dispatched concurrently in its own goroutine. Identity frames are extracted automatically and re-prepended to every reply so the DEALER peer can correlate responses.

ROUTER message framing (server receives): 

[identity, "", payload]

Reply framing (server sends): 

[identity, "", "ok", encoded_response]
[identity, "", "error", error_message]

The loop runs until ctx is cancelled; it waits for all in-flight goroutines to drain before returning nil. A socket recv error stops the loop immediately.

Errors are delivered via ServeOptions.OnError using the same ServeError type as Serve. Format overrides are applied via reqreply.RouteHandle.WithRequestFormats and reqreply.RouteHandle.WithFormats before calling ServeRouter.

Example (ROUTER compute server): 

go func() {
    if err := zeromq.ServeRouter(ctx, sock, handle, handler,
        zeromq.ServeOptions{Observer: obs}); err != nil {
        log.Error("serve stopped", "err", err)
    }
}()

func Subscribe 

func Subscribe[T any](
	ctx context.Context,
	sock FramedSocket,
	handle *events.ChannelHandle[T],
	fn func(context.Context, T) error,
	opts SubscribeOptions,
	formats ...format.Format[T],
) error

Subscribe blocks and processes incoming messages from a SUB or PULL socket. Each message is decoded using handle's codec and delivered to fn.

For SUB sockets, Subscribe registers handle.Topic as the ZMQ subscription prefix filter before entering the receive loop. For PULL sockets the filter is a no-op — call FramedSocket.SetSubscription("") separately if needed.

Messages are expected in [topic, payload] frame format. The topic frame is used for observer reporting; the payload frame is decoded by the codec.

The loop runs until ctx is cancelled (returns nil) or a socket error occurs (returns the error). Run Subscribe in a dedicated goroutine.

The optional formats parameter overrides the channel handle's default JSON codec. Priority: call-time formats > handle.SubscribeFormats > handle.Formats > handle.Decode (JSON fallback).

Example (PUB/SUB sensor readings): 

go func() {
    if err := zeromq.Subscribe(ctx, sock, readingsHandle, func(ctx context.Context, r SensorReading) error {
        return store.Save(ctx, r)
    }, zeromq.SubscribeOptions{Observer: obs}); err != nil {
        log.Error("subscribe stopped", "err", err)
    }
}()

func SubscribeAdapter 

func SubscribeAdapter[T any](
	sock FramedSocket,
	handle *events.ChannelHandle[T],
	fmt format.Format[T],
	opts SubscribeAdapterOptions,
) ports.SourceAdapter[T]

SubscribeAdapter returns a ports.SourceAdapter backed by the ZeroMQ PUB/SUB receive loop. Use with ports.SourcePort.Bind: 

domain.SensorReadings.Bind(ctx, zeromq.SubscribeAdapter(
    sock, sensorHandle,
    format.JSON(ReadingCodec),
    zeromq.SubscribeAdapterOptions{Buffer: 8},
))

func TopicVarsFromMessage added in v0.12.0 

func TopicVarsFromMessage[T any](handle *events.ChannelHandle[T], topic string) (map[string]string, error)

TopicVarsFromMessage is the inverse of events.ChannelHandle.BuildTopic. It matches a concrete ZeroMQ topic (the first frame of a [topic, payload] message, as read by Subscribe) against the channel's topic template and returns the extracted variable values — the zeromq equivalent of adapters/mqtt5.TopicVarsFromMessage and adapters/mqtt.TopicVarsFromMessage, adapted for zeromq's plain-string topic (no message struct to extract it from; ZeroMQ PUB/SUB frames the topic as its own frame).

Template syntax: "{varName}" placeholders capture everything up to the next "/" (never crossing a segment boundary); literal text must match exactly. No MQTT-style wildcard support (+/#) — ZeroMQ PUB/SUB topic filtering is prefix-based, not wildcard-based, so channel topic templates declared for zeromq are not expected to contain them.

Typical usage — channel defined with go-codex template variables declared via merge-capable events.NewTopicParam fields, so the extracted vars can be merged directly into the payload via events.ChannelHandle.DecodeMerged (this is exactly what Subscribe does internally whenever the channel declares merge fields): 

vars, err := zeromq.TopicVarsFromMessage(sensorChannel, topic)
// vars["sensorID"] == "f47ac10b-..."

Returns TopicMismatchError if the concrete topic does not match the template structure (wrong number of segments or a literal segment does not match). Returns events.InvalidTopicError if the concrete topic fails the builder-level topic codec (see events.WithTopicConstraints). Returns events.TopicParamError if an extracted variable fails its registered events.TopicParam codec.

Types

type CallError 

type CallError struct {
	Err error
}

CallError wraps REQ-socket failures: encode, send, receive, or decode. 

var callErr zeromq.CallError
if errors.As(err, &callErr) {
    slog.Error("zmq call failed", "error", callErr)
}

func (CallError) Error 

func (e CallError) Error() string

func (CallError) LogValue 

func (e CallError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (CallError) Unwrap 

func (e CallError) Unwrap() error

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

type CallOptions 

type CallOptions struct {
	// Observer, when non-nil, receives per-call lifecycle events:
	// [stats.Observer.RecordRequest] is called with method "ZMQ-REQ" or
	// "ZMQ-DEALER", the route path, status 200 on success, and status 0 or
	// 500 on failure. Per-field decode errors are reported with location "body".
	// Topic variable errors are reported with location "topic_var".
	// Defaults to [stats.NoopObserver] when nil.
	Observer stats.Observer

	// Vars, when non-nil, substitutes {varName} placeholders in the route topic
	// template before encoding. Uses [reqreply.RouteHandle.BuildTopic] to
	// resolve and codec-validate each variable.
	//
	// In ZMQ REQ/REP, the resolved topic is used for observer reporting only —
	// the actual routing is socket-based. Validation still runs on each variable.
	//
	// Example — template topic "compute/{tenantID}/add":
	//
	//	zeromq.Call(ctx, sock, handle, req,
	//	    zeromq.CallOptions{Vars: map[string]string{"tenantID": "acme"}})
	//
	// Returns [CallError] wrapping [reqreply.RouteParamError] or
	// [reqreply.MissingRouteParamError] on validation failure.
	Vars map[string]string
}

CallOptions configures Call and CallDealer.

type CallStreamOptions 

type CallStreamOptions struct {
	// Vars substitutes {varName} placeholders in the route topic template.
	//
	// When nil, vars are derived PER-ITEM from each item's own
	// merge-field-declared struct fields (the same convenience [CallHandle]
	// provides). When set to a non-nil map (including an explicitly empty
	// one), that map is used as-is for every request (static vars only) —
	// the escape hatch, unchanged from prior behavior.
	Vars map[string]string
	// Observer receives per-call lifecycle events.
	Observer stats.Observer
	// Buffer is the output Stream channel buffer size. Default 0.
	Buffer int
}

CallStreamOptions configures CallAdapter.

type CorrelationError 

type CorrelationError struct {
	// Seq is the unmatched sequence number from the response frame.
	Seq uint64
	Err error
}

CorrelationError is sent to [Stream.Errors] by [CallDealerStream] when a response frame arrives with a sequence number that does not match any pending request — typically a stale reply from a previous session.

func (CorrelationError) Error 

func (e CorrelationError) Error() string

func (CorrelationError) LogValue 

func (e CorrelationError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (CorrelationError) Unwrap 

func (e CorrelationError) Unwrap() error

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

type DrainPublishOptions 

type DrainPublishOptions struct {
	// Vars substitutes {varName} placeholders in the channel handle's topic
	// template.
	//
	// When nil, topic vars are derived PER-ITEM from each item's own
	// merge-field-declared struct fields (the same convenience
	// [PublishHandle] provides) — every item may resolve to a different
	// concrete topic. When set to a non-nil map (including an explicitly
	// empty one), that map is used as-is for every item (static topic vars
	// only) — the escape hatch, unchanged from prior behavior.
	Vars map[string]string
	// OnError, when non-nil, is called for encode failures ([PublishEncodeError]),
	// socket send failures ([SocketError]), or upstream stream errors.
	OnError func(error)
	// Observer receives per-publish lifecycle events.
	Observer stats.Observer
}

DrainPublishOptions configures PublishAdapter publish behaviour.

type ErrorKind 

type ErrorKind int

ErrorKind classifies the origin of a SubscribeError or ServeError. 

const (
	// KindDecode indicates the message payload could not be decoded or
	// failed codec validation.
	KindDecode ErrorKind = iota

	// KindHandler indicates the application handler returned an error after
	// successful decoding.
	KindHandler

	// KindEncode indicates the response could not be encoded (server side)
	// or the outgoing payload failed codec validation (publish side).
	KindEncode
)

func (ErrorKind) String 

func (k ErrorKind) String() string

type FramedSocket 

type FramedSocket interface {
	// SendFrames sends a multi-frame ZMQ message. Each element of frames is
	// one frame; all but the last are sent with the SNDMORE flag.
	SendFrames(frames [][]byte) error

	// RecvFrames receives the next multi-frame ZMQ message.
	// Returns (nil, [ErrTimeout]) when no message arrives within the
	// socket's configured receive timeout (set via [SetRecvTimeout]).
	// Returns (nil, err) on socket errors.
	RecvFrames() ([][]byte, error)

	// SetSubscription registers a topic prefix filter on a SUB socket.
	// The ZMQ broker delivers only messages whose first frame starts with topic.
	// Call with an empty string to receive all messages on a PULL socket.
	// Calling on non-SUB sockets that ignore the option should return nil.
	SetSubscription(topic string) error

	// SetRecvTimeout configures how long [RecvFrames] blocks before returning
	// [ErrTimeout]. Set to a short interval (e.g. 100 ms) so that receive loops
	// can check context cancellation periodically.
	SetRecvTimeout(d time.Duration) error
}

FramedSocket is the transport interface used by this adapter. It abstracts a ZMQ socket down to the minimal surface needed for multi-frame send/receive and subscription management.

Wire it to your preferred ZMQ library. Example with pebbe/zmq4: 

import zmq "github.com/pebbe/zmq4"

type pebbeSocket struct{ s *zmq.Socket }

func (p *pebbeSocket) SendFrames(frames [][]byte) error {
    for i, f := range frames {
        flag := zmq.SNDMORE
        if i == len(frames)-1 {
            flag = 0
        }
        if _, err := p.s.SendBytes(f, flag); err != nil {
            return err
        }
    }
    return nil
}

func (p *pebbeSocket) RecvFrames() ([][]byte, error) {
    frames, err := p.s.RecvMessageBytes(0)
    if err != nil {
        if zmq.AsErrno(err) == zmq.EAGAIN {
            return nil, zeromq.ErrTimeout
        }
        return nil, err
    }
    return frames, nil
}

func (p *pebbeSocket) SetSubscription(topic string) error {
    return p.s.SetSubscribe(topic)
}

func (p *pebbeSocket) SetRecvTimeout(d time.Duration) error {
    return p.s.SetRcvtimeo(d)
}

type NoLatestValueError 

type NoLatestValueError struct {
	// Topic is the route topic (from the RouteHandle).
	Topic string
}

NoLatestValueError is passed to ServeLatestOptions.OnError when a request arrives before the source stream has produced any value. The REP socket sends an error reply; this error is informational for the server operator.

func (NoLatestValueError) Error 

func (e NoLatestValueError) Error() string

func (NoLatestValueError) LogValue 

func (e NoLatestValueError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

type PipelineNoResponseError 

type PipelineNoResponseError struct {
	// Topic is the route topic (from the RouteHandle).
	Topic string
}

PipelineNoResponseError is returned by AsPipelineFunc when [stream.Collect] returns without any value — either the pipeline emitted nothing, or the request context was cancelled before the pipeline produced a result.

func (PipelineNoResponseError) Error 

func (e PipelineNoResponseError) Error() string

func (PipelineNoResponseError) LogValue 

func (e PipelineNoResponseError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

type PublishEncodeError 

type PublishEncodeError struct {
	// Topic is the resolved topic (after template substitution if vars were provided).
	Topic string
	// Err is the underlying codec validation or marshal error.
	Err error
}

PublishEncodeError is returned by Publish when encoding the outgoing message payload fails (codec validation or marshal error).

Use errors.As to extract the topic and underlying error: 

var encErr zeromq.PublishEncodeError
if errors.As(err, &encErr) {
    slog.Error("publish encode failed", "error", encErr)
}

func (PublishEncodeError) Error 

func (e PublishEncodeError) Error() string

func (PublishEncodeError) LogValue 

func (e PublishEncodeError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (PublishEncodeError) Unwrap 

func (e PublishEncodeError) Unwrap() error

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

type PublishOptions 

type PublishOptions struct {
	// Observer, when non-nil, receives per-publish lifecycle events:
	// [stats.Observer.RecordPublish] is called with success=true on broker send
	// and success=false on encode failure or send error. Per-field payload encode
	// errors are reported via [stats.Observer.RecordValidationError] with location
	// "payload". Topic variable errors are reported with location "topic_var".
	// Defaults to [stats.NoopObserver] when nil.
	Observer stats.Observer
}

PublishOptions configures Publish.

type ServeError 

type ServeError struct {
	Kind ErrorKind
	Err  error
}

ServeError is delivered to ServeOptions.OnError on REP-side failures. The Kind field identifies whether the failure occurred during decode, handler execution, or response encoding. 

var serveErr zeromq.ServeError
if errors.As(err, &serveErr) {
    slog.Warn("serve failed", "error", serveErr)
}

func (ServeError) Error 

func (e ServeError) Error() string

func (ServeError) LogValue 

func (e ServeError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (ServeError) Unwrap 

func (e ServeError) Unwrap() error

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

type ServeLatestError 

type ServeLatestError struct {
	Op  string
	Err error
}

ServeLatestError is passed to ServeLatestOptions.OnError when a socket, decode, or encode operation fails during the ServeLatest serve loop.

Op values: "recv" (socket read), "decode" (request decode), "encode" (response encode), "send" (socket write).

func (ServeLatestError) Error 

func (e ServeLatestError) Error() string

func (ServeLatestError) LogValue 

func (e ServeLatestError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (ServeLatestError) Unwrap 

func (e ServeLatestError) Unwrap() error

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

type ServeLatestOptions 

type ServeLatestOptions struct {
	// OnError, when non-nil, is called for socket errors ([ServeLatestError]),
	// no-value conditions ([NoLatestValueError]), or encode failures.
	OnError func(error)

	// Observer receives per-request lifecycle events via [stats.Observer.RecordRequest].
	Observer stats.Observer
}

ServeLatestOptions configures ServeLatest.

type ServeOptions 

type ServeOptions struct {
	// OnError, when non-nil, is called with a typed [ServeError] on decode,
	// handler, or encode failure. The REP socket always sends an error reply
	// frame to avoid leaving the REQ peer stuck; OnError is informational.
	// If nil, errors are silently discarded (the error reply is still sent).
	OnError func(ServeError)

	// Observer, when non-nil, receives per-request lifecycle events:
	// [stats.Observer.RecordRequest] is called with method "ZMQ-REP", the
	// route path, status 200 on success, and status 0 on failure.
	// Per-field validation errors are reported with location "body".
	// Defaults to [stats.NoopObserver] when nil.
	Observer stats.Observer
}

ServeOptions configures Serve.

type SocketError 

type SocketError struct {
	// Op identifies the socket operation that failed.
	// Common values: "set_subscription", "set_recv_timeout", "recv", "send".
	Op string
	// Err is the underlying socket or OS error.
	Err error
}

SocketError wraps socket-level infrastructure failures: socket option configuration (SetSubscription, SetRecvTimeout) and transport I/O (recv, send). It is distinct from codec-level errors (SubscribeError, ServeError, CallError) which reflect application-layer decode/handler/encode failures.

The Op field identifies which socket operation failed so callers can distinguish a recv failure from a configuration failure without string matching: 

var sockErr zeromq.SocketError
if errors.As(err, &sockErr) {
    switch sockErr.Op {
    case "recv":               // socket connection died mid-loop
    case "send":               // socket send failed
    case "set_recv_timeout":   // could not configure socket option
    case "set_subscription":   // could not set SUB filter
    }
    slog.Error("socket failed", "error", sockErr)
}

func (SocketError) Error 

func (e SocketError) Error() string

func (SocketError) LogValue 

func (e SocketError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (SocketError) Unwrap 

func (e SocketError) Unwrap() error

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

type SubscribeAdapterOptions 

type SubscribeAdapterOptions struct {
	Buffer int
}

SubscribeAdapterOptions configures SubscribeAdapter.

type SubscribeError 

type SubscribeError struct {
	Kind  ErrorKind
	Topic string
	Err   error
}

SubscribeError is delivered to SubscribeOptions.OnError with a typed [Kind] so callers can distinguish decode/validation failures from application errors without string matching.

Use errors.As to extract the kind, topic, and underlying error: 

var subErr zeromq.SubscribeError
if errors.As(err, &subErr) {
    slog.Warn("subscribe failed", "error", subErr) // LogValue emits structured fields
}

func (SubscribeError) Error 

func (e SubscribeError) Error() string

func (SubscribeError) LogValue 

func (e SubscribeError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (SubscribeError) Unwrap 

func (e SubscribeError) Unwrap() error

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

type SubscribeOptions 

type SubscribeOptions struct {
	// OnError, when non-nil, is called with a typed [SubscribeError] on decode
	// or application handler failure. If nil, errors are silently discarded.
	OnError func(SubscribeError)

	// Observer, when non-nil, receives per-message lifecycle events:
	// [stats.Observer.RecordSubscribe] is called with success=true on clean
	// handler completion and success=false on any failure. Per-field payload
	// validation errors are reported via [stats.Observer.RecordValidationError]
	// with location "payload".
	// Defaults to [stats.NoopObserver] when nil.
	Observer stats.Observer
}

SubscribeOptions configures Subscribe.

type TopicMismatchError added in v0.12.0 

type TopicMismatchError struct {
	Template string // the channel topic template (e.g. "sensors/{sensorID}/data")
	Topic    string // the received concrete topic (e.g. "sensors/abc/extra/data")
}

TopicMismatchError is returned by TopicVarsFromMessage when the received topic does not match the structure of the channel's topic template. Mirrors adapters/mqtt5.TopicMismatchError/adapters/mqtt.TopicMismatchError (same shape, no wildcard-specific fields since zeromq templates carry none).

func (TopicMismatchError) Error added in v0.12.0 

func (e TopicMismatchError) Error() string

func (TopicMismatchError) LogValue added in v0.12.0 

func (e TopicMismatchError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

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.