Bon (Bus of Nodes)

Bon is a lightweight, decoupled, and resilient Event Bus library and protocol designed for the Dataist ecosystem. Built with Clean Architecture (Ports & Adapters) principles, bon provides a robust abstraction layer for asynchronous service-to-service communication.

---

Features

• Clean Architecture: Strict separation between core business logic interfaces and infrastructure adapters.
• NATS Adapter: High-performance message transport powered by NATS with automatic reconnection handling.
• Offline Local Buffer: Automatically queues messages locally during network outages and flushes them once reconnected.
• Smart Retry Mechanism: Built-in exponential backoff strategy to handle transient network and broker failures gracefully.
• Built-in Validation: Validates event structures prior to publishing to maintain data consistency.

---

Installation

To add bon to your Go project, run:

go get [github.com/dataist_ir/bon](https://github.com/dataist_ir/bon)

Quick Start

Here is a simple example demonstrating how to initialize the NATS bus, subscribe to a subject, and publish an event:

package main
import (
"context""fmt""log""time"
natsadapter "[github.com/dataist_ir/bon/pkg/adapters/nats](https://github.com/dataist_ir/bon/pkg/adapters/nats)""[github.com/dataist_ir/bon/pkg/bus](https://github.com/dataist_ir/bon/pkg/bus)"
)
funcmain() {
ctx:=context.Background()
// Initialize the NATS bus adapterclient, err:=natsadapter.NewNatsBus("nats://localhost:4222")
iferr!=nil {
log.Fatalf("Failed to connect: %v", err)
}
deferclient.Close()
// Subscribe to a topic/subjecterr=client.Subscribe(ctx, "user.created", func(ctx context.Context, event bus.Event) error {
fmt.Printf("Received event: %s from %s\n", event.Type, event.Sender)
returnnil
})
iferr!=nil {
log.Fatalf("Failed to subscribe: %v", err)
}
// Construct an eventevent:= bus.Event{
Type: "user.created",
Version: "1.0.0",
Sender: "auth-service",
Timestamp: time.Now().Format(time.RFC3339),
Payload: map[string]interface{}{"user_id": "42"},
}
// Publish the eventerr=client.Publish(ctx, "user.created", event)
iferr!=nil {
log.Fatalf("Failed to publish: %v", err)
}
// Wait briefly to allow async delivery in the exampletime.Sleep(1*time.Second)
fmt.Println("Example finished successfully.")
}

Project Structure

.
├── CHANGELOG.md
├── go.mod
├── go.sum
├── LICENSE
├── pkg
│ ├── adapters
│ │ └── nats # NATSinfrastructureimplementation & offlinebuffer
│ ├── bus
│ │ ├── inmemory.go # In-memorybusfortesting
│ │ ├── interface.go # Coreinterfaces (Publisher, Subscriber)
│ │ ├── message.go # Eventmodeldefinition
│ │ └── validation.go # Payload & eventvalidators
│ └── examples
│ └── main.go # Usageexamples
├── README.md
└── VERSION