gap

README

gap

GAP is a lightweight, event-driven messaging library for Go. It uses the outbox pattern, so a message is written in the same database transaction as your data, then delivered once that transaction commits. It speaks RabbitMQ and Kafka today, and it can store messages in MySQL or in any database GORM already supports. More brokers and databases are part of the plan. The library is shaped so those can be added without tearing up what you already have.

Operational Dashboard

Monitor message delivery at a glance with the built-in operational dashboard. See publishing and consumption health, success rates, in-flight work, failures, backlog age, and active runtime components from one place.

Features

• Outbox Pattern: Reliable message publishing with database transactions
• Multiple Brokers: Support for RabbitMQ, Kafka (and extensible for others)
• Storage Backends: GORM integration for MySQL, PostgreSQL, etc.
• Code Generation: Automatic handler generation with gapc
• Type Safety: Generic-based API for type-safe message handling
• Dependency Injection: Built-in DI container for handlers

Installation

GAP's broker and storage integrations are separate Go modules. Install the core module together with the integrations your application uses. For the RabbitMQ and GORM setup used below:

go get github.com/loopopen/gap
go get github.com/loopopen/gap/broker/xrabbitmq
go get github.com/loopopen/gap/storage/xgorm

Other available integrations include broker/xkafka, broker/xconfluent, storage/xmysql, and storage/xpostgres.

The bundled broker and storage modules share internal contracts with the core module. Application code should import only the core module and the integrations it uses.

Quick Start

Here's a complete example using RabbitMQ and MySQL with GORM:

1. Define Events

package event

import "github.com/google/uuid"

type OrderCreated struct {
    SN string
}

func NewOrderCreated() *OrderCreated {
    return &OrderCreated{
        SN: uuid.New().String(),
    }
}

func (e OrderCreated) Topic() string {
    return "order.created"
}

2. Set up Publisher and Subscriber

package main

import (
    "context"
    "fmt"
    "log/slog"
    "os/signal"
    "syscall"
    "time"

    "github.com/loopopen/gap"
    "github.com/loopopen/gap/broker/xrabbitmq"
    "github.com/loopopen/gap/storage/xgorm"
    "gorm.io/driver/mysql"
    "gorm.io/gorm"
)

func main() {
    dsn := "root:root@tcp(127.0.0.1:3306)/example?charset=utf8mb4&parseTime=True&loc=Local"
    url := "amqp://guest:guest@localhost:5672/example"

    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()

    db, _ := gorm.Open(mysql.Open(dsn))

    // Create the application-scoped Gap instance
    g := gap.New(
        gap.ServiceName("my-service.worker"),
        xgorm.UseGorm(
            xgorm.DB(db),
        ),
    )

    // Create publisher
    pub := g.NewEventPublisher(
        xrabbitmq.UseRabbitMQ(
            xrabbitmq.URL(url),
        ),
    )

    // Set up subscriber
    g.Subscribe(
        gap.From(pub),
        gap.Inject(/*handler_dependencies*/),
    )

    // Publish messages in a transaction
    go func() {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()

        for range ticker.C {
            slog.Info("Publishing message...")
            err := xgorm.DoInTx(ctx, func(ctx context.Context, txer gap.Txer) error {
                // Your business logic here...

                // Bind the publisher so the message and business data are
                // committed in the same transaction.
                txPub, err := pub.Bind(txer)
                if err != nil {
                    return err
                }
                return txPub.Publish(ctx, event.NewOrderCreated())
            }, db)
            if err != nil {
                slog.Error("failed to create order", slog.Any("err", err))
            }
        }
    }()

    <-ctx.Done()

    drainCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    if err := g.Shutdown(drainCtx); err != nil {
        slog.Warn("gap shutdown failed", slog.Any("err", err))
    }
}

//go:generate go run github.com/loopopen/gap/cmd/gapc -file=$GOFILE

// @subscribe
func handle(/*dependency-list*/) gap.Handler[*event.OrderCreated] {
    return func(ctx context.Context, msg *event.OrderCreated, headers map[string]string) error {
        slog.Info(fmt.Sprintf("received event: %s, %s", msg.Topic(), msg.SN))
        return nil
    }
}

3. Generate Handlers

Run the code generation:

go generate ./...

This will generate the necessary handler code automatically.

4. Graceful Shutdown

The library supports graceful shutdown through an explicit, runtime-owned shutdown sequence.

