outbox

Documentation

Index

• Constants
• Variables
• type Message
  • func (m Message) ToMessage() (msg.Message, error)
• type MessageProcessor
• type MessageStore
• type PollingProcessor
  • func NewPollingProcessor(in MessageStore, out msg.MessagePublisher, options ...PollingProcessorOption) *PollingProcessor
  • func (p *PollingProcessor) Start(ctx context.Context) error
  • func (p *PollingProcessor) Stop(ctx context.Context) (err error)
• type PollingProcessorOption
  • func WithPollingProcessorLogger(logger log.Logger) PollingProcessorOption
  • func WithPollingProcessorMessagesPerPolling(messagesPerPolling int) PollingProcessorOption
  • func WithPollingProcessorPollingInterval(pollingInterval time.Duration) PollingProcessorOption
  • func WithPollingProcessorPurgeInterval(purgeInterval time.Duration) PollingProcessorOption
  • func WithPollingProcessorPurgeOlderThan(purgeOtherThan time.Duration) PollingProcessorOption
  • func WithPollingProcessorRetryer(retryer retry.Retryer) PollingProcessorOption

Constants

const (
	DefaultMessagesPerPolling = 500
	DefaultPollingInterval    = 500 * time.Millisecond
	DefaultPurgeOlderThan     = 60 * time.Second
	DefaultPurgeInterval      = 30 * time.Second

	DefaultMaxRetries               = 100
	DefaultRetryMultiplier          = 1.25
	DefaultRetryRandomizationFactor = 0.33
)

Variables

var DefaultRetryer = retry.NewExponentialBackoff(
	retry.WithBackoffInitialInterval(DefaultPollingInterval),
	retry.WithBackoffMaxRetries(DefaultMaxRetries),
	retry.WithBackoffMultiplier(DefaultRetryMultiplier),
	retry.WithBackoffRandomizationFactor(DefaultRetryRandomizationFactor),
)

Types

type Message

type Message struct {
	MessageID   string
	Destination string
	Payload     []byte
	Headers     []byte
}

func (Message) ToMessage

func (m Message) ToMessage() (msg.Message, error)

type MessageProcessor

type MessageProcessor interface {
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
}

type MessageStore

type MessageStore interface {
	Fetch(ctx context.Context, limit int) ([]Message, error)
	Save(ctx context.Context, message Message) error
	MarkPublished(ctx context.Context, messageIDs []string) error
	PurgePublished(ctx context.Context, olderThan time.Duration) error
}

type PollingProcessor

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

func NewPollingProcessor

func NewPollingProcessor(in MessageStore, out msg.MessagePublisher, options ...PollingProcessorOption) *PollingProcessor

func (*PollingProcessor) Start

func (p *PollingProcessor) Start(ctx context.Context) error

func (*PollingProcessor) Stop

func (p *PollingProcessor) Stop(ctx context.Context) (err error)

type PollingProcessorOption

type PollingProcessorOption func(*PollingProcessor)

func WithPollingProcessorLogger

func WithPollingProcessorLogger(logger log.Logger) PollingProcessorOption

func WithPollingProcessorMessagesPerPolling

func WithPollingProcessorMessagesPerPolling(messagesPerPolling int) PollingProcessorOption

func WithPollingProcessorPollingInterval

func WithPollingProcessorPollingInterval(pollingInterval time.Duration) PollingProcessorOption

func WithPollingProcessorPurgeInterval

func WithPollingProcessorPurgeInterval(purgeInterval time.Duration) PollingProcessorOption

func WithPollingProcessorPurgeOlderThan

func WithPollingProcessorPurgeOlderThan(purgeOtherThan time.Duration) PollingProcessorOption

func WithPollingProcessorRetryer

func WithPollingProcessorRetryer(retryer retry.Retryer) PollingProcessorOption