notice

Documentation

Index

• Constants
• func Process(ctx context.Context, ch <-chan MessageCloser, handler Handler)
• func ProcessFunc(ctx context.Context, ch <-chan MessageCloser, h HandlerFunc)
• func Subscribe(ctx context.Context, nc *nats.Conn, subject string, bufferSize int) (<-chan MessageCloser, error)
• func WithConsistentRead(repo eventsourcex.Repository, nc *nats.Conn, subject string, ...) eventsourcex.Repository
• type DAO
• type Handler
• type HandlerFunc
  • func NewDBHandler(accessor dbase.Accessor, dao DAO, store eventsource.Store, ...) HandlerFunc
  • func (fn HandlerFunc) Process(ctx context.Context, notice Message)
• type Message
• type MessageCloser
• type UnmarshalFunc

Constants

const (
	// Group provides the name of the nats group
	Group = "vavende-identities-dbase"
)

Functions

func Process

func Process(ctx context.Context, ch <-chan MessageCloser, handler Handler)

Process reads from the channel and invokes the Handler. Process exits when either the context is canceled or the channel is closed.

Process guarantees that only one invocation of a Handler will operate upon a given aggregateID at a time

func ProcessFunc

func ProcessFunc(ctx context.Context, ch <-chan MessageCloser, h HandlerFunc)

ProcessFunc provides a convenience func wrapper around Process

func Subscribe

func Subscribe(ctx context.Context, nc *nats.Conn, subject string, bufferSize int) (<-chan MessageCloser, error)

Subscribe listens for notices on the nats subject provided

func WithConsistentRead

func WithConsistentRead(repo eventsourcex.Repository, nc *nats.Conn, subject string, timeout time.Duration) eventsourcex.Repository

WithConsistentRead provides a faux consistent read. Should wrap WithNotifier to ensure that the NoticesSubject.{ID} is subscribed to prior to the command being executed.

Types

type DAO

type DAO interface {
	// Version returns the current version number of the aggregate; returns 0, nil if the aggregate wasn't found
	Version(ctx context.Context, db *gorm.DB, aggregateID string) (int, error)

	// HandleEvent updates the read model with the specified event
	HandleEvent(ctx context.Context, db *gorm.DB, event eventsource.Event) error
}

DAO provides the shape of the required db access layer

type Handler

type Handler interface {
	// Process will be called as notices are received
	Process(ctx context.Context, notice Message)
}

Handler accepts a notice and performs the necessary operations to update the read model

type HandlerFunc

type HandlerFunc func(ctx context.Context, notice Message)

HandlerFunc provides a convenient func wrapper around Handler

func NewDBHandler

func NewDBHandler(accessor dbase.Accessor, dao DAO, store eventsource.Store, unmarshal UnmarshalFunc) HandlerFunc

NewDBHandler generates a new Handler for database read models

func (HandlerFunc) Process

func (fn HandlerFunc) Process(ctx context.Context, notice Message)

Process implements the Handler interface

type Message

type Message interface {
	AggregateID() string
}

Notice contains a request by the submitted of a command to update a specific read model

type MessageCloser

type MessageCloser interface {
	Message
	io.Closer
}

type UnmarshalFunc

type UnmarshalFunc func([]byte) (eventsource.Event, error)

UnmarshalFunc reads a []byte and returns an event