ebus

package module

Go to main page

Versions in this module

v1

v1.3.0

Sep 26, 2025

Changes in this version

+ func ApplyAllUnsafe[T, ID any](env T, event *Event[T, ID], tx Tx) (handled int, err error)

+ func Clone(b []byte) []byte

+ func CommitAllUnsafe[T, ID any](env T, event *Event[T, ID])

+ func RollbackRangeUnsafe[T, ID any](env T, ev *Event[T, ID], from, to int)

+ func RunAll[T any, ID any, Res any](env T, bus *Bus[T, ID], payloads []Payload[T]) (Res, error)

+ func ValidateAll[T, ID any](env T, event *Event[T, ID]) error

type Bus

+ func (b *Bus[T, ID]) PublishAll(env T, events []Payload[T]) error

+ func (b *Bus[T, ID]) PublishRaw(env T, dec Decoder[T], pt PayloadType, body []byte) error

+ func (b *Bus[T, ID]) PublishRaws(env T, dec Decoder[T], raws ...Raw[ID]) error

+ type ContextCarrier interface

+ Context func() context.Context

+ type Decoder interface

+ Decode func(pt PayloadType, raw []byte) (Payload[T], error)

type EventMiddleware

+ func LoggingByID[T any, ID any](logf func(format string, args ...any)) EventMiddleware[T, ID]

+ func Recovery[T, ID any](logf func(format string, args ...any)) EventMiddleware[T, ID]

+ func RetryWithContext[T any, ID any](retries int, delay time.Duration) EventMiddleware[T, ID]

+ type ExpoJitterPolicy struct

+ Base time.Duration

+ Factor float64

+ Max time.Duration

+ Rand *rand.Rand

+ func (p *ExpoJitterPolicy) Backoff(attempt int) time.Duration

+ func (p *ExpoJitterPolicy) ShouldRetry(attempt int, err error) bool

+ type JSONDecoder struct

+ func NewJSONDecoder[T any]() *JSONDecoder[T]

+ func (d *JSONDecoder[T]) Decode(pt PayloadType, raw []byte) (Payload[T], error)

+ type Raw struct

+ Body []byte

+ Meta struct{ ... }

+ Type PayloadType

+ type RawKeeper interface

+ Raw func() *Raw[ID]

+ SetRaw func(raw *Raw[ID])

+ type RawPayload struct

+ func (p *RawPayload[ID]) Raw() *Raw[ID]

+ func (p *RawPayload[ID]) SetRaw(r *Raw[ID])

+ type Registry struct

+ func NewRegistry[T any]() Registry[T]

+ func (r *Registry[T]) MustRegister(payloadType PayloadType, factory func() Payload[T])

+ func (r *Registry[T]) Register(payloadType PayloadType, factory func() Payload[T]) error

+ type Stager interface

+ PutRaw func(raw *Raw[ID]) error

+ type Transactional interface

+ BeginTx func(readonly bool) (Tx, error)

+ type Tx interface

+ Commit func() error

+ Rollback func() error

v1.1.0

Apr 1, 2025

Changes in this version

+ type EventHooks struct

+ OnAfterCommit func(ctx T, event *Event[T, ID])

+ OnBeforeHandle func(ctx T, event *Event[T, ID])

type EventMiddleware

+ func LoggingById[T any, ID any](logf func(format string, args ...any)) EventMiddleware[T, ID]

+ func LoggingByTypeName[T any, ID any](payloadTypeNameHandler PayloadTypeNameHandler, ...) EventMiddleware[T, ID]

+ func RetryWithFallback[T, ID any](policy RetryPolicy, fallback RetryFallbackHandler[T, ID]) EventMiddleware[T, ID]

+ func RetryWithPolicy[T, ID any](policy RetryPolicy) EventMiddleware[T, ID]

+ func Retry[T, ID any](retries int, delay time.Duration) EventMiddleware[T, ID]

+ func WithEventHooks[T, ID any](hooks EventHooks[T, ID]) EventMiddleware[T, ID]

+ type PayloadTypeNameHandler func(pt PayloadType) string

+ type RetryFallbackHandler func(evt *Event[T, ID], err error)

+ type RetryPolicy interface

+ Backoff func(attempt int) time.Duration

+ ShouldRetry func(attempt int, err error) bool

v1.0.1

Mar 31, 2025

v1.0.0

Mar 31, 2025

Changes in this version

+ var ErrEmptyPayloads = errors.New("empty payloads")

+ var ErrPayloadIsNotCommand = errors.New("payload is not command or not match type")

+ func Dispatcher[T, ID any](ctx T, event *Event[T, ID]) error

+ func Run[T any, ID any, Res any](ctx T, bus *Bus[T, ID], payloads []Payload[T]) (Res, error)

+ type Bus struct

+ func NewDefault[T, ID any](idGenHandler IDGenHandler[T, ID], middlewares ...EventMiddleware[T, ID]) *Bus[T, ID]

+ func NewWithTx[T any, ID any](idGenHandler IDGenHandler[T, ID], txHandler TxHandler[T], ...) *Bus[T, ID]

+ func New[T, ID any](idGenHandler IDGenHandler[T, ID], handler EventHandler[T, ID], ...) *Bus[T, ID]

+ func (b *Bus[T, ID]) Apply(ctx T, event *Event[T, ID]) error

+ func (b *Bus[T, ID]) Publish(ctx T, events []Payload[T]) error

+ type CommandEvent interface

+ Result func() Res

+ type CommitListener func(ctx T, payload Payload[T])

+ type Event struct

+ Created time.Time

+ ID ID

+ Payloads []Payload[T]

+ type EventHandler func(ctx T, event *Event[T, ID]) error

+ func TxDispatcher[T, ID any](inTx TxHandler[T]) EventHandler[T, ID]

+ type EventMiddleware func(next EventHandler[T, ID]) EventHandler[T, ID]

+ type ID8Byte [8]byte

+ type IDGenHandler func(ctx T) (ID, error)

+ func NewID8ByteHandler[T any]() IDGenHandler[T, ID8Byte]

+ type Payload interface

+ Commit func(ctx T)

+ Handle func(ctx T) error

+ PayloadType func() PayloadType

+ Rollback func(ctx T)

+ Validate func(ctx T) error

+ type PayloadType int

+ type Subscribers struct

+ func NewSubscribers[T, ID any]() *Subscribers[T, ID]

+ func (s *Subscribers[T, ID]) Notifier() EventMiddleware[T, ID]

+ func (s *Subscribers[T, ID]) Subscribe(payload Payload[T], handler CommitListener[T])

+ type TxHandler func(ctx T, handler func() error) error

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