Documentation
¶
Overview ¶
Package outbox implements the transactional-outbox pattern: events are written into an outbox table within the same database transaction as the business data, and a background Relay publishes pending events at-least-once.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Enqueue ¶
Enqueue JSON-encodes payload and inserts a pending Event using db, which may be a *bun.DB or a bun.Tx. Pass your business transaction so the event commits atomically with your writes.
func Migrate ¶
Migrate creates the outbox table (if it does not exist) with an index on (published_at, id) so pending-event scans stay cheap, and the outbox_dead table used by WithMaxAttempts.
Types ¶
type DeadEvent ¶
type DeadEvent struct {
bun.BaseModel `bun:"table:outbox_dead,alias:od"`
ID int64 `bun:"id,pk"`
Topic string `bun:"topic,notnull"`
Payload []byte `bun:"payload,notnull"`
CreatedAt time.Time `bun:"created_at,notnull"`
FailedAt time.Time `bun:"failed_at,notnull"`
Error string `bun:"error,notnull"`
}
DeadEvent is an event the relay gave up on (see WithMaxAttempts). It keeps the original ID, topic, payload, and creation time, plus when and why publishing was abandoned. Re-enqueue it with Enqueue to retry.
type Event ¶
type Event struct {
bun.BaseModel `bun:"table:outbox,alias:o"`
ID int64 `bun:"id,pk,autoincrement"`
Topic string `bun:"topic,notnull"`
Payload []byte `bun:"payload,notnull"`
CreatedAt time.Time `bun:"created_at,notnull"`
PublishedAt *time.Time `bun:"published_at"`
}
Event is one row of the outbox. Payload holds the JSON-encoded event body. PublishedAt is nil while the event is pending and set once the relay has published it.
type Publisher ¶
Publisher publishes a single event's raw payload bytes to its topic.
func NewRedisPublisher ¶
func NewRedisPublisher(client redis.UniversalClient) Publisher
NewRedisPublisher returns a Publisher that publishes the raw payload bytes on the topic channel via client.Publish, without re-encoding. The channel message is therefore exactly the stored JSON, so redis.Subscribe[T] on the consumer side decodes it directly.
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay reads pending events from the outbox and publishes them.
func NewRelay ¶
func NewRelay(db *bun.DB, pub Publisher, opts ...RelayOption) *Relay
NewRelay returns a Relay that publishes outbox events from db via pub.
func (*Relay) Process ¶
Process publishes one batch of pending events, oldest first, up to the configured batch size. Each event's PublishedAt is set only after its Publish succeeds; a Publish error stops the batch and is returned, leaving the failed event and the rest pending (unless WithMaxAttempts dead-letters it, in which case the batch continues). It returns the number of events published. When a Locker is configured and another replica holds the lock, Process returns (0, nil). Process must not be called concurrently on the same Relay.
Example ¶
Write an event in the same transaction as the business data, then let the relay publish it. If the transaction rolls back, no event is ever published.
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/v2/outbox"
)
// recordingPublisher prints each event it would publish.
type recordingPublisher struct{}
func (recordingPublisher) Publish(_ context.Context, topic string, payload []byte) error {
fmt.Printf("publish %s %s\n", topic, payload)
return nil
}
// Write an event in the same transaction as the business data, then let the
// relay publish it. If the transaction rolls back, no event is ever published.
func main() {
sqldb, _ := sql.Open(sqliteshim.ShimName, "file::memory:?cache=shared")
db := bun.NewDB(sqldb, sqlitedialect.New())
defer db.Close()
ctx := context.Background()
_ = outbox.Migrate(ctx, db)
// Business write + event, atomically.
tx, _ := db.BeginTx(ctx, nil)
_ = outbox.Enqueue(ctx, tx, "order.created", map[string]int{"order_id": 42})
_ = tx.Commit()
// Background relay publishes committed events at-least-once.
relay := outbox.NewRelay(db, recordingPublisher{})
n, _ := relay.Process(ctx)
fmt.Println("published:", n)
}
Output: publish order.created {"order_id":42} published: 1
type RelayOption ¶
type RelayOption func(*Relay)
RelayOption configures a Relay.
func WithBatchSize ¶
func WithBatchSize(n int) RelayOption
WithBatchSize sets the maximum number of events published per Process call. The default is 100.
func WithInterval ¶
func WithInterval(d time.Duration) RelayOption
WithInterval sets how often Run calls Process. The default is one second.
func WithLock ¶
WithLock makes Process run each batch under a distributed lock, so only one replica relays at a time. The lock is held for ttl and renewed before every event, so ttl only needs to exceed the time to publish a single event.
func WithLogger ¶
func WithLogger(l *logger.Logger) RelayOption
WithLogger sets the logger Run reports Process errors to. The default is logger.Default().
func WithMaxAttempts ¶
func WithMaxAttempts(n int) RelayOption
WithMaxAttempts moves an event to the outbox_dead table (see DeadEvent) once its publish has failed n times in a row, so one poison event cannot block the queue forever. The default, 0, retries the head event indefinitely.
Source Files