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
- type Message
- type MessageCloser
- type UnmarshalFunc
Constants ¶
const (
// Group provides the name of the nats group
Group = "vavende-identities-dbase"
)
Variables ¶
This section is empty.
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 ¶
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
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 UnmarshalFunc ¶
type UnmarshalFunc func([]byte) (eventsource.Event, error)
UnmarshalFunc reads a []byte and returns an event
Source Files