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, and storage/xmysql.
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 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...")
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 := gap.WaitDrain(drainCtx); err != nil {
slog.Warn("gap failed to drain", 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 context cancellation and the built-in drain helper.
gap.WaitDrain(ctx context.Context) errorgives 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:
- Start MySQL and RabbitMQ services
- Create a database named
example - Run the example:
cd examples/rabbitmq-gorm-mysql-example
go run .
API Reference
Publishers
gap.NewPublisher[T](opts...) Publisher[T]: Create a typed publishergap.NewEventPublisher(opts...) EventPublisher: Create an event publisherpublisher.Bind(tx Txer) (Publisher[T], error): Bind a publisher to a transaction for the outbox pattern
Subscribers
gap.Subscribe(opts...): Set up subscribers with handlers
Options
gap.WithDrain(ctx context.Context, timeoutSeconds int): Enable drain mode for graceful shutdowngap.ServiceName(name string): Set service namegap.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
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
- type Handler
- type Options
- type OptionsGetter
- type Publisher
- 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 HandleTopic ¶
func HandleTopicWithinGroup ¶
func NewDashboardHandler ¶
func NewDashboardHandler(opts OptionsGetter) http.Handler
Types ¶
type EventPublisher ¶
type EventPublisher = internal.EventPublisher
func NewEventPublisher ¶
func NewEventPublisher(opts ...shoot.Option[Options, *Options]) EventPublisher
type OptionsGetter ¶
type OptionsGetter internal.OptionsGetter
Source Files
Directories