92 lines
2.7 KiB
Go
92 lines
2.7 KiB
Go
package events
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// EventBus is the platform event bus interface.
|
|
// Publish sends a typed payload to the given subject.
|
|
// Subscribe registers a handler for a subject pattern (supports NATS wildcards).
|
|
// The returned func() unsubscribes when called.
|
|
type EventBus interface {
|
|
Publish(subject string, payload any) error
|
|
Subscribe(subject string, handler func(subject string, data []byte)) (func(), error)
|
|
Healthy() bool
|
|
Close()
|
|
}
|
|
|
|
// NATSBus is the NATS-backed EventBus. Events are published to core NATS subjects.
|
|
// Phase 2A uses core NATS (ephemeral). Phase 2B will add JetStream for CI durability.
|
|
type NATSBus struct {
|
|
nc *nats.Conn
|
|
}
|
|
|
|
func NewNATSBus(url string) (*NATSBus, error) {
|
|
nc, err := nats.Connect(url,
|
|
nats.MaxReconnects(-1),
|
|
nats.ReconnectWait(2*time.Second),
|
|
nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
|
|
if err != nil {
|
|
log.Printf("nats: disconnected: %v", err)
|
|
}
|
|
}),
|
|
nats.ReconnectHandler(func(_ *nats.Conn) {
|
|
log.Printf("nats: reconnected")
|
|
}),
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("nats connect %s: %w", url, err)
|
|
}
|
|
log.Printf("nats: connected to %s", url)
|
|
return &NATSBus{nc: nc}, nil
|
|
}
|
|
|
|
func (b *NATSBus) Publish(subject string, payload any) error {
|
|
data, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return fmt.Errorf("marshal: %w", err)
|
|
}
|
|
return b.nc.Publish(subject, data)
|
|
}
|
|
|
|
func (b *NATSBus) Subscribe(subject string, handler func(subject string, data []byte)) (func(), error) {
|
|
sub, err := b.nc.Subscribe(subject, func(msg *nats.Msg) {
|
|
handler(msg.Subject, msg.Data)
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return func() { sub.Unsubscribe() }, nil //nolint:errcheck
|
|
}
|
|
|
|
func (b *NATSBus) Healthy() bool { return b.nc.IsConnected() }
|
|
|
|
func (b *NATSBus) Close() {
|
|
if err := b.nc.Drain(); err != nil {
|
|
log.Printf("nats: drain: %v", err)
|
|
}
|
|
}
|
|
|
|
// NoOpBus is a no-op EventBus used when NATS_URL is not configured.
|
|
// Events are silently dropped. The app works normally; just no real-time push.
|
|
type NoOpBus struct{}
|
|
|
|
func (NoOpBus) Publish(_ string, _ any) error { return nil }
|
|
func (NoOpBus) Subscribe(_ string, _ func(string, []byte)) (func(), error) { return func() {}, nil }
|
|
func (NoOpBus) Healthy() bool { return true }
|
|
func (NoOpBus) Close() {}
|
|
|
|
// New returns a NATSBus if url is non-empty, otherwise a NoOpBus.
|
|
func New(url string) (EventBus, error) {
|
|
if url == "" {
|
|
log.Printf("events: NATS_URL not set — using no-op bus (real-time push disabled)")
|
|
return NoOpBus{}, nil
|
|
}
|
|
return NewNATSBus(url)
|
|
}
|