nats

README

eh-nats

Eventbus with NATS/NATS Streaming backend for https://github.com/looplab/eventhorizon

Example init

Connection configuration

natsUrl := "nats://localhost:4222"
subject := "ehBus"
clusterId := "test-cluster"
clientId := "client-id"

Create connection

nc, err := nats.Connect(natsUrl)
conn, err := stan.Connect(clusterId, clientId, stan.NatsConn(nc))

or

conn, err := stan.Connect(clusterId, clientId, stan.NatsURL(natsUrl))

Create bus using connection

eventBus, err := nats.NewEventBus(conn, subject)

Documentation

Index

• func ErrCouldNotMarshalEvent(err error) error
• func ErrCouldNotPublishEvent(err error) error
• func ErrCouldNotUnmarshalEvent(err error) error
• type Error
  • func (e Error) Error() string
• type Event
• type EventBus
  • func NewEventBus(nc stan.Conn, subjectPrefix string) (*EventBus, error)
  • func (b *EventBus) AddHandler(m eh.EventMatcher, h eh.EventHandler)
  • func (b *EventBus) AddObserver(m eh.EventMatcher, h eh.EventHandler)
  • func (b *EventBus) Errors() <-chan Error
  • func (b *EventBus) PublishEvent(ctx context.Context, event eh.Event) error

Functions

func ErrCouldNotMarshalEvent

func ErrCouldNotMarshalEvent(err error) error

ErrCouldNotMarshalEvent is when an event could not be marshaled into BSON.

func ErrCouldNotPublishEvent

func ErrCouldNotPublishEvent(err error) error

ErrCouldNotPublishEvent is when kafka-client cannot send event to kafka

func ErrCouldNotUnmarshalEvent

func ErrCouldNotUnmarshalEvent(err error) error

ErrCouldNotUnmarshalEvent is when an event could not be unmarshaled into BSON.

Types

type Error

type Error struct {
	Err   error
	Ctx   context.Context
	Event eh.Event
}

Error is an async error containing the error and the event.

func (Error) Error

func (e Error) Error() string

Error implements the Error method of the error interface.

type Event

type Event struct {
	EventType     eh.EventType
	Timestamp     time.Time
	AggregateType eh.AggregateType
	AggregateID   eh.UUID
	Version       int
	Context       map[string]interface{}
	RawData       *bson.Raw `bson:",omitempty"`
}

type EventBus

type EventBus struct {
	// contains filtered or unexported fields
}

func NewEventBus

func NewEventBus(nc stan.Conn, subjectPrefix string) (*EventBus, error)

func (*EventBus) AddHandler

func (b *EventBus) AddHandler(m eh.EventMatcher, h eh.EventHandler)

AddHandler implements the AddHandler method of the eventhorizon.EventBus interface.

func (*EventBus) AddObserver

func (b *EventBus) AddObserver(m eh.EventMatcher, h eh.EventHandler)

AddObserver implements the AddObserver method of the eventhorizon.EventBus interface.

func (*EventBus) Errors

func (b *EventBus) Errors() <-chan Error

Errors returns an error channel where async handling errors are sent.

func (*EventBus) PublishEvent

func (b *EventBus) PublishEvent(ctx context.Context, event eh.Event) error