bus

README

Bus

Installation

go get -u github.com/gowool/bus

License

Distributed under MIT License, please see license file within the code for more details.

Documentation

Index

• Variables
• func HandleError(ctx context.Context, errCh <-chan error, handler ErrorHandler)
• type Bus
• type Error
  • func NewError(name, group, consumer string) *Error
  • func (e *Error) Error() string
  • func (e *Error) Is(target error) bool
  • func (e *Error) SetErr(err error) *Error
  • func (e *Error) SetEventDate(eventDate time.Time) *Error
  • func (e *Error) SetEventID(eventID uuid.UUID) *Error
  • func (e *Error) SetID(id string) *Error
  • func (e *Error) Unwrap() error
• type ErrorHandler
• type Event
• type Handler
• type HandlerFunc
  • func (h HandlerFunc) Handle(ctx context.Context, event Event, additional map[string]any) error
• type MetricsBus
  • func NewMetricsBus(inner Bus) (*MetricsBus, error)
  • func (b *MetricsBus) Publish(ctx context.Context, event Event) error
  • func (b *MetricsBus) Subscribe(ctx context.Context, name string, handler Handler) error
• type Middleware
• type Option
  • func WithClientID(clientID string) Option
  • func WithLogError(logError func(err error)) Option
  • func WithPublishOption(options ...PublishOption) Option
• type PublishOption
  • func WithApprox(approx bool) PublishOption
  • func WithApproxMaxLen(maxLen int64) PublishOption
  • func WithID(id string) PublishOption
  • func WithLimit(limit int64) PublishOption
  • func WithMaxLen(maxLen int64) PublishOption
  • func WithMinID(minID string) PublishOption
  • func WithValues(values map[string]interface{}) PublishOption
• type Publisher
• type RedisBus
  • func NewRedisBus(client redis.UniversalClient, options ...Option) *RedisBus
  • func (b *RedisBus) Errors() <-chan error
  • func (b *RedisBus) Middleware(middlewares ...Middleware)
  • func (b *RedisBus) Publish(ctx context.Context, event Event) error
  • func (b *RedisBus) Stop(ctx context.Context) error
  • func (b *RedisBus) Subscribe(ctx context.Context, name string, handler Handler) error
• type Subscriber
• type TracingBus
  • func NewTracingBus(inner Bus) *TracingBus
  • func (b *TracingBus) Publish(ctx context.Context, event Event) error
  • func (b *TracingBus) Subscribe(ctx context.Context, name string, handler Handler) error

Variables

var (
	ErrMissingID           = errors.New("missing id")
	ErrMissingDate         = errors.New("missing date")
	ErrMissingName         = errors.New("missing name")
	ErrMissingHandler      = errors.New("missing handler")
	ErrHandlerAlreadyAdded = errors.New("handler already added")
)

Functions

func HandleError

func HandleError(ctx context.Context, errCh <-chan error, handler ErrorHandler)

Types

type Bus

type Bus interface {
	Publisher
	Subscriber
}

type Error

type Error struct {
	ID        string    `json:"id,omitempty"`
	Name      string    `json:"name,omitempty"`
	Group     string    `json:"group,omitempty"`
	Consumer  string    `json:"consumer,omitempty"`
	Err       error     `json:"error,omitempty"`
	EventID   uuid.UUID `json:"event_id,omitempty"`
	EventDate time.Time `json:"event_date,omitempty"`
}

func NewError

func NewError(name, group, consumer string) *Error

func (*Error) Error

func (e *Error) Error() string

func (*Error) Is

func (e *Error) Is(target error) bool

func (*Error) SetErr

func (e *Error) SetErr(err error) *Error

func (*Error) SetEventDate

func (e *Error) SetEventDate(eventDate time.Time) *Error

func (*Error) SetEventID

func (e *Error) SetEventID(eventID uuid.UUID) *Error

func (*Error) SetID

func (e *Error) SetID(id string) *Error

func (*Error) Unwrap

func (e *Error) Unwrap() error

type ErrorHandler

type ErrorHandler func(ctx context.Context, err error)

type Event

type Event interface {
	ID() uuid.UUID
	Date() time.Time
	Name() string
	Data() []byte
}

type Handler

type Handler interface {
	Handle(ctx context.Context, event Event, additional map[string]any) error
}

type HandlerFunc

type HandlerFunc func(ctx context.Context, event Event, additional map[string]any) error

func (HandlerFunc) Handle

func (h HandlerFunc) Handle(ctx context.Context, event Event, additional map[string]any) error

type MetricsBus

type MetricsBus struct {
	Bus
	// contains filtered or unexported fields
}

func NewMetricsBus

func NewMetricsBus(inner Bus) (*MetricsBus, error)

func (*MetricsBus) Publish

func (b *MetricsBus) Publish(ctx context.Context, event Event) error

func (*MetricsBus) Subscribe

func (b *MetricsBus) Subscribe(ctx context.Context, name string, handler Handler) error

type Middleware

type Middleware func(ctx context.Context, event Event, additional map[string]any, next Handler) error

type Option

type Option func(b *RedisBus)

func WithClientID

func WithClientID(clientID string) Option

func WithLogError

func WithLogError(logError func(err error)) Option

func WithPublishOption

func WithPublishOption(options ...PublishOption) Option

type PublishOption

type PublishOption func(a *addArgs)

func WithApprox

func WithApprox(approx bool) PublishOption

func WithApproxMaxLen

func WithApproxMaxLen(maxLen int64) PublishOption

func WithID

func WithID(id string) PublishOption

func WithLimit

func WithLimit(limit int64) PublishOption

func WithMaxLen

func WithMaxLen(maxLen int64) PublishOption

func WithMinID

func WithMinID(minID string) PublishOption

func WithValues

func WithValues(values map[string]interface{}) PublishOption

type Publisher

type Publisher interface {
	Publish(ctx context.Context, event Event) error
}

type RedisBus

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

func NewRedisBus

func NewRedisBus(client redis.UniversalClient, options ...Option) *RedisBus

func (*RedisBus) Errors

func (b *RedisBus) Errors() <-chan error

func (*RedisBus) Middleware

func (b *RedisBus) Middleware(middlewares ...Middleware)

func (*RedisBus) Publish

func (b *RedisBus) Publish(ctx context.Context, event Event) error

func (*RedisBus) Stop

func (b *RedisBus) Stop(ctx context.Context) error

func (*RedisBus) Subscribe

func (b *RedisBus) Subscribe(ctx context.Context, name string, handler Handler) error

type Subscriber

type Subscriber interface {
	Middleware(middlewares ...Middleware)
	Subscribe(ctx context.Context, name string, handler Handler) error
	Stop(ctx context.Context) error
	Errors() <-chan error
}

type TracingBus

type TracingBus struct {
	Bus
	// contains filtered or unexported fields
}

func NewTracingBus

func NewTracingBus(inner Bus) *TracingBus

func (*TracingBus) Publish

func (b *TracingBus) Publish(ctx context.Context, event Event) error

func (*TracingBus) Subscribe

func (b *TracingBus) Subscribe(ctx context.Context, name string, handler Handler) error