layers

Documentation

Index

• type DatasetRequest
• type Layer
  • func NewLayer(lc fx.Lifecycle, cmgr *conf.ConfigurationManager, env *conf.Env) *Layer
  • func (l *Layer) Dataset(request db.DatasetRequest) (ReadableDataset, error)
  • func (l *Layer) DoesDatasetExist(datasetName string) bool
  • func (l *Layer) GetContext(datasetName string) map[string]any
  • func (l *Layer) GetDatasetNames() []string
  • func (l *Layer) GetDatasetPostNames() []string
  • func (l *Layer) GetTableDefinition(datasetName string) *conf.TableMapping
• type PostLayer
  • func NewPostLayer(cmgr *conf.ConfigurationManager, logger *zap.SugaredLogger) *PostLayer
  • func (postLayer *PostLayer) Dataset(request db.DatasetRequest) (WriteableDataset, error)
• type PostRepository
• type PostgresDataset
  • func NewPostgresDataset(pg *pgxpool.Pool, table *db.ReadTable, writeTable *db.WriteTable, ...) *PostgresDataset
  • func (ds *PostgresDataset) Read(ctx context.Context, entities chan<- *uda.Entity) error
  • func (ds *PostgresDataset) ReadChanges(ctx context.Context, since string, entities chan<- *uda.Entity) (string, error)
  • func (ds *PostgresDataset) Write(ctx context.Context, entities <-chan *uda.Entity, entityContext *uda.Context) error
• type ReadableDataset
• type Repository
• type WriteableDataset

Types

type DatasetRequest

type DatasetRequest struct {
	DatasetName string
	Since       string
	Limit       int64
}

type Layer

type Layer struct {
	Repo *Repository //exported because it needs to deferred from main
	// contains filtered or unexported fields
}

func NewLayer

func NewLayer(lc fx.Lifecycle, cmgr *conf.ConfigurationManager, env *conf.Env) *Layer

func (*Layer) Dataset

func (l *Layer) Dataset(request db.DatasetRequest) (ReadableDataset, error)

func (*Layer) DoesDatasetExist

func (l *Layer) DoesDatasetExist(datasetName string) bool

func (*Layer) GetContext

func (l *Layer) GetContext(datasetName string) map[string]any

func (*Layer) GetDatasetNames

func (l *Layer) GetDatasetNames() []string

func (*Layer) GetDatasetPostNames

func (l *Layer) GetDatasetPostNames() []string

func (*Layer) GetTableDefinition

func (l *Layer) GetTableDefinition(datasetName string) *conf.TableMapping

type PostLayer

type PostLayer struct {
	PostRepo *PostRepository //exported because it needs to deferred from main??
	// contains filtered or unexported fields
}

func NewPostLayer

func NewPostLayer(cmgr *conf.ConfigurationManager, logger *zap.SugaredLogger) *PostLayer

func (*PostLayer) Dataset

func (postLayer *PostLayer) Dataset(request db.DatasetRequest) (WriteableDataset, error)

type PostRepository

type PostRepository struct {
	DB *pgxpool.Pool
	// contains filtered or unexported fields
}

type PostgresDataset

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

func NewPostgresDataset

func NewPostgresDataset(pg *pgxpool.Pool, table *db.ReadTable, writeTable *db.WriteTable, request db.DatasetRequest) *PostgresDataset

func (*PostgresDataset) Read

func (ds *PostgresDataset) Read(ctx context.Context, entities chan<- *uda.Entity) error

Read reads from a postgres query result, and emits an uda.Entity to the entities chan

func (*PostgresDataset) ReadChanges

func (ds *PostgresDataset) ReadChanges(ctx context.Context, since string, entities chan<- *uda.Entity) (string, error)

func (*PostgresDataset) Write

func (ds *PostgresDataset) Write(ctx context.Context, entities <-chan *uda.Entity, entityContext *uda.Context) error

Write takes a chan of uda.Entity and queues this in a batch request. Once the batch request has batchChunkSize in entities, it attempts to commit the batch in a transaction. If the transaction fails, an error is returned from the writer, and control is returned to the caller.

type ReadableDataset

type ReadableDataset interface {
	Read(ctx context.Context, entities chan<- *uda.Entity) error
	ReadChanges(ctx context.Context, since string, entities chan<- *uda.Entity) (string, error)
}

type Repository

type Repository struct {
	DB *pgxpool.Pool
	// contains filtered or unexported fields
}

type WriteableDataset

type WriteableDataset interface {
	Write(ctx context.Context, entities <-chan *uda.Entity, entityContext *uda.Context) error
}