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)"
)

func main() {
	ctx := context.Background()
	
	// Initialize the NATS bus adapter
	client, err := natsadapter.NewNatsBus("nats://localhost:4222")
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer client.Close()

	// Subscribe to a topic/subject
	err = 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)
		return nil
	})
	if err != nil {
		log.Fatalf("Failed to subscribe: %v", err)
	}

	// Construct an event
	event := 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 event
	err = client.Publish(ctx, "user.created", event)
	if err != nil {
		log.Fatalf("Failed to publish: %v", err)
	}

	// Wait briefly to allow async delivery in the example
	time.Sleep(1 * time.Second)
	fmt.Println("Example finished successfully.")
}

Project Structure

.
├── CHANGELOG.md
├── go.mod
├── go.sum
├── LICENSE
├── pkg
│   ├── adapters
│   │   └── nats       # NATS infrastructure implementation & offline buffer
│   ├── bus
│   │   ├── inmemory.go    # In-memory bus for testing
│   │   ├── interface.go   # Core interfaces (Publisher, Subscriber)
│   │   ├── message.go     # Event model definition
│   │   └── validation.go  # Payload & event validators
│   └── examples
│       └── main.go        # Usage examples
├── README.md
└── VERSION