databroker

Documentation

Overview

Package databroker contains a data broker implementation.

Index

• Variables
• type ConfigSource
  • func NewConfigSource(ctx context.Context, underlying config.Source, ...) *ConfigSource
  • func (src *ConfigSource) GetConfig() *config.Config
• type Server
  • func New(options ...ServerOption) *Server
  • func (srv *Server) Get(ctx context.Context, req *databroker.GetRequest) (*databroker.GetResponse, error)
  • func (srv *Server) Put(ctx context.Context, req *databroker.PutRequest) (*databroker.PutResponse, error)
  • func (srv *Server) Query(ctx context.Context, req *databroker.QueryRequest) (*databroker.QueryResponse, error)
  • func (srv *Server) SetOptions(ctx context.Context, req *databroker.SetOptionsRequest) (*databroker.SetOptionsResponse, error)
  • func (srv *Server) Sync(req *databroker.SyncRequest, stream databroker.DataBrokerService_SyncServer) error
  • func (srv *Server) SyncLatest(req *databroker.SyncLatestRequest, ...) error
  • func (srv *Server) UpdateConfig(options ...ServerOption)
• type ServerOption
  • func WithDeletePermanentlyAfter(dur time.Duration) ServerOption
  • func WithGetAllPageSize(pageSize int) ServerOption
  • func WithGetSharedKey(getSharedKey func() ([]byte, error)) ServerOption
  • func WithStorageCAFile(filePath string) ServerOption
  • func WithStorageCertSkipVerify(storageCertSkipVerify bool) ServerOption
  • func WithStorageCertificate(certificate *tls.Certificate) ServerOption
  • func WithStorageConnectionString(connStr string) ServerOption
  • func WithStorageType(typ string) ServerOption

Variables

var (
	// DefaultDeletePermanentlyAfter is the default amount of time to wait before deleting
	// a record permanently.
	DefaultDeletePermanentlyAfter = time.Hour
	// DefaultStorageType is the default storage type that Server use
	DefaultStorageType = "memory"
	// DefaultGetAllPageSize is the default page size for GetAll calls.
	DefaultGetAllPageSize = 50
)

Types

type ConfigSource

type ConfigSource struct {
	config.ChangeDispatcher
	// contains filtered or unexported fields
}

ConfigSource provides a new Config source that decorates an underlying config with configuration derived from the data broker.

func NewConfigSource

func NewConfigSource(ctx context.Context, underlying config.Source, listeners ...config.ChangeListener) *ConfigSource

NewConfigSource creates a new ConfigSource.

func (*ConfigSource) GetConfig

func (src *ConfigSource) GetConfig() *config.Config

GetConfig gets the current config.

type Server

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

Server implements the databroker service using an in memory database.

func New

func New(options ...ServerOption) *Server

New creates a new server.

func (*Server) Get

func (srv *Server) Get(ctx context.Context, req *databroker.GetRequest) (*databroker.GetResponse, error)

Get gets a record from the in-memory list.

func (*Server) Put

func (srv *Server) Put(ctx context.Context, req *databroker.PutRequest) (*databroker.PutResponse, error)

Put updates an existing record or adds a new one.

func (*Server) Query

func (srv *Server) Query(ctx context.Context, req *databroker.QueryRequest) (*databroker.QueryResponse, error)

Query queries for records.

func (*Server) SetOptions

func (srv *Server) SetOptions(ctx context.Context, req *databroker.SetOptionsRequest) (*databroker.SetOptionsResponse, error)

SetOptions sets options for a type in the databroker.

func (*Server) Sync

func (srv *Server) Sync(req *databroker.SyncRequest, stream databroker.DataBrokerService_SyncServer) error

Sync streams updates for the given record type.

func (*Server) SyncLatest

func (srv *Server) SyncLatest(req *databroker.SyncLatestRequest, stream databroker.DataBrokerService_SyncLatestServer) error

SyncLatest returns the latest value of every record in the databroker as a stream of records.

func (*Server) UpdateConfig

func (srv *Server) UpdateConfig(options ...ServerOption)

UpdateConfig updates the server with the new options.

type ServerOption

type ServerOption func(*serverConfig)

A ServerOption customizes the server.

func WithDeletePermanentlyAfter

func WithDeletePermanentlyAfter(dur time.Duration) ServerOption

WithDeletePermanentlyAfter sets the deletePermanentlyAfter duration. If a record is deleted via Delete, it will be permanently deleted after the given duration.

func WithGetAllPageSize

func WithGetAllPageSize(pageSize int) ServerOption

WithGetAllPageSize sets the page size for GetAll calls.

func WithGetSharedKey

func WithGetSharedKey(getSharedKey func() ([]byte, error)) ServerOption

WithGetSharedKey sets the secret in the config.

func WithStorageCAFile

func WithStorageCAFile(filePath string) ServerOption

WithStorageCAFile sets the CA file in the config.

func WithStorageCertSkipVerify

func WithStorageCertSkipVerify(storageCertSkipVerify bool) ServerOption

WithStorageCertSkipVerify sets the storageCertSkipVerify in the config.

func WithStorageCertificate

func WithStorageCertificate(certificate *tls.Certificate) ServerOption

WithStorageCertificate sets the storageCertificate in the config.

func WithStorageConnectionString

func WithStorageConnectionString(connStr string) ServerOption

WithStorageConnectionString sets the DSN for storage.

func WithStorageType

func WithStorageType(typ string) ServerOption

WithStorageType sets the storage type.