The highest tagged major version is
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 ¶
Types ¶
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. It returns the number of events published. When a Locker is configured and another replica holds the lock, Process returns (0, nil).
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/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.
Source Files