broker

package module
v0.0.1 Latest Latest
Warning

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

 Go to latest Published: Jun 10, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// DefaultCodec is the default codec for broker
	DefaultCodec = encoding.GetCodec("json")
)

Functions

func Creator 

func Creator[T any]() *T

Creator is a generic function that creates a new instance of type T

func Marshal 

func Marshal(codec encoding.Codec, msg any) ([]byte, error)

Marshal encodes a message into bytes using the provided codec.

func Unmarshal 

func Unmarshal(codec encoding.Codec, inputData []byte, outValue any) error

Unmarshal decodes bytes into a message using the provided codec.

Types

type Binder 

type Binder func() any

Binder defines a function that returns an instance for binding message body

type Broker 

type Broker interface {
	// Name returns the broker name
	Name() string

	// Options returns the broker options
	Options() Options

	// Address returns the broker address
	Address() string

	// Init initializes the broker with options
	Init(...Option) error

	// Connect connects to the broker
	Connect() error

	// Disconnect disconnects from the broker
	Disconnect() error

	// Publish publishes a message to a topic
	Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error

	// Subscribe subscribes to a topic with a handler and binder
	Subscribe(topic string, handler Handler, binder Binder, opts ...SubscribeOption) (Subscriber, error)

	// Request sends a request message and waits for a response
	Request(ctx context.Context, topic string, msg *Message, opts ...RequestOption) (*Message, error)
}

Broker defines the message broker interface

type Event 

type Event interface {
	// Topic returns the topic of the event
	Topic() string

	// Message returns the message associated with the event
	Message() *Message
	// RawMessage returns the original/raw message
	RawMessage() any

	// Ack acknowledges the message
	Ack() error

	// Error returns any error associated with the event
	Error() error
}

Event defines the message event interface

type Handler 

type Handler func(ctx context.Context, evt Event) error

Handler defines the handler invoked by subscribers

func ChainSubscriberMiddleware 

func ChainSubscriberMiddleware(h Handler, mws []SubscriberMiddleware) Handler

ChainSubscriberMiddleware chains multiple SubscriberMiddleware into a single Handler.

func TypedToEventHandler 

func TypedToEventHandler[T any](h TypedHandler[T]) Handler

TypedToEventHandler converts a TypedHandler to a generic Handler

type Headers 

type Headers map[string]string

Headers defines message headers as a map of string key-value pairs

type Message 

type Message struct {
	// ID message ID
	ID string

	// Headers message headers
	Headers Headers

	// Body message body
	Body any

	// Key message key (Kafka Key / RocketMQ Key / RabbitMQ RoutingKey...)
	Key string

	// Metadata can hold additional metadata about the message
	Metadata Metadata

	// Partition message partition
	Partition int
	// Offset message offset
	Offset int64

	// Msg original/raw message (kafka.Message, amqp.Delivery...)
	Msg any
}

Message defines the message structure

func NewMessage 

func NewMessage(body any, opts ...MessageOption) *Message

NewMessage creates a new Message with the provided body and options

func NewMessageWithContext 

func NewMessageWithContext(ctx context.Context, body any, keys ...any) *Message

NewMessageWithContext creates a new Message and injects context values into its headers

func (*Message) AckSuccess 

func (m *Message) AckSuccess() error

AckSuccess acknowledges the message if the underlying Msg supports it

func (*Message) BodyBytes 

func (m *Message) BodyBytes() []byte

BodyBytes returns the message body as a byte slice if possible

func (*Message) Clone 

func (m *Message) Clone() *Message

Clone creates a shallow copy of the message

func (*Message) ExtractContext 

func (m *Message) ExtractContext(baseCtx context.Context) context.Context

ExtractContext extracts the message metadata into the provided context

func (*Message) ExtractToContext 

func (m *Message) ExtractToContext(baseCtx context.Context, keyMapping map[string]any) context.Context

ExtractToContext extracts specified headers from the message into the context based on the provided key mapping

func (*Message) GetHeader 

func (m *Message) GetHeader(key string) string

