postgres

Documentation

Index

• Variables
• func QuoteIdentifier(name string) string
• func QuoteString(str string) string
• type AggregateProjector
  • func NewAggregateProjector(db *sql.DB, eventStore driverSQL.ReadOnlyEventStore, eventStoreTable string, ...) (*AggregateProjector, error)
  • func (a *AggregateProjector) Run(ctx context.Context) error
  • func (a *AggregateProjector) RunAndListen(ctx context.Context, listener driverSQL.Listener) error
• type ConjoinedEventStore
  • func NewConjoinedEventStore(eventstore *EventStore, resolver goengine.MessagePayloadResolver, ...) (*ConjoinedEventStore, error)
  • func (e *ConjoinedEventStore) AppendTo(ctx context.Context, streamName goengine.StreamName, ...) error
• type ConjoinedMessageHandler
• type EventStore
  • func NewEventStore(persistenceStrategy driverSQL.PersistenceStrategy, db *sql.DB, ...) (*EventStore, error)
  • func (e *EventStore) AppendTo(ctx context.Context, streamName goengine.StreamName, ...) error
  • func (e *EventStore) AppendToWithExecer(ctx context.Context, conn driverSQL.Execer, streamName goengine.StreamName, ...) error
  • func (e *EventStore) Create(ctx context.Context, streamName goengine.StreamName) error
  • func (e *EventStore) HasStream(ctx context.Context, streamName goengine.StreamName) bool
  • func (e *EventStore) Load(ctx context.Context, streamName goengine.StreamName, fromNumber int64, ...) (goengine.EventStream, error)
  • func (e *EventStore) LoadWithConnection(ctx context.Context, conn driverSQL.Queryer, streamName goengine.StreamName, ...) (goengine.EventStream, error)
• type StreamProjector
  • func NewStreamProjector(db *sql.DB, eventStore driverSQL.ReadOnlyEventStore, ...) (*StreamProjector, error)
  • func (s *StreamProjector) Run(ctx context.Context) error
  • func (s *StreamProjector) RunAndListen(ctx context.Context, listener driverSQL.Listener) error

Variables

var (
	// ErrNoCreateTableQueries occurs when table create queries are not presented in the strategy
	ErrNoCreateTableQueries = errors.New("goengine: create table queries are not provided")
	// ErrTableAlreadyExists occurs when table cannot be created as it exists already
	ErrTableAlreadyExists = errors.New("goengine: table already exists")
	// ErrTableNameEmpty occurs when table cannot be created because it has an empty name
	ErrTableNameEmpty = errors.New("goengine: table name could not be empty")
)

Functions

func QuoteIdentifier

func QuoteIdentifier(name string) string

QuoteIdentifier quotes an "identifier" (e.g. a table or a column name) to be used as part of an SQL statement.

func QuoteString

func QuoteString(str string) string

QuoteString returns the given string quoted

Types

type AggregateProjector

type AggregateProjector struct {
	sync.Mutex
	// contains filtered or unexported fields
}

AggregateProjector is a postgres projector used to execute a projection per aggregate instance against an event stream

func NewAggregateProjector

func NewAggregateProjector(
	db *sql.DB,
	eventStore driverSQL.ReadOnlyEventStore,
	eventStoreTable string,
	resolver goengine.MessagePayloadResolver,
	aggregateTypeName string,
	projection goengine.Projection,
	projectionTable string,
	projectionErrorHandler driverSQL.ProjectionErrorCallback,
	logger goengine.Logger,
) (*AggregateProjector, error)

NewAggregateProjector creates a new projector for a projection

func (*AggregateProjector) Run

func (a *AggregateProjector) Run(ctx context.Context) error

Run executes the projection and manages the state of the projection

func (*AggregateProjector) RunAndListen

func (a *AggregateProjector) RunAndListen(ctx context.Context, listener driverSQL.Listener) error

RunAndListen executes the projection and listens to any changes to the event store

type ConjoinedEventStore

type ConjoinedEventStore struct {
	*EventStore
	// contains filtered or unexported fields
}

ConjoinedEventStore a in postgres event store implementation which includes projection logic.

func NewConjoinedEventStore

func NewConjoinedEventStore(
	eventstore *EventStore,
	resolver goengine.MessagePayloadResolver,
	handlers map[string]ConjoinedMessageHandler,
) (*ConjoinedEventStore, error)

NewConjoinedEventStore return a new postgres.ConjoinedEventStore

func (*ConjoinedEventStore) AppendTo

func (e *ConjoinedEventStore) AppendTo(ctx context.Context, streamName goengine.StreamName, streamEvents []goengine.Message) error

AppendTo batch inserts Messages into the event stream table

type ConjoinedMessageHandler

type ConjoinedMessageHandler func(ctx context.Context, tx *sql.Tx, message goengine.Message) error

ConjoinedMessageHandler is a message handler called by the ConjoinedEventStore when a message is appended

type EventStore

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

EventStore a in postgres event store implementation

func NewEventStore

func NewEventStore(
	persistenceStrategy driverSQL.PersistenceStrategy,
	db *sql.DB,
	messageFactory driverSQL.MessageFactory,
	logger goengine.Logger,
) (*EventStore, error)

NewEventStore return a new postgres.EventStore

func (*EventStore) AppendTo

func (e *EventStore) AppendTo(ctx context.Context, streamName goengine.StreamName, streamEvents []goengine.Message) error

AppendTo batch inserts Messages into the event stream table

func (*EventStore) AppendToWithExecer

func (e *EventStore) AppendToWithExecer(ctx context.Context, conn driverSQL.Execer, streamName goengine.StreamName, streamEvents []goengine.Message) error

AppendToWithExecer batch inserts Messages into the event stream table using the provided Connection/Execer

func (*EventStore) Create

func (e *EventStore) Create(ctx context.Context, streamName goengine.StreamName) error

Create creates the database table, index etc needed for the event stream

func (*EventStore) HasStream

func (e *EventStore) HasStream(ctx context.Context, streamName goengine.StreamName) bool

HasStream returns true if the table for the eventstream already exists

func (*EventStore) Load

func (e *EventStore) Load(
	ctx context.Context,
	streamName goengine.StreamName,
	fromNumber int64,
	count *uint,
	matcher metadata.Matcher,
) (goengine.EventStream, error)

Load returns an eventstream based on the provided constraints

func (*EventStore) LoadWithConnection

func (e *EventStore) LoadWithConnection(
	ctx context.Context,
	conn driverSQL.Queryer,
	streamName goengine.StreamName,
	fromNumber int64,
	count *uint,
	matcher metadata.Matcher,
) (goengine.EventStream, error)

LoadWithConnection returns an eventstream based on the provided constraints using the provided sql.Conn

type StreamProjector

type StreamProjector struct {
	sync.Mutex
	// contains filtered or unexported fields
}

StreamProjector is a postgres projector used to execute a projection against an event stream.

func NewStreamProjector

func NewStreamProjector(
	db *sql.DB,
	eventStore driverSQL.ReadOnlyEventStore,
	resolver goengine.MessagePayloadResolver,
	projection goengine.Projection,
	projectionTable string,
	projectionErrorHandler driverSQL.ProjectionErrorCallback,
	logger goengine.Logger,
) (*StreamProjector, error)

NewStreamProjector creates a new projector for a projection

func (*StreamProjector) Run

func (s *StreamProjector) Run(ctx context.Context) error

Run executes the projection and manages the state of the projection

func (*StreamProjector) RunAndListen

func (s *StreamProjector) RunAndListen(ctx context.Context, listener driverSQL.Listener) error

RunAndListen executes the projection and listens to any changes to the event store