bus

README

archon-bus

Driver-agnostic messaging interfaces for agent platforms.

NATS is one driver — not the product vocabulary. Executors and domain code depend on bus.Bus / bus.Open, not on a broker SDK.

Install

go get github.com/diogoX451/archon-bus@v0.1.0
// driver (side-effect register):
go get github.com/diogoX451/archon-nats-bus@latest

Quick start

import (
    "github.com/diogoX451/archon-bus"
    _ "github.com/diogoX451/archon-nats-bus" // registers driver "nats"
)

b, err := bus.Open(bus.Config{
    Driver: "nats",                         // or ARCHON_BUS_DRIVER
    URL:    "nats://127.0.0.1:4222",         // or ARCHON_BUS_URL
    MaxReconnects: -1,
})
defer b.Close()

_, err = b.Subscribe("archon.need.http", func(ctx context.Context, msg bus.Message) error {
    // work...
    return msg.Ack()
})

Config surface (generic)

Field / env	Meaning
Driver / ARCHON_BUS_DRIVER	nats (default), future: memory, kafka, …
URL / ARCHON_BUS_URL	Driver-specific URL
MaxReconnects	Reconnecting drivers; -1 = forever

Legacy ARCHON_NATS_* remains valid in the Archon product config dual-read — OSS examples prefer ARCHON_BUS_*.

Writing a new driver

func init() {
    bus.Register("memory", func(cfg bus.Config) (bus.Bus, error) {
        return newMemoryBus(cfg), nil
    })
}

Implement bus.Bus (and optionally bus.Drainable). Keep broker types out of the public executor surface.

Related

• archon-nats-bus — JetStream driver
• archon-http-executor — uses bus.Open
• archon-oss

License

Apache-2.0

Documentation

Overview

Package bus defines a driver-agnostic messaging surface for agent platforms.

NATS/JetStream is one driver (see archon-nats-bus). Executors and product code should depend on these interfaces and open a bus via Open(Config), never import a broker SDK directly in business logic.

Part of the Archon open-source toolkit: https://github.com/diogoX451/archon-oss

Index

• func Drivers() []string
• func Register(name string, open Opener)
• type Bus
  • func MustOpen(cfg Config) Bus
  • func Open(cfg Config) (Bus, error)
• type Config
• type Drainable
• type Handler
• type Message
• type MsgMetadata
• type Opener
• type Subscription

Functions

func Drivers

func Drivers() []string

Drivers returns registered driver names (sorted not guaranteed).

func Register

func Register(name string, open Opener)

Register adds or replaces a bus driver opener. Name is case-insensitive. Call from driver packages' init() (e.g. archon-nats-bus).

Types

type Bus

type Bus interface {
	Publish(ctx context.Context, subject string, payload []byte) error
	Subscribe(subject string, handler Handler) (Subscription, error)
	Close() error
}

Bus is the minimal publish/subscribe surface every driver must implement.

func MustOpen

func MustOpen(cfg Config) Bus

MustOpen is Open that panics on error (examples / main only).

func Open

func Open(cfg Config) (Bus, error)

Open constructs a bus for cfg.Driver (default "nats").

type Config

type Config struct {
	// Driver selects the implementation: "nats", "jetstream", "memory", …
	// Empty defaults to "nats".
	Driver string
	// URL is driver-specific (nats://host:4222, memory, kafka://…, …).
	URL string
	// MaxReconnects / ReconnectWait apply to reconnecting drivers.
	// MaxReconnects -1 = reconnect forever (recommended for production).
	MaxReconnects int
	ReconnectWait time.Duration
	// SubscribeAckWait / SubscribeMaxDeliver apply to push consumers.
	SubscribeAckWait    time.Duration
	SubscribeMaxDeliver int
	// Extra holds driver-specific options without polluting the core API.
	Extra map[string]string
}

Config opens a bus for a named driver. Fields beyond Driver/URL are driver-specific hints (reconnect, ack wait, …).

type Drainable

type Drainable interface {
	Bus
	Drain(ctx context.Context) error
}

Drainable optionally supports graceful drain before Close.

type Handler

type Handler func(ctx context.Context, msg Message) error

Handler processes one message. Return nil after Ack (or when the driver auto-acks). Return a non-nil error for redelivery on supporting drivers.

type Message

type Message interface {
	Data() []byte
	Subject() string
	Ack() error
	Nak(delay ...time.Duration) error
	InProgress() error
	Metadata() (*MsgMetadata, error)
}

Message is a transport-agnostic bus message with manual ack support.

type MsgMetadata

type MsgMetadata struct {
	Sequence   uint64
	Time       time.Time
	Stream     string
	Consumer   string
	Deliveries int
}

MsgMetadata carries delivery info when the driver exposes it.

type Opener

type Opener func(cfg Config) (Bus, error)

Opener constructs a Bus from Config. Drivers register via Register.

type Subscription

type Subscription interface {
	Unsubscribe() error
}

Subscription can be cancelled.