GetHeader returns the value of a specific header key

func (*Message) GetMetadata 

func (m *Message) GetMetadata(key string) any

GetMetadata retrieves the value of a specific metadata key

func (*Message) HeadersCopy 

func (m *Message) HeadersCopy() Headers

HeadersCopy returns a copy of the message headers

func (*Message) InjectContext 

func (m *Message) InjectContext(ctx context.Context, keys ...any) *Message

InjectContext injects values from the context into the message headers based on the provided keys

func (*Message) MergeHeaders 

func (m *Message) MergeHeaders(hdrs Headers)

MergeHeaders merges the provided headers into the message headers

func (*Message) MetadataCopy 

func (m *Message) MetadataCopy() Metadata

MetadataCopy returns a copy of the message metadata

func (*Message) SetDelay 

func (m *Message) SetDelay(level string)

SetDelay sets the delay level for the message

func (*Message) SetHeader 

func (m *Message) SetHeader(key, value string)

SetHeader sets a header key-value pair

func (*Message) SetMetadata 

func (m *Message) SetMetadata(key string, value any)

SetMetadata sets a metadata key-value pair

func (*Message) WithBody 

func (m *Message) WithBody(body any) *Message

WithBody sets the message body and returns the message for chaining

func (*Message) WithHeader 

func (m *Message) WithHeader(key, value string) *Message

WithHeader sets a header key-value pair and returns the message for chaining

func (*Message) WithKey 

func (m *Message) WithKey(key string) *Message

WithKey sets the message key and returns the message for chaining

func (*Message) WithMetadata 

func (m *Message) WithMetadata(key string, value any) *Message

WithMetadata sets a metadata key-value pair and returns the message for chaining

type MessageOption 

type MessageOption func(*Message)

func WithHeader 

func WithHeader(key, value string) MessageOption

WithHeader add single header key-value pair

func WithHeaders 

func WithHeaders(h Headers) MessageOption

WithHeaders will copy the provided headers into Message.Headers (no-op if nil)

func WithID 

func WithID(id string) MessageOption

WithID set the message ID

func WithKey 

func WithKey(key string) MessageOption

WithKey set the message key

func WithMetadata 

func WithMetadata(md Metadata) MessageOption

WithMetadata will copy the provided metadata into Message.Metadata (no-op if nil)

func WithMetadataKV 

func WithMetadataKV(key string, value any) MessageOption

WithMetadataKV add single metadata key-value pair

func WithMsg 

func WithMsg(msg any) MessageOption

WithMsg set the original/raw message

func WithPartitionOffset 

func WithPartitionOffset(partition int, offset int64) MessageOption

WithPartitionOffset set the Partition and Offset of the Message

type Metadata 

type Metadata map[string]any

Metadata defines message metadata as a map of string key-value pairs

func GetMetadataFromContext 

func GetMetadataFromContext(ctx context.Context) Metadata

GetMetadataFromContext retrieves the message metadata from the context

type Option 

type Option func(*Options)

Option defines a function which sets some option.

func OptionContextWithValue 

func OptionContextWithValue(k, v any) Option

OptionContextWithValue sets a value in the broker option context

func WithAddress 

func WithAddress(addressList ...string) Option

WithAddress set broker address

func WithCodec 

func WithCodec(name string) Option

WithCodec set codec, support: json, proto.

func WithEnableSecure 

func WithEnableSecure(enable bool) Option

WithEnableSecure sets enable secure connection

func WithErrorHandler 

func WithErrorHandler(handler Handler) Option

WithErrorHandler sets error handler

func WithGlobalPropagator 

func WithGlobalPropagator() Option

WithGlobalPropagator sets global propagator

func WithGlobalTracerProvider 

func WithGlobalTracerProvider() Option

WithGlobalTracerProvider sets global tracer provider

func WithOptionContext 

func WithOptionContext(ctx context.Context) Option

WithOptionContext sets the broker option context

func WithPropagator 

func WithPropagator(propagator propagation.TextMapPropagator) Option

