repository

Documentation

Index

• Constants
• type InMemoryRepository
  • func NewInMemoryRepository() *InMemoryRepository
  • func (r *InMemoryRepository) AppendEvent(ctx context.Context, req *annales.AppendEventRequest) (*annales.AppendEventResponse, error)
  • func (r *InMemoryRepository) GetStreamEvents(ctx context.Context, req *annales.GetStreamEventsRequest) (*annales.GetStreamEventsResponse, error)
  • func (r *InMemoryRepository) GetStreamInfo(ctx context.Context, in *annales.GetStreamInfoRequest) (*annales.GetStreamInfoResponse, error)
• type MongoRepository
  • func NewMongoRepository(ctx context.Context, client *mongo.Client, dbName string) (*MongoRepository, error)
  • func (md *MongoRepository) AppendEvent(ctx context.Context, in *annales.AppendEventRequest) (*annales.AppendEventResponse, error)
  • func (j *MongoRepository) GetStreamEvents(ctx context.Context, req *annales.GetStreamEventsRequest) (*annales.GetStreamEventsResponse, error)
  • func (md *MongoRepository) GetStreamInfo(ctx context.Context, in *annales.GetStreamInfoRequest) (*annales.GetStreamInfoResponse, error)
• type PostgresRepository
  • func NewPostgresRepository(db *sql.DB) (*PostgresRepository, error)
  • func (pd *PostgresRepository) AppendEvent(ctx context.Context, req *annales.AppendEventRequest) (*annales.AppendEventResponse, error)
  • func (pd *PostgresRepository) GetStreamEvents(ctx context.Context, req *annales.GetStreamEventsRequest) (*annales.GetStreamEventsResponse, error)
  • func (pd *PostgresRepository) GetStreamInfo(ctx context.Context, in *annales.GetStreamInfoRequest) (*annales.GetStreamInfoResponse, error)
• type Repository

Constants

const (
	EventsCollectionName = "events"
)

Types

type InMemoryRepository

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

An in-memory repository for test purposes only.

func NewInMemoryRepository

func NewInMemoryRepository() *InMemoryRepository

func (*InMemoryRepository) AppendEvent

func (r *InMemoryRepository) AppendEvent(ctx context.Context, req *annales.AppendEventRequest) (*annales.AppendEventResponse, error)

func (*InMemoryRepository) GetStreamEvents

func (r *InMemoryRepository) GetStreamEvents(ctx context.Context, req *annales.GetStreamEventsRequest) (*annales.GetStreamEventsResponse, error)

FIXME events are not immutable as they would be from a database. I do not think it is necessary to fix for now.

func (*InMemoryRepository) GetStreamInfo

func (r *InMemoryRepository) GetStreamInfo(ctx context.Context, in *annales.GetStreamInfoRequest) (*annales.GetStreamInfoResponse, error)

type MongoRepository

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

func NewMongoRepository

func NewMongoRepository(ctx context.Context, client *mongo.Client, dbName string) (*MongoRepository, error)

func (*MongoRepository) AppendEvent

func (md *MongoRepository) AppendEvent(ctx context.Context, in *annales.AppendEventRequest) (*annales.AppendEventResponse, error)

AppendEventImpl appends a new event to the specified stream in MongoDB Assumes that all the optimistic locking has been performed by AbstractCachingJournal

func (*MongoRepository) GetStreamEvents

func (j *MongoRepository) GetStreamEvents(ctx context.Context, req *annales.GetStreamEventsRequest) (*annales.GetStreamEventsResponse, error)

func (*MongoRepository) GetStreamInfo

func (md *MongoRepository) GetStreamInfo(ctx context.Context, in *annales.GetStreamInfoRequest) (*annales.GetStreamInfoResponse, error)

type PostgresRepository

type PostgresRepository struct {
	annales.UnimplementedJournalServer
	// contains filtered or unexported fields
}

func NewPostgresRepository

func NewPostgresRepository(db *sql.DB) (*PostgresRepository, error)

func (*PostgresRepository) AppendEvent

func (pd *PostgresRepository) AppendEvent(ctx context.Context, req *annales.AppendEventRequest) (*annales.AppendEventResponse, error)

func (*PostgresRepository) GetStreamEvents

func (pd *PostgresRepository) GetStreamEvents(ctx context.Context, req *annales.GetStreamEventsRequest) (*annales.GetStreamEventsResponse, error)

GetEventsByStream retrieves all events belonging to a specific stream_id

func (*PostgresRepository) GetStreamInfo

func (pd *PostgresRepository) GetStreamInfo(ctx context.Context, in *annales.GetStreamInfoRequest) (*annales.GetStreamInfoResponse, error)

type Repository

type Repository interface {
	AppendEvent(ctx context.Context, in *annales.AppendEventRequest) (*annales.AppendEventResponse, error)
	GetStreamInfo(ctx context.Context, in *annales.GetStreamInfoRequest) (*annales.GetStreamInfoResponse, error)
}

A Repository is a database-specific implementation of the appender and the peeker. It is not require to think about optimistic locking or caching, that is a responsibility of the journal. Yes, the interface definition is a bit opaque but the objective is to avoid unnecessary allocations and data mappings.