• g.Shutdown(ctx context.Context) error stops intake, drains accepted work, and closes runtime-owned resources
• Canceling the application's signal context does not shut down Gap; call g.Shutdown explicitly with a dedicated drain context

Delivery Semantics

GAP uses at-least-once message processing and does not guarantee exactly-once execution. The same logical message may reach a handler more than once, for example after broker redelivery, lease expiration, or a process failure after a business side effect succeeds but before the message status is persisted.

Handlers must therefore be idempotent. Use a stable business key such as an order ID or event ID, and enforce idempotency in the same transaction as the handler's business changes—for example with a unique constraint or a processed-message record. Do not rely on message payload equality for deduplication, because publishing the same payload again may be intentional.

Usage Examples

For a complete working example, see ./examples/rabbitmq-gorm-mysql-example.

To run the example:

1. Start MySQL and RabbitMQ services
2. Create a database named example
3. Run the example:

cd examples/rabbitmq-gorm-mysql-example
go run .

API Reference

Publishers

• gap.New(opts...) *Gap: Create an application-scoped Gap instance
• gap.NewPublisher[T](g, opts...) Publisher[T] (Go 1.24+): Create a typed publisher
• g.NewPublisher[T](opts...) Publisher[T] (Go 1.27+): Create a typed publisher from a Gap instance
• g.NewEventPublisher(opts...) EventPublisher: Create an event publisher
• publisher.Bind(tx Txer) (Publisher[T], error): Bind a publisher to a transaction for the outbox pattern

Subscribers

• g.Subscribe(opts...): Set up subscribers with broker, handler, and dependency options
• g.DashboardHandler(): Create the dashboard HTTP handler
• g.Shutdown(ctx): Stop intake, drain accepted work, and close runtime-owned resources

Options

• Runtime Options are passed to gap.New: gap.UseDashboard and storage options such as xmysql.UseMySQL
• BrokerOptions are passed to publisher constructors: gap.BrokerName and broker options such as xrabbitmq.UseRabbitMQ
• Subscriber options are passed to g.Subscribe: gap.ServiceName and gap.Inject

Code Generation

Use the gapc tool to generate handler boilerplate:

go run github.com/loopopen/gap/cmd/gapc -file=main.go

Add annotations to your handler functions: (topic;group)

// @subscribe: topic="my.topic"
func MyHandler(/*dependency-list*/) gap.Handler[MyMessage] {
    // handler implementation
}

If MyMessage implements the gap.Event interface, the topic can be omitted:

// @subscribe
func MyHandler(/*dependency-list*/) gap.Handler[MyEvent] {
    // handler implementation
}

Or use receiver as a dependency:

// @subscribe: topic="my.topic"
func (svc *MySvc) MyHandler(/*other-dependencies*/) gap.Handler[MyMessage] {
    // handler implementation
}

Contributing

Contributions are welcome! Please feel free to submit issues and pull requests.

License

See LICENSE file for details.

Inspiring projects

• CAP

Documentation

Index

• Constants
• Variables
• func From(source BrokerOptionsGetter) shoot.Option[BrokerOptions, *BrokerOptions]
• func HandleTopic[T any](handler ingap.Handler[T], topic string) shoot.Option[BrokerOptions, *BrokerOptions]
• func HandleTopicWithinGroup[T any](group, topic string, handler ingap.Handler[T]) shoot.Option[BrokerOptions, *BrokerOptions]
• func Register(opts ...shoot.Option[BrokerOptions, *BrokerOptions])
• type BrokerOptions
• type BrokerOptionsGetter
• type Event
• type EventPublisher
  • func NewEventPublisher(g *Gap, opts ...shoot.Option[BrokerOptions, *BrokerOptions]) EventPublisher
• type Gap
  • func New(opts ...shoot.Option[Options, *Options]) *Gap
  • func (g *Gap) DashboardHandler() http.Handler
  • func (g *Gap) NewEventPublisher(opts ...shoot.Option[BrokerOptions, *BrokerOptions]) EventPublisher
  • func (g *Gap) NewPublisher[T any](opts ...shoot.Option[BrokerOptions, *BrokerOptions]) Publisher[T]
  • func (g *Gap) Shutdown(ctx context.Context) error
  • func (g *Gap) Subscribe(opts ...shoot.Option[BrokerOptions, *BrokerOptions])
