gap

README

gap

A lightweight, event-driven messaging library for Go. It provides outbox pattern implementation with support for RabbitMQ, Kafka and MySQL (or GORM-based storage), and is designed to support additional brokers and databases in the future.

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

go get github.com/loopopen/gap

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/storage/xgorm"
    "github.com/loopopen/gap/broker/xrabbitmq"
    "gorm.io/driver/mysql"
    "gorm.io/gorm"
    "gorm.io/gorm/logger"
)

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 publisher
    pub := gap.NewEventPublisher(
        gap.WithDrain(ctx, 5),
        xrabbitmq.UseRabbitMQ(
            xrabbitmq.URL(url),
        ),
        xgorm.UseGorm(
            xgorm.DB(db),
        ),
    )

    // Set up subscriber
    gap.Subscribe(
        gap.From(pub),
        gap.ServiceName("my-service.worker"),
        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...")
            db.Transaction(func(tx *gorm.DB) error {
                // Your business logic here...

                // Publish message (outbox pattern ensures reliability)
                return pub.Publish(ctx, event.NewOrderCreated())
            })
        }
    }()

    <-ctx.Done()
    stop()
}

//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 context cancellation and the built-in drain helper.

• gap.WaitDrain(ctx context.Context) gives the publisher/subscriber time to finish in-flight messages before exiting

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.NewPublisher[T](opts...) Publisher[T]: Create a typed publisher
• gap.NewEventPublisher(opts...) EventPublisher: Create an event publisher
• publisher.Bind(tx Txer) Publisher[T]: Bind publisher to a transaction for outbox pattern

Subscribers

• gap.Subscribe(opts...): Set up subscribers with handlers

Options

• gap.WithDrain(ctx context.Context, timeoutSeconds int): Enable drain mode for graceful shutdown
• gap.ServiceName(name string): Set service name
• gap.Inject(deps...): Inject dependencies into handlers

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(opts OptionsGetter) shoot.Option[Options, *Options]
• func HandleTopic[T any](handler gap.Handler[T], topic string) shoot.Option[Options, *Options]
• func HandleTopicWithinGroup[T any](group, topic string, handler gap.Handler[T]) shoot.Option[Options, *Options]
• func NewDashboardHandler(opts OptionsGetter) http.Handler
• func WithDrain(ctx context.Context, timeoutSeconds int) shoot.Option[Options, *Options]
• type Event
• type EventPublisher
  • func NewEventPublisher(opts ...shoot.Option[Options, *Options]) EventPublisher
• type Handler
• type Options
• type OptionsGetter
  • func Subscribe(opts ...shoot.Option[Options, *Options]) OptionsGetter
• type Publisher
  • func NewPublisher[T any](opts ...shoot.Option[Options, *Options]) Publisher[T]
• type Txer

Constants

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

Variables

var (
	Version               = gap.Version
	ServiceName           = gap.ServiceName
	DefaultGroup          = gap.DefaultGroup
	WorkerID              = gap.WorkerID
	ClaimBatchSize        = gap.ClaimBatchSize
	MaxRetries            = gap.MaxRetries
	LookbackSeconds       = gap.LookbackSeconds
	PumpIntervalSeconds   = gap.PumpIntervalSeconds
	MaxPublishConcurrency = gap.MaxPublishConcurrency
	PublishBufferSize     = gap.PublishBufferSize
	WorkConcurrencyFactor = gap.WorkConcurrencyFactor
	UseDashboard          = gap.UseDashboard
	Inject                = gap.Inject
)

var Pair = internal.Pair

var WaitDrain = pump.WaitDrain

Functions

func From

func From(opts OptionsGetter) shoot.Option[Options, *Options]

func HandleTopic

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

func HandleTopicWithinGroup

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

func NewDashboardHandler

func NewDashboardHandler(opts OptionsGetter) http.Handler

func WithDrain

func WithDrain(ctx context.Context, timeoutSeconds int) shoot.Option[Options, *Options]

Types

type Event

type Event = internal.Event

type EventPublisher

type EventPublisher = internal.EventPublisher

func NewEventPublisher

func NewEventPublisher(opts ...shoot.Option[Options, *Options]) EventPublisher

type Handler

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

type Options

type Options = gap.Options

type OptionsGetter

type OptionsGetter internal.OptionsGetter

func Subscribe

func Subscribe(opts ...shoot.Option[Options, *Options]) OptionsGetter

type Publisher

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

func NewPublisher

func NewPublisher[T any](opts ...shoot.Option[Options, *Options]) Publisher[T]

type Txer

type Txer = txer.Txer