messenger

May 12, 2024

Changes in this version

+ type ErrorHandler interface

+ Error func(ctx context.Context, err error)

+ type Metadata map[string]string

+ func (m *Metadata) Scan(value any) error

+ func (m Metadata) Get(key string) string

+ func (m Metadata) Set(key, value string)

+ func (m Metadata) Value() (driver.Value, error)

type Option

+ func WithErrorHandler(l ErrorHandler) Option

+ type Subscription interface

+ Handle func(ctx context.Context, msg Message) error

+ Name func() string

+ func NewSubscription(name string, h SubscriptionHandler) Subscription

+ type SubscriptionHandler func(ctx context.Context, msg Message) error

Dec 17, 2023

Changes in this version

type GenericMessage

+ MsgAt time.Time

+ MsgID string

+ MsgMetadata map[string]string

+ MsgPayload []byte

+ MsgPublished bool

type Message

+ At func() time.Time

+ Published func() bool

Dec 7, 2023

Changes in this version

type GenericMessage

+ Published bool

+ func (m *GenericMessage) GetAt() time.Time

+ func (m *GenericMessage) GetPublished() bool

type Message

+ GetAt func() time.Time

+ GetPublished func() bool

Jul 17, 2023

Changes in this version

type GenericMessage

+ At time.Time

+ Id string

+ func (m *GenericMessage) GetMetadata() map[string]string

+ func (m *GenericMessage) GetPayload() []byte

+ func (m *GenericMessage) ID() string

+ func (m *GenericMessage) SetMetadata(key, value string) *GenericMessage

type Message

+ GetMetadata func() map[string]string

+ GetPayload func() []byte

+ ID func() string

+ type Messenger struct

+ func NewMessenger(store Store, publisher Publisher, opts ...Option) *Messenger

+ func (w *Messenger) Clean(ctx context.Context) error

+ func (w *Messenger) Publish(ctx context.Context) error

+ func (w *Messenger) Start(ctx context.Context) error

type Option

+ func WithInterval(p time.Duration) Option

+ type Publisher interface

+ Publish func(ctx context.Context, msg Message) error

Feb 10, 2023

Changes in this version

+ type ErrorLogger interface

+ Error func(err error)

+ type Option func(*Worker)

+ func WithCleanUp(period time.Duration, expiration time.Duration) Option

+ func WithErrorLogger(l ErrorLogger) Option

+ func WithPublishBatchSize(bs int) Option

+ func WithPublishPeriod(p time.Duration) Option

+ type Queue interface

+ Publish func(ctx context.Context, msg *store.Message) error

+ type Store interface

+ DeletePublishedByExpiration func(ctx context.Context, exp time.Duration) error

+ Messages func(ctx context.Context, batch int) ([]*store.Message, error)

+ Published func(ctx context.Context, msg ...*store.Message) error

+ type Worker struct

+ func NewWorker(store Store, queue Queue, opts ...Option) *Worker

+ func (w *Worker) Clean(ctx context.Context) error

+ func (w *Worker) Publish(ctx context.Context) error

+ func (w *Worker) Start(ctx context.Context) error

+ func (w *Worker) StartClean(ctx context.Context) error

+ func (w *Worker) StartPublisher(ctx context.Context) error

Dec 27, 2022

Changes in this version

+ var ErrEmptyMessagePayload = errors.New("empty message payload")

+ type GenericMessage struct

+ func NewMessage(payload []byte) (*GenericMessage, error)

+ func (m *GenericMessage) AddMetadata(key, value string)

+ func (m *GenericMessage) Metadata() map[string]string

+ func (m *GenericMessage) Payload() interface{}

+ type Message interface

+ Metadata func() map[string]string

+ Payload func() interface{}