nats

Documentation

Index

• type Config
• type Conn
  • func NewConn(conn stan.Conn) *Conn
  • func (cn *Conn) Msg(subject string, seqID uint64) (*stan.Msg, error)
  • func (cn *Conn) MsgSeqRange(subject string, seqStart uint64, seqEnd uint64) ([]*stan.Msg, error)
  • func (cn *Conn) SubscribeChan(subject string, msgChan chan *stan.Msg, opts ...stan.SubscriptionOption) (stan.Subscription, error)
• type EventStore
  • func Connect(cfg Config) (*EventStore, error)
  • func NewEventStore(conn *WildcardConn, cfg Config) *EventStore
  • func (es *EventStore) Append(event *fes.Event) error
  • func (es *EventStore) Close() error
  • func (es *EventStore) Get(aggregate fes.Aggregate) ([]*fes.Event, error)
  • func (es *EventStore) List(matcher fes.AggregateMatcher) ([]fes.Aggregate, error)
  • func (es *EventStore) RunConnectionChecker()
  • func (es *EventStore) Watch(aggregate fes.Aggregate) error
• type WildcardConn
  • func NewWildcardConn(conn stan.Conn) *WildcardConn
  • func (wc *WildcardConn) List(matcher fes.AggregateMatcher) ([]string, error)
  • func (wc *WildcardConn) Publish(subject string, data []byte) error
  • func (wc *WildcardConn) Subscribe(wildcardSubject string, cb stan.MsgHandler, opts ...stan.SubscriptionOption) (stan.Subscription, error)
• type WildcardSub
  • func (ws *WildcardSub) Close() error
  • func (ws *WildcardSub) Unsubscribe() error

Types

type Config

type Config struct {
	Cluster       string
	Client        string
	URL           string // e.g. nats://localhost:9300
	AutoReconnect bool
}

type Conn

type Conn struct {
	stan.Conn
}

Conn is a wrapper of 'stan.Conn' struct to augment the API with bounded subscriptions and channel-based subscriptions

func NewConn

func NewConn(conn stan.Conn) *Conn

func (*Conn) Msg

func (cn *Conn) Msg(subject string, seqID uint64) (*stan.Msg, error)

Msg has a python style element selector (-1 = len(events)-1)

func (*Conn) MsgSeqRange

func (cn *Conn) MsgSeqRange(subject string, seqStart uint64, seqEnd uint64) ([]*stan.Msg, error)

func (*Conn) SubscribeChan

func (cn *Conn) SubscribeChan(subject string, msgChan chan *stan.Msg, opts ...stan.SubscriptionOption) (stan.Subscription, error)

type EventStore

type EventStore struct {
	pubsub.Publisher

	Config Config
	// contains filtered or unexported fields
}

EventStore is a NATS-based implementation of the EventStore interface.

func Connect

func Connect(cfg Config) (*EventStore, error)

Connect to a NATS cluster using the config.

func NewEventStore

func NewEventStore(conn *WildcardConn, cfg Config) *EventStore

func (*EventStore) Append

func (es *EventStore) Append(event *fes.Event) error

Append publishes (and persists) an event on the NATS message queue

func (*EventStore) Close

func (es *EventStore) Close() error

func (*EventStore) Get

func (es *EventStore) Get(aggregate fes.Aggregate) ([]*fes.Event, error)

Get returns all events related to a specific aggregate

func (*EventStore) List

func (es *EventStore) List(matcher fes.AggregateMatcher) ([]fes.Aggregate, error)

List returns all entities of which the subject matches the matcher. A nil matcher is considered a 'match-all'.

func (*EventStore) RunConnectionChecker

func (es *EventStore) RunConnectionChecker()

func (*EventStore) Watch

func (es *EventStore) Watch(aggregate fes.Aggregate) error

Watch a aggregate type for new events. The events are emitted over the publisher interface.

type WildcardConn

type WildcardConn struct {
	*Conn
}

WildcardConn is an abstraction on top of Conn that provides wildcard support

func NewWildcardConn

func NewWildcardConn(conn stan.Conn) *WildcardConn

func (*WildcardConn) List

func (wc *WildcardConn) List(matcher fes.AggregateMatcher) ([]string, error)

List retrieves all mentioned entities on the activity channel. The results can be filtered using the matcher, with a nil matcher equivalent to a 'match-all'.

func (*WildcardConn) Publish

func (wc *WildcardConn) Publish(subject string, data []byte) error

func (*WildcardConn) Subscribe

func (wc *WildcardConn) Subscribe(wildcardSubject string, cb stan.MsgHandler, opts ...stan.SubscriptionOption) (stan.Subscription, error)

type WildcardSub

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

WildcardSub is an abstraction on top of stan.Subscription that provides wildcard support

func (*WildcardSub) Close

func (ws *WildcardSub) Close() error

func (*WildcardSub) Unsubscribe

func (ws *WildcardSub) Unsubscribe() error