• type Handler
• type Options
• type OptionsGetter
• type Publisher
  • func NewPublisher[T any](g *Gap, opts ...shoot.Option[BrokerOptions, *BrokerOptions]) Publisher[T]
• type Txer

Constants

const (
	KeysMessageID     = internal.KeysMessageID
	KeysMessageType   = internal.KeysMessageType
	KeysGroup         = internal.KeysGroup
	KeysCorrelationID = internal.KeysCorrelationID
	KeysBrokerName    = internal.KeysBrokerName
	KeysBrokerType    = internal.KeysBrokerType
)

Variables

var (
	Version               = ingap.Version
	ServiceName           = ingap.ServiceName
	DefaultGroup          = ingap.DefaultGroup
	WorkerID              = ingap.WorkerID
	ClaimBatchSize        = ingap.ClaimBatchSize
	MaxRetries            = ingap.MaxRetries
	LeaseTTLSeconds       = ingap.LeaseTTLSeconds
	RetryIntervalSeconds  = ingap.RetryIntervalSeconds
	PumpIntervalSeconds   = ingap.PumpIntervalSeconds
	MaxPublishConcurrency = ingap.MaxPublishConcurrency
	PublishBufferSize     = ingap.PublishBufferSize
	WorkConcurrencyFactor = ingap.WorkConcurrencyFactor
	UseDashboard          = ingap.UseDashboard
	Inject                = ingap.Inject
	BrokerName            = ingap.BrokerName
	ErrRuntimeStopping    = ingap.ErrRuntimeStopping
)

var Pair = internal.Pair

Functions

func From

func From(source BrokerOptionsGetter) shoot.Option[BrokerOptions, *BrokerOptions]

From copies a publisher's broker configuration into a broker option. It lets a subscriber reuse the configuration owned by that publisher.

func HandleTopic

func HandleTopic[T any](handler ingap.Handler[T], topic string) shoot.Option[BrokerOptions, *BrokerOptions]

func HandleTopicWithinGroup

func HandleTopicWithinGroup[T any](group, topic string, handler ingap.Handler[T]) shoot.Option[BrokerOptions, *BrokerOptions]

func Register

func Register(opts ...shoot.Option[BrokerOptions, *BrokerOptions])

Register stores generated handler and dependency definitions during package initialization. Applications start them with Gap.Subscribe.

Types

type BrokerOptions

type BrokerOptions = ingap.BrokerOptions

type BrokerOptionsGetter

type BrokerOptionsGetter internal.BrokerOptionsGetter

type Event

type Event = internal.Event

type EventPublisher

type EventPublisher = internal.EventPublisher

func NewEventPublisher

func NewEventPublisher(g *Gap, opts ...shoot.Option[BrokerOptions, *BrokerOptions]) EventPublisher

type Gap

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

Gap owns the shared storage, pump, dashboard, subscriptions, and lifecycle of an application. Publishers and subscribers supply their broker-specific configuration independently.

func New

func New(opts ...shoot.Option[Options, *Options]) *Gap

New creates an application-scoped Gap instance.

func (*Gap) DashboardHandler

func (g *Gap) DashboardHandler() http.Handler

func (*Gap) NewEventPublisher

func (g *Gap) NewEventPublisher(opts ...shoot.Option[BrokerOptions, *BrokerOptions]) EventPublisher

func (*Gap) NewPublisher

func (g *Gap) NewPublisher[T any](opts ...shoot.Option[BrokerOptions, *BrokerOptions]) Publisher[T]

func (*Gap) Shutdown

func (g *Gap) Shutdown(ctx context.Context) error

Shutdown stops intake, drains accepted work, cancels work that exceeds ctx, and closes runtime-owned resources. It is safe to call more than once.

func (*Gap) Subscribe

func (g *Gap) Subscribe(opts ...shoot.Option[BrokerOptions, *BrokerOptions])

Subscribe configures the broker and registers all configured handlers.

type Handler

type Handler[T any] = ingap.Handler[T]

type Options

type Options = ingap.Options

type OptionsGetter

type OptionsGetter internal.OptionsGetter

type Publisher

type Publisher[T any] = internal.Publisher[T]

func NewPublisher

func NewPublisher[T any](g *Gap, opts ...shoot.Option[BrokerOptions, *BrokerOptions]) Publisher[T]

type Txer

type Txer = txer.Txer