WithPropagator sets propagator

func WithPublishMiddlewares 

func WithPublishMiddlewares(mws ...PublishMiddleware) Option

WithPublishMiddlewares sets publish middlewares

func WithSubscriberMiddlewares 

func WithSubscriberMiddlewares(mws ...SubscriberMiddleware) Option

WithSubscriberMiddlewares sets subscriber middlewares

func WithTLSConfig 

func WithTLSConfig(config *tls.Config) Option

WithTLSConfig sets tls config for secure connection

func WithTracerProvider 

func WithTracerProvider(provider trace.TracerProvider) Option

WithTracerProvider sets tracer provider

type Options 

type Options struct {
	// Addrs is a Broker addresses
	Addrs []string

	// Codec is a Broker codec
	Codec encoding.Codec

	// ErrorHandler is a Broker error handler
	ErrorHandler Handler

	// Secure enable secure connection
	Secure bool
	// TLSConfig is tls config for secure connection
	TLSConfig *tls.Config

	// Context is broker option context
	Context context.Context

	// Tracings are tracing options
	Tracings []otlp.TracerOption

	// SubscriberMiddlewares applies to subscribe handlers
	SubscriberMiddlewares []SubscriberMiddleware

	// PublishMiddlewares applies to publish handlers
	PublishMiddlewares []PublishMiddleware
}

Options broker options

func NewOptions 

func NewOptions() Options

NewOptions creates default Options.

func NewOptionsAndApply 

func NewOptionsAndApply(opts ...Option) Options

NewOptionsAndApply creates Options and applies given Option functions.

func (*Options) Apply 

func (o *Options) Apply(opts ...Option)

Apply applies all options to the Options.

type PublishHandler 

type PublishHandler func(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error

PublishHandler defines the publishing handler

func ChainPublishMiddleware 

func ChainPublishMiddleware(h PublishHandler, mws []PublishMiddleware) PublishHandler

ChainPublishMiddleware chains multiple PublishMiddleware into a single PublishHandler.

func WrapLegacyPublishHandler 

func WrapLegacyPublishHandler(h func(ctx context.Context, topic string, msg any, opts ...PublishOption) error) PublishHandler

WrapLegacyPublishHandler wraps a legacy PublishHandler to the new PublishHandler

type PublishMiddleware 

type PublishMiddleware func(PublishHandler) PublishHandler

PublishMiddleware defines the publishing middleware

type PublishOption 

type PublishOption func(*PublishOptions)

PublishOption defines a function which sets some publish option.

func PublishContextWithValue 

func PublishContextWithValue(k, v any) PublishOption

PublishContextWithValue sets a value in the publish option context

func WithPublishAsync 

func WithPublishAsync(callback func(error)) PublishOption

WithPublishAsync sets the publishing to be asynchronous with a callback

func WithPublishBodyCodec 

func WithPublishBodyCodec(codec string) PublishOption

WithPublishBodyCodec sets the body codec for publishing

func WithPublishCallback 

func WithPublishCallback(callback func(error)) PublishOption

func WithPublishContext 

func WithPublishContext(ctx context.Context) PublishOption

WithPublishContext sets the context for publishing

func WithPublishRequiredAcks 

func WithPublishRequiredAcks(acks int) PublishOption

WithPublishRequiredAcks sets the number of required acknowledgments for publishing

func WithPublishRetries 

func WithPublishRetries(n int) PublishOption

WithPublishRetries sets the number of retries for publishing

func WithPublishTimeout 

func WithPublishTimeout(d time.Duration) PublishOption

WithPublishTimeout sets the publishing timeout

type PublishOptions 

type PublishOptions struct {
	// Context is published option context
	Context context.Context

	// Timeout is the publishing timeout
	Timeout time.Duration

	// Retries is the number of retries for publishing
	Retries int

	// Async indicates whether to publish asynchronously
	Async bool

	// Callback is the callback function for asynchronous publishing
	Callback func(err error)

	// RequiredAcks indicates the number of required acknowledgments
	RequiredAcks int

	// BodyCodec is the codec used for the message body (e.g., "json", "proto")
	BodyCodec string
}

PublishOptions publish options

func NewPublishOptions 

func NewPublishOptions(opts ...PublishOption) PublishOptions

NewPublishOptions creates default PublishOptions.

func (*PublishOptions) Apply 

func (o *PublishOptions) Apply(opts ...PublishOption)

Apply applies all options to the PublishOptions.

type RequestOption 

type RequestOption func(*RequestOptions)

RequestOption defines a function which sets some request option.

func RequestContextWithValue 

func RequestContextWithValue(k, v any) RequestOption

RequestContextWithValue sets a value in the request option context

func WithReplyTopic 

func WithReplyTopic(topic string) RequestOption

WithReplyTopic sets the reply topic for the request

func WithRequestBodyCodec 

func WithRequestBodyCodec(codec string) RequestOption

WithRequestBodyCodec sets the codec for the request body

func WithRequestContext 

func WithRequestContext(ctx context.Context) RequestOption

WithRequestContext sets the context for the request

func WithRequestTimeout 

func WithRequestTimeout(d time.Duration) RequestOption

WithRequestTimeout sets the timeout for the request

type RequestOptions 

type RequestOptions struct {
	// Context is request option context
	Context context.Context

	// Timeout is the request timeout
	Timeout time.Duration

	// ReplyTopic is the topic to which the reply should be sent
	ReplyTopic string

	// BodyCodec is the codec used for the message body (e.g., "json", "proto")
	BodyCodec string
}

RequestOptions request options

func NewRequestOptions 

func NewRequestOptions(opts ...RequestOption) RequestOptions

NewRequestOptions creates default RequestOptions.

func (*RequestOptions) Apply 

func (o *RequestOptions) Apply(opts ...RequestOption)

Apply applies all options to the RequestOptions.

type SubscribeOption 

type SubscribeOption func(*SubscribeOptions)

SubscribeOption defines a function which sets some subscribe option.

func DisableAutoAck 

func DisableAutoAck() SubscribeOption

DisableAutoAck sets AutoAck to false

func SubscribeContextWithValue 

func SubscribeContextWithValue(k, v any) SubscribeOption

SubscribeContextWithValue sets a value in the subscribe option context

func WithSubscribeAutoAck 

func WithSubscribeAutoAck(b bool) SubscribeOption

WithSubscribeAutoAck sets AutoAck to the given value

func WithSubscribeConcurrency 

func WithSubscribeConcurrency(n int) SubscribeOption

WithSubscribeConcurrency sets the number of concurrent handlers

func WithSubscribeContext 

func WithSubscribeContext(ctx context.Context) SubscribeOption

WithSubscribeContext sets the context for the subscription

func WithSubscribeGroupID 

func WithSubscribeGroupID(id string) SubscribeOption

WithSubscribeGroupID sets the group ID for the subscription (alias for WithQueueName) for kafka compatibility

func WithSubscribeMiddlewares 

func WithSubscribeMiddlewares(mws ...SubscriberMiddleware) SubscribeOption

WithSubscribeMiddlewares sets subscriber middlewares

func WithSubscribeQueueName 

func WithSubscribeQueueName(name string) SubscribeOption

WithSubscribeQueueName sets the queue name for the subscription

func WithSubscribeRetry 

func WithSubscribeRetry(maxRetries int, delay time.Duration) SubscribeOption

WithSubscribeRetry sets the maximum number of retries and delay between retries

type SubscribeOptions 

type SubscribeOptions struct {
	// Context is subscribed option context
	Context context.Context

	// AutoAck indicates whether to automatically acknowledge messages
	AutoAck bool

	// Queue is the name of the queue to subscribe to
	Queue string

	// Concurrency is the number of concurrent handlers
	Concurrency int

	// MaxRetries is the maximum number of retries for message processing
	MaxRetries int

	// RetryDelay is the delay between retries
	RetryDelay time.Duration

	// Middlewares are subscriber middlewares
	Middlewares []SubscriberMiddleware
}

SubscribeOptions subscribe options

func NewSubscribeOptions 

func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions

NewSubscribeOptions creates default SubscribeOptions.

func (*SubscribeOptions) Apply 

func (o *SubscribeOptions) Apply(opts ...SubscribeOption)

Apply applies all options to the SubscribeOptions.

type Subscriber 

type Subscriber interface {
	// Options returns the subscription options
	Options() SubscribeOptions

	// Topic returns the subscribed topic
	Topic() string

	// Unsubscribe unsubscribes from the topic
	Unsubscribe(removeFromManager bool) error
}

Subscriber defines the subscriber interface

func Subscribe 

func Subscribe[T any](broker Broker, topic string, handler TypedHandler[T], opts ...SubscribeOption) (Subscriber, error)

Subscribe is a helper function to subscribe to a topic with a typed handler

type SubscriberMap 

type SubscriberMap map[string]Subscriber

SubscriberMap is a map of topic strings to Subscribers

type SubscriberMiddleware 

type SubscriberMiddleware func(Handler) Handler

SubscriberMiddleware is broker subscriber middleware.

type SubscriberSyncMap 

type SubscriberSyncMap struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

SubscriberSyncMap is a thread-safe map for managing subscribers

func NewSubscriberSyncMap 

func NewSubscriberSyncMap() *SubscriberSyncMap

func (*SubscriberSyncMap) Add 

func (sm *SubscriberSyncMap) Add(topic string, sub Subscriber)

Add adds a Subscriber to the map

func (*SubscriberSyncMap) Clear 

func (sm *SubscriberSyncMap) Clear()

Clear clears all Subscribers from the map and unsubscribes them

func (*SubscriberSyncMap) ClearWithTimeout 

func (sm *SubscriberSyncMap) ClearWithTimeout(timeout time.Duration) []string

ClearWithTimeout clears all Subscribers with a timeout for each unsubscribe operation

func (*SubscriberSyncMap) ForceClear 

func (sm *SubscriberSyncMap) ForceClear()

ForceClear forces clearing the map without unsubscribing

func (*SubscriberSyncMap) Foreach 

func (sm *SubscriberSyncMap) Foreach(fnc func(topic string, sub Subscriber))

Foreach for each Subscriber in the map, applies the given function

func (*SubscriberSyncMap) Get 

func (sm *SubscriberSyncMap) Get(topic string) Subscriber

Get retrieves the Subscriber for a given topic

func (*SubscriberSyncMap) GetAll 

func (sm *SubscriberSyncMap) GetAll() SubscriberMap

GetAll returns a copy of the entire SubscriberMap

func (*SubscriberSyncMap) Has 

func (sm *SubscriberSyncMap) Has(topic string) bool

Has checks if a topic exists in the map

func (*SubscriberSyncMap) Len 

func (sm *SubscriberSyncMap) Len() int

Len returns the number of subscribers in the map

func (*SubscriberSyncMap) Remove 

func (sm *SubscriberSyncMap) Remove(topic string) error

Remove removes a Subscriber from the map and unsubscribes it

func (*SubscriberSyncMap) RemoveOnly 

func (sm *SubscriberSyncMap) RemoveOnly(topic string) bool

RemoveOnly removes a Subscriber from the map without unsubscribing it

func (*SubscriberSyncMap) RemoveWithTimeout 

func (sm *SubscriberSyncMap) RemoveWithTimeout(topic string, timeout time.Duration) error

RemoveWithTimeout removes a Subscriber with a timeout for the unsubscribe operation

func (*SubscriberSyncMap) Topics 

func (sm *SubscriberSyncMap) Topics() []string

Topics returns a slice of all subscribed topics

type TypedHandler 

type TypedHandler[T any] func(ctx context.Context, topic string, headers Headers, msg *T) error

TypedHandler defines a handler function with a specific message type

Source Files

View all Source files

Directories

Path Synopsis
kafka module
stomp module

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.