queue

Documentation

Index

• type AckStatus
• type Event
• type EventBus
  • func NewEventBus(ctx context.Context, size int) EventBus
• type EventHandler
• type OutboxEvent
• type OutboxRepository
  • func NewOutboxRepository(transactionFactory transactions.TransactionFactory) *OutboxRepository
  • func (r *OutboxRepository) InsertEvent(ctx context.Context, event *OutboxEvent) error
  • func (r *OutboxRepository) LoadPendingEvents(ctx context.Context) ([]*OutboxEvent, error)
  • func (r *OutboxRepository) MarkEventAsFailed(ctx context.Context, eventID int) error
  • func (r *OutboxRepository) MarkEventAsProcessed(ctx context.Context, eventID int) error
  • func (r *OutboxRepository) UpdateEventStatus(ctx context.Context, event *OutboxEvent) error

Types

type AckStatus

type AckStatus string

const (
	ACK  AckStatus = "ACK"  // message acknowledged, no need to retry.
	NACK AckStatus = "NACK" // message not acknowledged, need to retry.
)

type Event

type Event struct {
	ID        int           // ID is the identifier for the event in the database.
	Data      []byte        // Data is the binary payload of the event.
	Retry     int           // Retry indicates how many times this event has been retried.
	Topic     string        // Topic is the name of the topic to which the event is published.
	NextRetry time.Duration // NextRetry specifies the delay before the next retry attempt.
	AckStatus AckStatus     // AckStatus specifies whether the message is done.
}

Event represents a message or event that can be published to a topic within the EventBus.

type EventBus

type EventBus interface {
	Subscribe(topic string, handler EventHandler, delays []int, durationType time.Duration)
	Publish(topic string, data []byte)
	StartProcessing(ctx context.Context) error
	Stop()
	ExceededMaxRetries(event *Event) bool
	SetLogger(log *clog.CustomLogger)
	AddEventToCtx(ctx context.Context, event *Event) context.Context
	WithOutbox(factory transactions.TransactionFactory)
}

EventBus defines an interface for subscribing to topics, publishing events, and managing event processing.

func NewEventBus

func NewEventBus(ctx context.Context, size int) EventBus

NewEventBus creates a new instance of an eventBus with a specified buffer size for the event queue and attaches a logger.

type EventHandler

type EventHandler func(ctx context.Context, event *Event) AckStatus

EventHandler is a function type that processes an Event and returns an error if the processing fails.

type OutboxEvent

type OutboxEvent struct {
	ID        int       `pg:",pk"`        // Primary key
	Data      []byte    `pg:"data"`       // Data column
	Topic     string    `pg:"topic"`      // Topic column
	Retry     int       `pg:"retry"`      // Retry column
	NextRetry uint      `pg:"next_retry"` // NextRetry column as minutes
	AckStatus AckStatus `pg:"ack_status"` // AckStatus column
	CreatedAt int64     `pg:"created_at"` // CreatedAt column as Unix timestamp
	UpdatedAt int64     `pg:"updated_at"` // UpdatedAt column as Unix timestamp
}

OutboxEvent represents an event stored in the outbox table.

type OutboxRepository

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

OutboxRepository provides methods to interact with the outbox table.

func NewOutboxRepository

func NewOutboxRepository(transactionFactory transactions.TransactionFactory) *OutboxRepository

NewOutboxRepository creates a new OutboxRepository.

func (*OutboxRepository) InsertEvent

func (r *OutboxRepository) InsertEvent(ctx context.Context, event *OutboxEvent) error

InsertEvent inserts a new event into the outbox table or updates it if it already exists.

func (*OutboxRepository) LoadPendingEvents

func (r *OutboxRepository) LoadPendingEvents(ctx context.Context) ([]*OutboxEvent, error)

LoadPendingEvents loads all pending events from the outbox table.

func (*OutboxRepository) MarkEventAsFailed

func (r *OutboxRepository) MarkEventAsFailed(ctx context.Context, eventID int) error

MarkEventAsFailed marks an event as failed in the outbox table.

func (*OutboxRepository) MarkEventAsProcessed

func (r *OutboxRepository) MarkEventAsProcessed(ctx context.Context, eventID int) error

MarkEventAsProcessed marks an event as processed in the outbox table.

func (*OutboxRepository) UpdateEventStatus

func (r *OutboxRepository) UpdateEventStatus(ctx context.Context, event *OutboxEvent) error

UpdateEventStatus updates the status and retry count of an event in the outbox table.