events

package

v0.9.82 Latest Latest Go to latest Published: Sep 14, 2026 License: MIT Imports: 18 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/go42-dev/go42

Links

Open Source Insights

Documentation ¶

Index ¶

Variables
func ContextWithPropagation(ctx context.Context, fields map[string]string) context.Context
func IsPermanent(err error) bool
func Permanent(err error) error
func PropagationFromContext(ctx context.Context) map[string]string
type Backend
type NoopEngine
- func NewNoop() *NoopEngine
type Option
type Publisher
type Router
- func NewRouter(backend Backend, opts ...Option) (*Router, error)
type Subscriber
type TopicInitializer

Constants ¶

This section is empty.

Variables ¶

View Source

var ErrPublishCapacity = errors.New("publish capacity unavailable")

ErrPublishCapacity means waiting for an available publisher slot failed.

Functions ¶

func ContextWithPropagation ¶ added in v0.9.79

func ContextWithPropagation(ctx context.Context, fields map[string]string) context.Context

ContextWithPropagation restores correlation onto the consumer's context, retaining its local values, cancellation and deadline.

func IsPermanent ¶ added in v0.9.79

func IsPermanent(err error) bool

IsPermanent identifies invalid messages that retrying cannot repair.

func Permanent ¶

func Permanent(err error) error

func PropagationFromContext ¶ added in v0.9.79

func PropagationFromContext(ctx context.Context) map[string]string

PropagationFromContext captures the request ID and W3C trace context for asynchronous work. W3C trace context is propagated even when local span recording is disabled.

Types ¶

type Backend ¶

type Backend interface {
	Publisher() message.Publisher
	Subscriber() message.Subscriber
	Shutdown(ctx context.Context) error
}

type NoopEngine ¶

type NoopEngine struct{}

func NewNoop ¶

func NewNoop() *NoopEngine

func (*NoopEngine) Close ¶

func (e *NoopEngine) Close() error

func (*NoopEngine) Publish ¶

func (e *NoopEngine) Publish(_ string, _ ...*message.Message) error

func (*NoopEngine) Publisher ¶

func (e *NoopEngine) Publisher() message.Publisher

func (*NoopEngine) Shutdown ¶

func (e *NoopEngine) Shutdown(_ context.Context) error

func (e *NoopEngine) Subscribe(
	ctx context.Context,
	_ string,
) (<-chan *message.Message, error)

func (*NoopEngine) Subscriber ¶

func (e *NoopEngine) Subscriber() message.Subscriber

type Option ¶

type Option func(*Router)

func WithCloseTimeout ¶ added in v0.9.79

func WithCloseTimeout(timeout time.Duration) Option

WithCloseTimeout sets how long shutdown waits for message handlers. Zero uses the default of 30 seconds.

func WithDeadLetterTopicSuffix ¶ added in v0.9.79

func WithDeadLetterTopicSuffix(suffix string) Option

WithDeadLetterTopicSuffix sets the suffix appended to topics for failed messages.

func WithInitialBackoff ¶ added in v0.9.79

func WithInitialBackoff(backoff time.Duration) Option

WithInitialBackoff sets the initial backoff interval for consumer retries.

func WithLogger ¶

func WithLogger(logger *slog.Logger) Option

func WithMaxBackoff ¶ added in v0.9.79

func WithMaxBackoff(backoff time.Duration) Option

WithMaxBackoff sets the maximum backoff interval for consumer retries.

func WithMaxRetries ¶ added in v0.9.79

func WithMaxRetries(maxRetries int) Option

WithMaxRetries sets the number of consumer retries after the first attempt. Zero disables retries.

func WithPublishMaxInflight ¶ added in v0.9.79

func WithPublishMaxInflight(limit int) Option

WithPublishMaxInflight sets the maximum number of active broker calls shared by regular and dead-letter publishing. It must be positive and defaults to 32. Calls that outlive their context occupy capacity until the backend returns.

type Publisher ¶

type Publisher interface {
	Publish(ctx context.Context, topic string, id string, event []byte) error
}

type Router ¶

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

func NewRouter ¶

func NewRouter(backend Backend, opts ...Option) (*Router, error)

func (*Router) Errors ¶ added in v0.9.79

func (r *Router) Errors() <-chan error

Errors reports at most one unexpected termination after Start succeeds. It closes when the router stops; normal shutdown closes it without an error.

func (*Router) Publish ¶

func (r *Router) Publish(ctx context.Context, topic string, id string, event []byte) error

Publish requires an event ID. Reuse the same ID when retrying an event.

func (*Router) Shutdown ¶

func (r *Router) Shutdown(ctx context.Context) error

func (*Router) Start ¶

func (r *Router) Start(ctx context.Context) error

Start initializes subscriptions and begins processing messages. Call it once, after registering all handlers with Subscribe. The caller must exit on error; failed startup does not roll back subscriptions.

func (r *Router) Subscribe(
	topic string,
	handler func(ctx context.Context, event []byte) error,
) error

type Subscriber ¶

type Subscriber interface {
	Subscribe(
		topic string,
		handler func(ctx context.Context, event []byte) error,
	) error
}

type TopicInitializer ¶

type TopicInitializer interface {
	InitializeTopic(topic string) error
}

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
gochan
kafka
nats
rabbitmq

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL