README
¶
Asynchronous Message Transport
transport/message contains the broker-neutral contract for asynchronous
messages. It intentionally has no Kafka, NATS, RabbitMQ, or task-queue SDK
dependency.
subscriber := newSubscriber() // an application-owned adapter
server := message.NewServer(subscriber,
message.WithMiddleware(loggingMiddleware),
)
if err := server.Handle("accounts.created", handleAccountCreated); err != nil {
return err
}
app := forge.New(forge.WithServer(server))
return app.Run()
Adapters implement message.Publisher and message.Subscriber. A successful
Publisher.Publish call must document whether the broker acknowledged the
message or only accepted it into a local asynchronous buffer. A subscriber
must stop delivery when its lifetime context is canceled and must release all
resources from Subscription.Close.
Message.Body is encoded data and Message.Headers uses the same normalized,
multi-value metadata model as the HTTP and gRPC transports. Partition,
offset, acknowledgement handles, retry policy, and raw SDK values remain
adapter-specific.
Documentation
¶
Overview ¶
Package message defines the protocol-neutral contract for asynchronous message transports. Broker-specific clients belong in optional modules.
Example (ServerWideMiddleware) ¶
Example_serverWideMiddleware mirrors "Attaching server middleware » Server-wide": message servers take the same construction-time middleware option as HTTP and gRPC servers.
package main
// The example in this file mirrors the message-transport snippet in
// docs/agent/middleware.md so that the guide cannot drift from the API
// without breaking the build. When it stops compiling, fix the guide
// together with the example.
import (
"context"
"fmt"
"log/slog"
"github.com/sylphylabs/forge/middleware/logging"
"github.com/sylphylabs/forge/middleware/recovery"
"github.com/sylphylabs/forge/transport/message"
)
// nopSubscriber stands in for a broker adapter (contrib/message/...).
type nopSubscriber struct{}
func (nopSubscriber) Subscribe(context.Context, string, message.Handler) (message.Subscription, error) {
return nopSubscription{}, nil
}
type nopSubscription struct{}
func (nopSubscription) Close(context.Context) error { return nil }
// Example_serverWideMiddleware mirrors "Attaching server middleware »
// Server-wide": message servers take the same construction-time middleware
// option as HTTP and gRPC servers.
func main() {
logger := slog.Default()
subscriber := nopSubscriber{}
msgSrv := message.NewServer(subscriber,
message.WithMiddleware(recovery.Recovery(), logging.Server(logger)),
)
_ = msgSrv
fmt.Println("constructed")
}
Output: constructed
Index ¶
Examples ¶
Constants ¶
const KindMessage transport.Kind = "message"
KindMessage identifies the asynchronous message transport. It is declared here rather than in the transport package because transport.Kind is an open type.
Variables ¶
var ( // ErrNilSubscriber reports a server constructed without a subscriber. ErrNilSubscriber = errors.New("message: nil subscriber") // ErrNoBindings reports a server started without handlers. ErrNoBindings = errors.New("message: no bindings") // ErrAlreadyStarted reports a mutation after startup. ErrAlreadyStarted = errors.New("message: server already started") // ErrStopped reports a server that has already been stopped. ErrStopped = errors.New("message: server stopped") // ErrEmptyTopic reports an invalid destination. ErrEmptyTopic = errors.New("message: empty topic") // ErrNilHandler reports an invalid binding. ErrNilHandler = errors.New("message: nil handler") // ErrNilContext reports a nil lifecycle context. ErrNilContext = errors.New("message: nil context") )
Functions ¶
func DestinationFromServerContext ¶
DestinationFromServerContext returns the destination that delivered the message being handled, and reports whether one was present.
A handler reads its destination here rather than from a parameter, the way an HTTP handler reads its request. Under a wildcard subscription this is the concrete destination, not the pattern that matched it.
Types ¶
type Handler ¶
Handler delivers one message to an adapter's subscription.
It is the shape a broker adapter implements, not the shape an application writes: destination is a parameter here because an adapter has the value before any context exists to carry it. Applications register a middleware.UnaryHandler with Server.Handle and read the destination from the transport.Transporter in context, as HTTP and gRPC handlers read their operation.
Returning an error leaves acknowledgement and retry policy to the adapter; the core contract does not guess those semantics, because brokers do not agree on them. Kafka and MQTT 5 have no negative acknowledgement at all, while RabbitMQ and JetStream do. An adapter whose broker can act on a failed handler exposes that choice as a construction option, so that the decision is made where the delivery is settled rather than by a caller who might forget.
type Message ¶
Message is the portable part of a delivered message.
Body is the encoded payload. Broker-specific delivery state such as partition, offset, acknowledgement handles, and raw SDK messages must stay in an adapter rather than becoming part of this contract.
func New ¶
New creates a message and takes a copy of body. This makes the caller's buffer safe to reuse after New returns.
type Publisher ¶
Publisher publishes encoded messages to a destination.
The context covers the publish operation. Adapters must document whether a successful return means broker acknowledgement or only local enqueueing.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server coordinates subscriptions and gives them the standard transport lifecycle. It owns every subscription created by Start and closes them in reverse registration order.
func NewServer ¶
func NewServer(subscriber Subscriber, opts ...ServerOption) *Server
NewServer creates a message lifecycle coordinator. A nil subscriber or an invalid middleware chain is reported by Start so construction can remain side-effect free.
func (*Server) Handle ¶
func (s *Server) Handle(topic string, handler middleware.UnaryHandler) error
Handle registers one destination handler. It must be called before Start.
type ServerOption ¶
type ServerOption func(*Server)
ServerOption configures a message Server before it starts.
func WithEndpoint ¶
func WithEndpoint(endpoint string) ServerOption
WithEndpoint sets the broker endpoint reported to middleware through transport.Transporter. It is descriptive only: the subscriber owns the actual connection.
func WithMiddleware ¶
func WithMiddleware(m ...middleware.UnaryMiddleware) ServerOption
WithMiddleware attaches server-wide middleware to every binding.
The middleware is the same middleware.UnaryMiddleware HTTP and gRPC use, so recovery, logging, rate limiting, and the rest apply to a message consumer without a message-specific implementation of each. It is composed once, inside NewServer; a nil middleware, or one returning a nil handler, is reported by Start, the way a nil subscriber is.
func WithShutdownTimeout ¶
func WithShutdownTimeout(timeout time.Duration) ServerOption
WithShutdownTimeout bounds cleanup triggered by cancellation of Start's parent context. Explicit Stop callers provide their own context.
type Subscriber ¶
Subscriber creates subscriptions for delivered messages. The context is the subscription lifetime: cancellation must stop delivery and release the adapter's resources. Close remains available for bounded, explicit shutdown.
type Subscription ¶
Subscription is one active destination binding.
type Transport ¶
type Transport struct {
// contains filtered or unexported fields
}
Transport reports the message delivery in flight to middleware.
It intentionally does not implement transport.ReplyHeaderer: a delivered message has no reply header. Request/reply, where an adapter supports it, is an adapter capability rather than a property of the envelope.
func (*Transport) Operation ¶
Operation returns the concrete destination that delivered the message, which may differ from a wildcard subscription. It is the message transport's answer to "which call is this", and is opaque to callers.
func (*Transport) RequestHeader ¶
RequestHeader returns the headers of the delivered message.
Source Files