postgres

Documentation

Overview

Package postgres contains an Event Store implementation using PostgreSQL as backend data store.

Index

• func RunMigrations(dsn string) error
• type AppendToStoreFunc
• type Checkpointer
  • func (c Checkpointer) Read(ctx context.Context, subscriptionName string) (int64, error)
  • func (c Checkpointer) Write(ctx context.Context, subscriptionName string, sequenceNumber int64) error
• type Deserializer
• type DeserializerFunc
  • func (fn DeserializerFunc) Deserialize(eventType string, eventStreamID stream.ID, data []byte) (eventually.Payload, error)
• type EventStore
  • func NewEventStore(db *sql.DB, serde Serde, options ...Option) EventStore
  • func (st EventStore) Append(ctx context.Context, id stream.ID, expected eventstore.VersionCheck, ...) (v int64, err error)
  • func (st EventStore) LatestSequenceNumber(ctx context.Context) (int64, error)
  • func (st EventStore) Stream(ctx context.Context, es eventstore.EventStream, target stream.Target, ...) error
• type FusedSerde
• type JSONRegistry
  • func NewJSONRegistry() JSONRegistry
  • func (r JSONRegistry) Deserialize(eventType string, eventStreamID stream.ID, data []byte) (eventually.Payload, error)
  • func (r JSONRegistry) Register(events ...eventually.Payload) error
  • func (r JSONRegistry) Serialize(eventType string, eventStreamID stream.ID, event eventually.Payload) ([]byte, error)
• type Option
  • func WithAppendMiddleware(wrap func(AppendToStoreFunc) AppendToStoreFunc) Option
• type Serde
• type Serializer
• type SerializerFunc
  • func (fn SerializerFunc) Serialize(eventType string, eventStreamID stream.ID, event eventually.Payload) ([]byte, error)

Functions

func RunMigrations

func RunMigrations(dsn string) error

RunMigrations performs the migrations for the Postgres database.

Types

type AppendToStoreFunc

type AppendToStoreFunc func(
	ctx context.Context,
	tx *sql.Tx,
	id stream.ID,
	expected eventstore.VersionCheck,
	eventName string,
	payload []byte,
	metadata []byte,
) (int64, error)

AppendToStoreFunc is the function type used by the postgres.EventStore to delegate the append call to the database instace.

type Checkpointer

type Checkpointer struct {
	DB     *sql.DB
	Logger logger.Logger
}

Checkpointer is a checkpoint.Checkpointer implementation using Postgres as a storage backend.

func (Checkpointer) Read

func (c Checkpointer) Read(ctx context.Context, subscriptionName string) (int64, error)

Read reads the latest checkpointed sequence number of the subscription specified.

func (Checkpointer) Write

func (c Checkpointer) Write(ctx context.Context, subscriptionName string, sequenceNumber int64) error

Write checkpoints the sequence number value provided for the specified subscription.

type Deserializer

type Deserializer interface {
	Deserialize(eventType string, eventStreamID stream.ID, data []byte) (eventually.Payload, error)
}

Deserializer can be used by the EventStore to delegate deserialization of a Domain Event from binary format to its corresponding Domain Object.

type DeserializerFunc

type DeserializerFunc func(eventType string, eventStreamID stream.ID, data []byte) (eventually.Payload, error)

DeserializerFunc is a functional type that implements the Deserializer interface.

func (DeserializerFunc) Deserialize

func (fn DeserializerFunc) Deserialize(eventType string, eventStreamID stream.ID, data []byte) (eventually.Payload, error)

Deserialize delegates the function call to the inner function value.

type EventStore

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

EventStore is an eventstore.Store implementation which uses PostgreSQL as backend datastore.

func NewEventStore

func NewEventStore(db *sql.DB, serde Serde, options ...Option) EventStore

NewEventStore creates a new EventStore using the database connection pool provided.

func (EventStore) Append

func (st EventStore) Append(
	ctx context.Context,
	id stream.ID,
	expected eventstore.VersionCheck,
	events ...eventually.Event,
) (v int64, err error)

Append inserts the specified Domain Events into the Event Stream specified by the current instance, returning the new version of the Event Stream.

A version can be specified to enable an Optimistic Concurrency check on append, by using the expected version of the Event Stream prior to appending the new Events.

Alternatively, VersionCheckAny can be used if no Optimistic Concurrency check should be carried out.

func (EventStore) LatestSequenceNumber

func (st EventStore) LatestSequenceNumber(ctx context.Context) (int64, error)

LatestSequenceNumber returns the latest Sequence Number used by a Domain Event committed to the Event Store.

func (EventStore) Stream

func (st EventStore) Stream(
	ctx context.Context,
	es eventstore.EventStream,
	target stream.Target,
	selectt eventstore.Select,
) error

Stream opens one or more Event Streams depending on the provided target.

In case of multi-Event Streams targets, the Select value specified will be applied over the value of the Global Sequence Number of the events. In case of a single Event Stream, this is applied over the Version value.

type FusedSerde

type FusedSerde struct {
	Serializer
	Deserializer
}

FusedSerde is a convenience type to fuse a Serializer and Deserializer in a Serde instance.

type JSONRegistry

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

JSONRegistry is a Serde type that serializes and deserializes into and from the JSON representation of eventually.Payload types registered.

Given the current limitation of Go with generics, the only way to provide type information for deserialization is to use interfaces and reflection.

func NewJSONRegistry

func NewJSONRegistry() JSONRegistry

NewJSONRegistry creates a new registry for deserializing event types, using the provided deserializer.

func (JSONRegistry) Deserialize

func (r JSONRegistry) Deserialize(eventType string, eventStreamID stream.ID, data []byte) (eventually.Payload, error)

Deserialize attempts to deserialize a raw message with the type referenced by the supplied event type identifier.

func (JSONRegistry) Register

func (r JSONRegistry) Register(events ...eventually.Payload) error

Register adds the type information to this registry for all the provided Payload types.

An error is returned if any of the provided events is nil, or if two different event types with the same type identifier (from the Payload.Name() method) have been provided.

func (JSONRegistry) Serialize

func (r JSONRegistry) Serialize(eventType string, eventStreamID stream.ID, event eventually.Payload) ([]byte, error)

Serialize serializes a Domain Event using its JSON representation.

type Option

type Option func(EventStore) EventStore

Option defines a type for providing additional constructor adjustments for postgres.EventStore.

func WithAppendMiddleware

func WithAppendMiddleware(wrap func(AppendToStoreFunc) AppendToStoreFunc) Option

WithAppendMiddleware allows overriding the internal logic for appending events within a transaction.

type Serde

type Serde interface {
	Serializer
	Deserializer
}

Serde is a serializer/deserializer type that can be used by the EventStore to serialize Domain Events to and deserialize Domain Events from the store.

type Serializer

type Serializer interface {
	Serialize(eventType string, eventStreamID stream.ID, event eventually.Payload) ([]byte, error)
}

Serializer can be used by the EventStore to delegate serialization of a Domain Event from the eventually.Payload format (domain object) to binary format.

type SerializerFunc

type SerializerFunc func(eventType string, eventStreamID stream.ID, event eventually.Payload) ([]byte, error)

SerializerFunc is a functional type that implements the Serializer interface.

func (SerializerFunc) Serialize

func (fn SerializerFunc) Serialize(eventType string, eventStreamID stream.ID, event eventually.Payload) ([]byte, error)

Serialize delegates the function call to the inner function value.