datastore

Documentation

Overview

Package datastore implements components to fetch data from postgres or other sources.

The package contains a cache component, to fetch a value only once.

All components of the datastore are independent and can be put together.

The data flows in both directions. With the Get() method, keys can be requested. With the Update() method, values get pushed in the other direction. All components implement the two methods and thereby the flow.Flow interface.

Index

• type FlowPostgres
  • func NewFlowPostgres(lookup environment.Environmenter, updater flow.Updater) (*FlowPostgres, error)
  • func (p *FlowPostgres) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key][]byte, error)
  • func (p *FlowPostgres) HistoryInformation(ctx context.Context, fqid string, w io.Writer) error
  • func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][]byte, error))
• type FlowVoteCount
  • func NewFlowVoteCount(lookup environment.Environmenter) *FlowVoteCount
  • func (s *FlowVoteCount) Connect(ctx context.Context, eventProvider func() (<-chan time.Time, func() bool), ...)
  • func (s *FlowVoteCount) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key][]byte, error)
  • func (s *FlowVoteCount) Update(ctx context.Context, updateFn func(map[dskey.Key][]byte, error))
• type InvalidDataError
  • func (err InvalidDataError) Error() string

Types

type FlowPostgres

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

FlowPostgres uses postgres to get the connections.

func NewFlowPostgres

func NewFlowPostgres(lookup environment.Environmenter, updater flow.Updater) (*FlowPostgres, error)

NewFlowPostgres initializes a SourcePostgres.

TODO: This should be unexported, but there is an import cycle in the tests.

func (*FlowPostgres) Get

func (p *FlowPostgres) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key][]byte, error)

Get fetches the keys from postgres.

func (*FlowPostgres) HistoryInformation

func (p *FlowPostgres) HistoryInformation(ctx context.Context, fqid string, w io.Writer) error

HistoryInformation fetches the history information for one fqid.

func (*FlowPostgres) Update

func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][]byte, error))

Update calls the updater.

type FlowVoteCount

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

FlowVoteCount is a datastore flow for the poll/vote_count value.

func NewFlowVoteCount

func NewFlowVoteCount(lookup environment.Environmenter) *FlowVoteCount

NewFlowVoteCount initializes the object.

func (*FlowVoteCount) Connect

func (s *FlowVoteCount) Connect(ctx context.Context, eventProvider func() (<-chan time.Time, func() bool), errHandler func(error))

Connect creates a connection to the vote service and makes sure, it stays open.

eventProvider is a function that returns a channel. If the connection fails, this function fetches such a channel and waits for a signal before it tries to open a new connection.

func (*FlowVoteCount) Get

func (s *FlowVoteCount) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key][]byte, error)

Get is called when a key is not in the cache.

func (*FlowVoteCount) Update

func (s *FlowVoteCount) Update(ctx context.Context, updateFn func(map[dskey.Key][]byte, error))

Update has to be called frequently. It blocks, until there is new data.

type InvalidDataError

type InvalidDataError struct {
	Key   dskey.Key
	Value []byte
}

InvalidDataError is returned, when a key has an invalid value in the database.

func (InvalidDataError) Error

func (err InvalidDataError) Error() string