The highest tagged major version is
Documentation
¶
Overview ¶
Package inbox implements the idempotent-consumer pattern: a dedup table that lets a consumer process each message exactly once under at-least-once delivery. It is the consumer-side complement to the outbox package.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Once ¶
func Once(ctx context.Context, db *bun.DB, messageID string, fn func(ctx context.Context, tx bun.Tx) error) error
Once runs fn exactly once for messageID. It inserts a dedup marker and runs fn in a single transaction: a messageID already recorded skips fn and returns nil; if fn returns an error the transaction (marker included) rolls back, so the message can be reprocessed later. For that guarantee, fn must perform its work through the provided tx.
Example ¶
Process a message exactly once, even if it is delivered twice (at-least-once).
package main
import (
"context"
"database/sql"
"fmt"
"github.com/uptrace/bun"
"github.com/uptrace/bun/dialect/sqlitedialect"
"github.com/uptrace/bun/driver/sqliteshim"
"github.com/rahmadafandi/fibr/inbox"
)
func main() {
sqldb, _ := sql.Open(sqliteshim.ShimName, "file::memory:?cache=shared")
db := bun.NewDB(sqldb, sqlitedialect.New())
defer db.Close()
ctx := context.Background()
_ = inbox.Migrate(ctx, db)
handle := func(_ context.Context, _ bun.Tx) error {
fmt.Println("processing order.created")
return nil
}
// Same message delivered twice.
_ = inbox.Once(ctx, db, "order.created:42", handle)
_ = inbox.Once(ctx, db, "order.created:42", handle)
}
Output: processing order.created
Types ¶
Source Files
Click to show internal directories.
Click to hide internal directories.