postgresq

Documentation

Index

• Constants
• type NotificationMessage
  • func (nm *NotificationMessage) FromDomain(domainMessage notification.Message)
  • func (nm *NotificationMessage) ToDomain() notification.Message
• type Queue
  • func New(logger log.Logger, dbConfig db.Config, opts ...QueueOption) (*Queue, error)
  • func (q *Queue) Cleanup(ctx context.Context, filter queues.FilterCleanup) error
  • func (q *Queue) Dequeue(ctx context.Context, receiverTypes []string, batchSize int, ...) error
  • func (q *Queue) Enqueue(ctx context.Context, ms ...notification.Message) error
  • func (q *Queue) ErrorCallback(ctx context.Context, ms notification.Message) error
  • func (q *Queue) ListMessages(ctx context.Context, notificationID string) ([]notification.Message, error)
  • func (q *Queue) Stop(ctx context.Context) error
  • func (q *Queue) SuccessCallback(ctx context.Context, ms notification.Message) error
• type QueueOption
  • func WithStrategy(s Strategy) QueueOption
• type Strategy

Constants

const (
	MessageQueueTableName     = "message_queue"
	MessageQueueSchemaName    = "notification"
	MessageQueueTableFullName = MessageQueueSchemaName + "." + MessageQueueTableName
)

Types

type NotificationMessage

type NotificationMessage struct {
	ID              string         `db:"id"`
	NotificationIDs pgc.ListString `db:"notification_ids"`

	Status string `db:"status"`

	ReceiverType string           `db:"receiver_type"`
	Configs      pgc.StringAnyMap `db:"configs"`
	Details      pgc.StringAnyMap `db:"details"`
	Metadata     pgc.StringAnyMap `db:"metadata"`
	LastError    sql.NullString   `db:"last_error"`

	MaxTries  int  `db:"max_tries"`
	TryCount  int  `db:"try_count"`
	Retryable bool `db:"retryable"`

	ExpiredAt sql.NullTime `db:"expired_at"`
	CreatedAt time.Time    `db:"created_at"`
	UpdatedAt time.Time    `db:"updated_at"`
}

func (*NotificationMessage) FromDomain

func (nm *NotificationMessage) FromDomain(domainMessage notification.Message)

func (*NotificationMessage) ToDomain

func (nm *NotificationMessage) ToDomain() notification.Message

type Queue

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

func New

func New(logger log.Logger, dbConfig db.Config, opts ...QueueOption) (*Queue, error)

New creates a new queue instance

func (*Queue) Cleanup

func (q *Queue) Cleanup(ctx context.Context, filter queues.FilterCleanup) error

func (*Queue) Dequeue

func (q *Queue) Dequeue(ctx context.Context, receiverTypes []string, batchSize int, handlerFn func(context.Context, []notification.Message) error) error

Dequeue pop the queue based on specific filters (receiver types or batch size) and process the messages with handlerFn message left in pending state that has expired or been updated long time ago means there was a failure when transforming row into a struct

func (*Queue) Enqueue

func (q *Queue) Enqueue(ctx context.Context, ms ...notification.Message) error

Enqueue pushes messages to the queue

func (*Queue) ErrorCallback

func (q *Queue) ErrorCallback(ctx context.Context, ms notification.Message) error

ErrorCallback is a callback that will be called once the message is failed to be handled by handlerFn

func (*Queue) ListMessages

func (q *Queue) ListMessages(ctx context.Context, notificationID string) ([]notification.Message, error)

func (*Queue) Stop

func (q *Queue) Stop(ctx context.Context) error

Stop will close the db

func (*Queue) SuccessCallback

func (q *Queue) SuccessCallback(ctx context.Context, ms notification.Message) error

SuccessCallback is a callback that will be called once the message is succesfully handled by handlerFn

type QueueOption

type QueueOption func(*Queue)

func WithStrategy

func WithStrategy(s Strategy) QueueOption

type Strategy

type Strategy string

const (
	StrategyDefault Strategy = "default"
	StrategyDLQ     Strategy = "dlq"
)