dedup

package

v0.2.3 Latest Latest Go to latest Published: Mar 20, 2026 License: Apache-2.0, MIT Imports: 13 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/storacha/piri

Links

Open Source Insights

Documentation ¶

Index ¶

Variables
func Setup(ctx context.Context, db *sql.DB) error
func SetupPostgres(ctx context.Context, db *sql.DB) error
func SetupWithDialect(ctx context.Context, db *sql.DB, d dialect.Dialect) error
type HashFunc
type NewOpts
type Queue
- func New(opts NewOpts) (*Queue, error)

Constants ¶

This section is empty.

Variables ¶

View Source

var SchemaPostgres string

View Source

var SchemaSQLite string

Functions ¶

func Setup ¶

func Setup(ctx context.Context, db *sql.DB) error

Setup sets up the dedup queue schema using SQLite dialect (default).

func SetupPostgres ¶ added in v0.2.3

func SetupPostgres(ctx context.Context, db *sql.DB) error

SetupPostgres sets up the dedup queue schema using PostgreSQL dialect.

func SetupWithDialect ¶ added in v0.2.3

func SetupWithDialect(ctx context.Context, db *sql.DB, d dialect.Dialect) error

SetupWithDialect sets up the dedup queue schema using the specified dialect.

Types ¶

type HashFunc ¶

type HashFunc func([]byte) []byte

type NewOpts ¶

type NewOpts struct {
	DB                *sql.DB
	Name              string
	MaxReceive        int
	Timeout           time.Duration
	DedupeEnabled     *bool
	BlockRepeatsOnDLQ *bool
	HashFunc          HashFunc
	Logger            logger.StandardLogger
	Dialect           dialect.Dialect
}

type Queue ¶

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

func New ¶

func New(opts NewOpts) (*Queue, error)

func (*Queue) Delete ¶

func (q *Queue) Delete(ctx context.Context, id queue.ID) error

func (*Queue) Extend ¶

func (q *Queue) Extend(ctx context.Context, id queue.ID, delay time.Duration) error

func (*Queue) MaxReceive ¶

func (q *Queue) MaxReceive() int

func (*Queue) MoveToDeadLetter ¶

func (q *Queue) MoveToDeadLetter(ctx context.Context, id queue.ID, jobName, failureReason, errorMsg string) error

func (*Queue) Receive ¶

func (q *Queue) Receive(ctx context.Context) (*queue.Message, error)

func (*Queue) ReceiveAndWait ¶

func (q *Queue) ReceiveAndWait(ctx context.Context, interval time.Duration) (*queue.Message, error)

func (*Queue) Send ¶

func (q *Queue) Send(ctx context.Context, m queue.Message) error

func (*Queue) SendAndGetID ¶

func (q *Queue) SendAndGetID(ctx context.Context, m queue.Message) (queue.ID, error)

func (*Queue) SendTx ¶

func (q *Queue) SendTx(ctx context.Context, tx *sql.Tx, m queue.Message) error

func (*Queue) Timeout ¶

func (q *Queue) Timeout() time.Duration

Source Files ¶

View all Source files

queue.go

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL