edatpgx

Documentation

Index

• Constants
• Variables
• func NewSnapshotStore(client Client, options ...SnapshotStoreOption) es.AggregateRootStoreMiddleware
• func ReceiverSessionMiddleware(conn *pgxpool.Pool, logger edatlog.Logger) func(msg.MessageReceiver) msg.MessageReceiver
• func RpcSessionStreamInterceptor(_ *pgxpool.Pool, logger edatlog.Logger) grpc.StreamServerInterceptor
• func RpcSessionUnrayInterceptor(conn *pgxpool.Pool, logger edatlog.Logger) grpc.UnaryServerInterceptor
• func WebSessionMiddleware(conn *pgxpool.Pool, logger edatlog.Logger) func(http.Handler) http.Handler
• type Client
  • func NewSessionClient() Client
• type EventStore
  • func NewEventStore(cliet Client, options ...EventStoreOption) *EventStore
  • func (s *EventStore) Load(ctx context.Context, root *es.AggregateRoot) error
  • func (s *EventStore) Save(ctx context.Context, root *es.AggregateRoot) (err error)
• type EventStoreOption
  • func WithEventStoreLogger(logger edatlog.Logger) EventStoreOption
  • func WithEventStoreTableName(tableName string) EventStoreOption
• type MessageStore
  • func NewMessageStore(client Client, options ...MessageStoreOption) *MessageStore
  • func (s *MessageStore) Close(ctx context.Context) error
  • func (s *MessageStore) Fetch(ctx context.Context, limit int) ([]outbox.Message, error)
  • func (s *MessageStore) MarkPublished(ctx context.Context, messageIDs []string) error
  • func (s *MessageStore) PurgePublished(ctx context.Context, olderThan time.Duration) error
  • func (s *MessageStore) Save(ctx context.Context, message outbox.Message) error
  • func (s *MessageStore) Send(ctx context.Context, channel string, message msg.Message) error
• type MessageStoreOption
  • func WithMessageStoreLogger(logger edatlog.Logger) MessageStoreOption
  • func WithMessageStoreTableName(tableName string) MessageStoreOption
• type SagaInstanceStore
  • func NewSagaInstanceStore(client Client, options ...SagaInstanceStoreOption) *SagaInstanceStore
  • func (s *SagaInstanceStore) Find(ctx context.Context, sagaName, sagaID string) (*saga.Instance, error)
  • func (s *SagaInstanceStore) Save(ctx context.Context, sagaInstance *saga.Instance) error
  • func (s *SagaInstanceStore) Update(ctx context.Context, sagaInstance *saga.Instance) error
• type SagaInstanceStoreOption
  • func WithSagaInstanceStoreLogger(logger edatlog.Logger) SagaInstanceStoreOption
  • func WithSagaInstanceStoreTableName(tableName string) SagaInstanceStoreOption
• type SnapshotStore
  • func (s *SnapshotStore) Load(ctx context.Context, root *es.AggregateRoot) error
  • func (s *SnapshotStore) Save(ctx context.Context, root *es.AggregateRoot) error
• type SnapshotStoreOption
  • func WithSnapshotStoreLogger(logger edatlog.Logger) SnapshotStoreOption
  • func WithSnapshotStoreStrategy(strategy es.SnapshotStrategy) SnapshotStoreOption
  • func WithSnapshotStoreTableName(tableName string) SnapshotStoreOption

Constants

const (
	DefaultEventTableName        = "events"
	DefaultMessageTableName      = "messages"
	DefaultSagaInstanceTableName = "saga_instances"
	DefaultSnapshotTableName     = "snapshots"

	CreateEventsTableSQL = `` /* 429-byte string literal not displayed */

	CreateMessagesTableSQL = `` /* 386-byte string literal not displayed */

	CreateMessagesUnpublishedIndexSQL = `CREATE INDEX unpublished_idx ON messages (created_at) WHERE not published`
	CreateMessagesPublishedIndexSQL   = `CREATE INDEX published_idx ON messages (modified_at) WHERE published`

	CreateSagaInstancesTableSQL = `` /* 410-byte string literal not displayed */

	CreateSnapshotsTableSQL = `` /* 344-byte string literal not displayed */

)

Variables

var ErrInvalidTxValue = errors.New("tx value is not a pgx.Tx type")

var ErrTxNotInContext = errors.New("pgx.Tx is not set for session")

Functions

func NewSnapshotStore

func NewSnapshotStore(client Client, options ...SnapshotStoreOption) es.AggregateRootStoreMiddleware

func ReceiverSessionMiddleware

func ReceiverSessionMiddleware(conn *pgxpool.Pool, logger edatlog.Logger) func(msg.MessageReceiver) msg.MessageReceiver

func RpcSessionStreamInterceptor

func RpcSessionStreamInterceptor(_ *pgxpool.Pool, logger edatlog.Logger) grpc.StreamServerInterceptor

func RpcSessionUnrayInterceptor

func RpcSessionUnrayInterceptor(conn *pgxpool.Pool, logger edatlog.Logger) grpc.UnaryServerInterceptor

func WebSessionMiddleware

func WebSessionMiddleware(conn *pgxpool.Pool, logger edatlog.Logger) func(http.Handler) http.Handler

Types

type Client

type Client interface {
	Exec(ctx context.Context, sql string, arguments ...interface{}) (pgconn.CommandTag, error)
	Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row
	QueryFunc(ctx context.Context, sql string, args []interface{}, scans []interface{}, f func(pgx.QueryFuncRow) error) (pgconn.CommandTag, error)
	SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
	Begin(ctx context.Context) (pgx.Tx, error)
	BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
}

Client covers a subset of what both pgx.Conn or pgxpool.Pool provide

func NewSessionClient

func NewSessionClient() Client

NewSessionClient returns a pgx.Conn or pgxpool.Pool compatible client that uses an active transaction from context

type EventStore

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

func NewEventStore

func NewEventStore(cliet Client, options ...EventStoreOption) *EventStore

func (*EventStore) Load

func (s *EventStore) Load(ctx context.Context, root *es.AggregateRoot) error

func (*EventStore) Save

func (s *EventStore) Save(ctx context.Context, root *es.AggregateRoot) (err error)

type EventStoreOption

type EventStoreOption func(store *EventStore)

func WithEventStoreLogger

func WithEventStoreLogger(logger edatlog.Logger) EventStoreOption

func WithEventStoreTableName

func WithEventStoreTableName(tableName string) EventStoreOption

type MessageStore

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

func NewMessageStore

func NewMessageStore(client Client, options ...MessageStoreOption) *MessageStore

func (*MessageStore) Close

func (s *MessageStore) Close(ctx context.Context) error

func (*MessageStore) Fetch

func (s *MessageStore) Fetch(ctx context.Context, limit int) ([]outbox.Message, error)

func (*MessageStore) MarkPublished

func (s *MessageStore) MarkPublished(ctx context.Context, messageIDs []string) error

func (*MessageStore) PurgePublished

func (s *MessageStore) PurgePublished(ctx context.Context, olderThan time.Duration) error

func (*MessageStore) Save

func (s *MessageStore) Save(ctx context.Context, message outbox.Message) error

func (*MessageStore) Send

func (s *MessageStore) Send(ctx context.Context, channel string, message msg.Message) error

type MessageStoreOption

type MessageStoreOption func(store *MessageStore)

func WithMessageStoreLogger

func WithMessageStoreLogger(logger edatlog.Logger) MessageStoreOption

func WithMessageStoreTableName

func WithMessageStoreTableName(tableName string) MessageStoreOption

type SagaInstanceStore

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

func NewSagaInstanceStore

func NewSagaInstanceStore(client Client, options ...SagaInstanceStoreOption) *SagaInstanceStore

func (*SagaInstanceStore) Find

func (s *SagaInstanceStore) Find(ctx context.Context, sagaName, sagaID string) (*saga.Instance, error)

func (*SagaInstanceStore) Save

func (s *SagaInstanceStore) Save(ctx context.Context, sagaInstance *saga.Instance) error

func (*SagaInstanceStore) Update

func (s *SagaInstanceStore) Update(ctx context.Context, sagaInstance *saga.Instance) error

type SagaInstanceStoreOption

type SagaInstanceStoreOption func(store *SagaInstanceStore)

func WithSagaInstanceStoreLogger

func WithSagaInstanceStoreLogger(logger edatlog.Logger) SagaInstanceStoreOption

func WithSagaInstanceStoreTableName

func WithSagaInstanceStoreTableName(tableName string) SagaInstanceStoreOption

type SnapshotStore

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

func (*SnapshotStore) Load

func (s *SnapshotStore) Load(ctx context.Context, root *es.AggregateRoot) error

func (*SnapshotStore) Save

func (s *SnapshotStore) Save(ctx context.Context, root *es.AggregateRoot) error

type SnapshotStoreOption

type SnapshotStoreOption func(store *SnapshotStore)

func WithSnapshotStoreLogger

func WithSnapshotStoreLogger(logger edatlog.Logger) SnapshotStoreOption

func WithSnapshotStoreStrategy

func WithSnapshotStoreStrategy(strategy es.SnapshotStrategy) SnapshotStoreOption

func WithSnapshotStoreTableName

func WithSnapshotStoreTableName(tableName string) SnapshotStoreOption