inmemdatastore

Documentation

Index

• Constants
• func GetAvroCodec(schema string) (*goavro.Codec, error)
• func GetDatastoreShardId(key string, numShards int) uint64
• type AvroFileWriter
  • func NewAvroFileWriter(ctx context.Context, wg *sync.WaitGroup, cfg AvroFileWriterConfig) *AvroFileWriter
  • func (a *AvroFileWriter) Shutdown()
  • func (a *AvroFileWriter) Write(data interface{}) error
• type AvroFileWriterConfig
• type AvroSerializer
  • func (a *AvroSerializer) Serialize(record map[string]interface{}) (interface{}, error)
• type Config
• type Datastore
  • func NewDatastore(id uint64) *Datastore
• type Datastores
• type InMemDataStore
  • func NewInMemDatastore(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, cfg Config) *InMemDataStore
  • func (ds *InMemDataStore) Get(key string) (interface{}, error)
  • func (ds *InMemDataStore) GetAll() map[string]interface{}
  • func (ds *InMemDataStore) GetDatastores() Datastores
  • func (ds *InMemDataStore) Put(key string, val map[string]interface{}) error
  • func (ds *InMemDataStore) Shutdown()
  • func (ds *InMemDataStore) Start()
• type NoopSerializer
  • func (n *NoopSerializer) Serialize(record map[string]interface{}) (interface{}, error)
• type PersistenceChan
• type Persister
  • func NewPersister(ctx context.Context, wg *sync.WaitGroup, cfg PersisterConfig) *Persister
  • func (p *Persister) Run()
• type PersisterConfig
• type Persisters
• type Serializer
  • func NewAvroSerializer(avroSchema string) Serializer
  • func NewNoopSerializer() Serializer
• type Writer

Constants

const (
	RecFieldId = "id"
)

Functions

func GetAvroCodec

func GetAvroCodec(schema string) (*goavro.Codec, error)

func GetDatastoreShardId

func GetDatastoreShardId(key string, numShards int) uint64

Types

type AvroFileWriter

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

func NewAvroFileWriter

func NewAvroFileWriter(ctx context.Context, wg *sync.WaitGroup, cfg AvroFileWriterConfig) *AvroFileWriter

func (*AvroFileWriter) Shutdown

func (a *AvroFileWriter) Shutdown()

func (*AvroFileWriter) Write

func (a *AvroFileWriter) Write(data interface{}) error

type AvroFileWriterConfig

type AvroFileWriterConfig struct {
	Id         int
	AvroSchema string
	OutputDir  string
}

type AvroSerializer

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

func (*AvroSerializer) Serialize

func (a *AvroSerializer) Serialize(record map[string]interface{}) (interface{}, error)

type Config

type Config struct {
	NumDatastoreShards      int
	PersistenceChan         PersistenceChan
	PersistanceChanBuffSize int
	Persisters              Persisters
	// The top-level key in the map[string]interface{} records that will be stored in the
	// InMemoryDataStore.  This is required for the Put method to process a new record.
	RecordTimestampKey string
}

type Datastore

type Datastore struct {
	Id        uint64
	Data      map[string]interface{}
	NumReads  int64
	NumWrites int64
	// contains filtered or unexported fields
}

func NewDatastore

func NewDatastore(id uint64) *Datastore

type Datastores

type Datastores map[uint64]*Datastore

type InMemDataStore

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

func NewInMemDatastore

func NewInMemDatastore(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, cfg Config) *InMemDataStore

func (*InMemDataStore) Get

func (ds *InMemDataStore) Get(key string) (interface{}, error)

func (*InMemDataStore) GetAll

func (ds *InMemDataStore) GetAll() map[string]interface{}

GetAll will return the entire in memory data store. For a production grade piece of software we would have to do something other than just returning a reference to the private map, such as a deep copy. For now, this is just here to facilitate testing.

func (*InMemDataStore) GetDatastores

func (ds *InMemDataStore) GetDatastores() Datastores

func (*InMemDataStore) Put

func (ds *InMemDataStore) Put(key string, val map[string]interface{}) error

func (*InMemDataStore) Shutdown

func (ds *InMemDataStore) Shutdown()

Shutdown will signal the Serializers to close their open file handles and shutdown the IMDS.

func (*InMemDataStore) Start

func (ds *InMemDataStore) Start()

Start will spin up the Persisters and when it returns will be ready for reads and writes.

type NoopSerializer

type NoopSerializer struct{}

func (*NoopSerializer) Serialize

func (n *NoopSerializer) Serialize(record map[string]interface{}) (interface{}, error)

type PersistenceChan

type PersistenceChan chan map[string]interface{}

type Persister

type Persister struct {
	Serializer
	Writer
	// contains filtered or unexported fields
}

func NewPersister

func NewPersister(ctx context.Context, wg *sync.WaitGroup, cfg PersisterConfig) *Persister

func (*Persister) Run

func (p *Persister) Run()

type PersisterConfig

type PersisterConfig struct {
	Id         int
	Serializer Serializer
	Writer     Writer
	InputChan  PersistenceChan
}

type Persisters

type Persisters map[int]*Persister

type Serializer

type Serializer interface {
	Serialize(map[string]interface{}) (interface{}, error)
}

func NewAvroSerializer

func NewAvroSerializer(avroSchema string) Serializer

func NewNoopSerializer

func NewNoopSerializer() Serializer

type Writer

type Writer interface {
	Write(interface{}) error
	Shutdown()
}