queue

Jun 28, 2024

Changes in this version

type PostgresQueueRepository

+ func (r *PostgresQueueRepository) Delete(tx *sql.Tx, m *Message) error

type QueueRepository

+ Delete func(tx *sql.Tx, message *Message) error

Jun 27, 2024

Changes in this version

+ const DefaultNameField

type Message

+ Name string

+ type PostgresQueueRepository struct

+ func (r *PostgresQueueRepository) Create(tx *sql.Tx, m *Message) error

+ func (r *PostgresQueueRepository) CreateTable(tx *sql.Tx) error

+ func (r *PostgresQueueRepository) DeleteAll(tx *sql.Tx) error

+ func (r *PostgresQueueRepository) DeleteOld(tx *sql.Tx, moment time.Time) (int, error)

+ func (r *PostgresQueueRepository) FindNext(tx *sql.Tx) (*Message, error)

+ func (r *PostgresQueueRepository) Save(tx *sql.Tx, m *Message) error

+ type QueueQueryConfig struct

+ CreatedAtField string

+ ErrorField string

+ IdField string

+ IsProcessedField string

+ IsSuccessField string

+ NameField string

+ PayloadField string

+ ProcessedAtField string

+ TableName string

+ type QueueRepository interface

+ Create func(tx *sql.Tx, task *Message) error

+ DeleteOld func(tx *sql.Tx, t time.Time) (int, error)

+ FindNext func(tx *sql.Tx) (*Message, error)

+ Save func(tx *sql.Tx, task *Message) error

+ func NewPostgresQueueRepository(config ...QueueQueryConfig) QueueRepository

Jun 27, 2024

Changes in this version

+ const DefaultCreatedAtField

+ const DefaultProcessedAtField

type Message

+ CreatedAt time.Time

+ ProcessedAt *time.Time

type MessageQueryConfig

+ CreatedAtField string

+ ProcessedAtField string

Jun 26, 2024

Changes in this version

+ const DefaultCreatedField

+ const DefaultProcessedField

type Message

+ Created time.Time

+ Processed *time.Time

Jun 25, 2024

Changes in this version

+ const DefaultCreatedTsField

+ const DefaultErrorField

+ const DefaultIdField

+ const DefaultIsProcessedField

+ const DefaultIsSuccessField

+ const DefaultPayloadField

+ const DefaultProcessedTsField

+ const DefaultQueueField

+ const DefaultTableName

+ type Message struct

+ CreatedTs time.Time

+ Error *string

+ Id int

+ IsProcessed bool

+ IsSuccess *bool

+ Payload string

+ ProcessedTs *time.Time

+ Queue *string

+ func (e Message) LogValue() slog.Value

+ type MessageHandler func(tx *sql.Tx, message *Message) error

+ type MessageQueryConfig struct

+ CreatedTsField string

+ ErrorField string

+ IdField string

+ IsProcessedField string

+ IsSuccessField string

+ PayloadField string

+ ProcessedTsField string

+ QueueField string

+ TableName string

+ type MessageRepository interface

+ Create func(tx *sql.Tx, m *Message) error

+ DeleteProcessed func(tx *sql.Tx, t time.Time) (int, error)

+ FindNext func(tx *sql.Tx) (*Message, error)

+ Save func(tx *sql.Tx, m *Message) error

+ func NewDefaultPostgresMessageRepository() MessageRepository

+ func NewPostgresMessageRepository(conig *MessageQueryConfig) MessageRepository

+ type PostgresMessageRepository struct

+ Config *MessageQueryConfig

+ func (r *PostgresMessageRepository) Create(tx *sql.Tx, m *Message) error

+ func (r *PostgresMessageRepository) CreateTable(tx *sql.Tx) error

+ func (r *PostgresMessageRepository) DeleteAll(tx *sql.Tx) error

+ func (r *PostgresMessageRepository) DeleteProcessed(tx *sql.Tx, t time.Time) (int, error)

+ func (r *PostgresMessageRepository) FindNext(tx *sql.Tx) (*Message, error)

+ func (r *PostgresMessageRepository) Save(tx *sql.Tx, m *Message) error

+ type Worker struct

+ func NewWorker(db *sql.DB, config *WorkerConfig) *Worker

+ func (s *Worker) Start()

+ func (s *Worker) StartContext(ctx context.Context)

+ func (s *Worker) Stop()

+ type WorkerConfig struct

+ Cron string

+ Handler MessageHandler

+ Log *slog.Logger

+ Repository MessageRepository