Documentation
¶
Overview ¶
Package message provides protobuf-first integration message primitives for direct, non-transactional communication across boundaries.
The package is separate from ddd/event. Domain events model facts raised inside a bounded context and are usually dispatched after persistence. Integration messages in this package wrap protobuf payloads with routing and trace metadata for direct publication without transactional guarantees.
Message Kind is a semantic contract identifier used for handler routing and payload resolution. It is not a topic, subject, queue, partition, or routing key, although providers may map it to those concepts.
Subscriber only means handler registration. Broker runtime concerns such as polling, acknowledgement, offset commit, retry, redelivery, and dead-letter routing belong to provider or application code.
Handler failures and provider runtime failures are separate layers. Handler.Handle returns message-level failures for one delivered integration message. Runner.Run returns why a provider runtime loop terminated. Context cancellation or expiration returns ctx.Err(); non-context Run errors mean the runtime cannot continue safely.
Message ID, Kind, OccurredAt, Key, and Headers are transport-neutral fields. Providers decide how to encode them into their own envelopes. This package intentionally does not define Kafka-style header names.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( ErrEmptyKind = errors.New("message kind is empty") ErrNilPayload = errors.New("message payload is nil") ErrEmptyID = errors.New("message id is empty") ErrNilHandler = errors.New("message handler is nil") ErrNoListening = errors.New("message handler listens to no kinds") ErrUnhandledKind = errors.New("message kind is unhandled") ErrUnknownKind = errors.New("unknown message kind") // ErrNilPayloadResolver means no resolver function was configured. ErrNilPayloadResolver = errors.New("message payload resolver is nil") // ErrNilPayloadFactory means a resolver or registry factory could not // allocate a protobuf target for decoding. It is distinct from // ErrNilPayload, which applies to constructing a Message with an invalid // payload value. ErrNilPayloadFactory = errors.New("message payload factory is nil") )
Functions ¶
This section is empty.
Types ¶
type Handler ¶
Handler processes integration messages for the kinds returned by Listening.
Handle returns nil when the message has been accepted and fully handled by this handler. A provider may ack, commit, or otherwise mark delivery complete after all matching handlers return nil.
A non-nil error is a message-level failure: this specific message was not successfully handled. Providers may apply their own retry, redelivery, dead-letter, drop, or failure-recording policy. A Handler error is not, by itself, a Runner runtime-loop failure. Business outcomes that should not use provider failure handling should be handled inside the handler and then return nil.
type Kind ¶
type Kind string
Kind is the semantic integration message contract type.
Kind is used for handler matching and payload resolution. It is not a broker topic, queue, subject, exchange, partition, or routing-key contract. Provider adapters may map a Kind to their own envelope address.
type Message ¶
type Message struct {
// contains filtered or unexported fields
}
Message is a protobuf integration DTO plus transport-neutral metadata.
func (Message) OccurredAt ¶
OccurredAt returns the time the represented business fact occurred.
type Option ¶
type Option func(*messageConfig)
func WithHeader ¶
func WithHeaders ¶
func WithOccurredAt ¶
type PayloadRegistry ¶ added in v0.9.1
type PayloadRegistry struct {
// contains filtered or unexported fields
}
PayloadRegistry is an in-memory Kind-to-protobuf factory registry.
It is transport-neutral: Kafka topics, RabbitMQ routing keys, NATS subjects, offsets, acknowledgements, retries, and dead-letter behavior belong to provider/application code. The registry only maps a semantic message kind to the protobuf DTO type needed to decode that kind's payload.
func NewPayloadRegistry ¶ added in v0.9.1
func NewPayloadRegistry() *PayloadRegistry
NewPayloadRegistry creates an empty payload registry.
type PayloadResolver ¶ added in v0.9.1
PayloadResolver allocates an empty protobuf payload for a message kind.
Broker and storage adapters use the returned message as the target for unmarshalling bytes into the protobuf DTO contract identified by Kind. Returning nil or a typed-nil protobuf message is a resolver configuration error reported as ErrNilPayloadFactory.
type PayloadResolverFunc ¶ added in v0.9.1
PayloadResolverFunc adapts a function into a PayloadResolver.
type Router ¶
type Router struct {
// contains filtered or unexported fields
}
Router registers handlers by Kind and dispatches messages to them sequentially.
Example ¶
package main
import (
"context"
"fmt"
"time"
"github.com/go-jimu/components/ddd/message"
"google.golang.org/protobuf/types/known/wrapperspb"
)
type exampleMessageHandler struct {
seen *[]string
}
func (h exampleMessageHandler) Listening() []message.Kind {
return []message.Kind{"customer.created"}
}
func (h exampleMessageHandler) Handle(_ context.Context, msg message.Message) error {
*h.seen = append(*h.seen, msg.Key())
return nil
}
func main() {
router := message.NewRouter()
seen := make([]string, 0, 1)
if err := router.Subscribe(exampleMessageHandler{seen: &seen}); err != nil {
panic(err)
}
msg, err := message.New(
"customer.created",
wrapperspb.String("alice"),
message.WithID("msg-1"),
message.WithKey("customer-1"),
message.WithOccurredAt(time.Unix(0, 0).UTC()),
)
if err != nil {
panic(err)
}
if err := router.Handle(context.Background(), msg); err != nil {
panic(err)
}
fmt.Println(msg.ID(), msg.Kind(), seen[0])
}
Output: msg-1 customer.created customer-1
func (*Router) Handle ¶
Handle dispatches msg to all handlers registered for msg.Kind().
Handlers run sequentially in subscription order. Routing stops at the first handler error and returns that error so the caller can avoid acknowledging a partially handled message. ErrUnhandledKind is returned when no handler is registered for the message kind.
type Runner ¶ added in v0.9.1
Runner is an optional runtime loop capability for providers that actively consume messages.
Run blocks until the provider runtime loop terminates. If ctx is canceled or expires, Run returns ctx.Err(). A non-context error returned by Run means the provider runtime cannot continue safely, for example because polling, acknowledgement, commit, or provider-owned failure handling failed.
Run errors are runtime-level failures. Handler.Handle errors are message-level failures and should be handled by the provider's documented message failure policy rather than being treated as Run failures by default.
type Subscriber ¶
Subscriber registers message handlers.
Subscribe is a handler registration operation only. It does not imply that a broker consumer has started polling, reserved partitions, acknowledged records, committed offsets, or joined a consumer group. Providers that own a runtime loop should expose that separately, for example by implementing Runner.
Source Files
Directories