state

README

State Stores

State Stores provide a common way to interact with different data store implementations, and allow users to opt-in to advanced capabilities using defined metadata.

Currently supported state stores are:

• AWS DynamoDB
• Aerospike
• Azure Blob Storage
• Azure CosmosDB
• Azure Table Storage
• Cassandra
• Cloud Firestore (Datastore mode)
• CloudState
• Couchbase
• Etcd
• HashiCorp Consul
• Hazelcast
• Memcached
• MongoDB
• PostgreSQL
• Redis
• RethinkDB
• SQL Server
• Zookeeper

Implementing a new State Store

A compliant state store needs to implement one or more interfaces: Store and TransactionalStore.

The interface for Store:

type Store interface {
	Init(metadata Metadata) error
	Delete(req *DeleteRequest) error
	BulkDelete(req []DeleteRequest) error
	Get(req *GetRequest) (*GetResponse, error)
	Set(req *SetRequest) error
	BulkSet(req []SetRequest) error
}

The interface for TransactionalStore:

type TransactionalStore interface {
	Init(metadata Metadata) error
	Multi(reqs []TransactionalRequest) error
}

See the documentation site for examples.

Documentation

Index

• Constants
• func CheckRequestOptions(options interface{}) error
• func DeleteWithOptions(method func(req *DeleteRequest) error, req *DeleteRequest) error
• func SetWithOptions(method func(req *SetRequest) error, req *SetRequest) error
• type BulkGetResponse
• type BulkStore
• type DefaultBulkStore
  • func NewDefaultBulkStore(store Store) DefaultBulkStore
  • func (b *DefaultBulkStore) BulkDelete(req []DeleteRequest) error
  • func (b *DefaultBulkStore) BulkGet(req []GetRequest) (bool, []BulkGetResponse, error)
  • func (b *DefaultBulkStore) BulkSet(req []SetRequest) error
  • func (b *DefaultBulkStore) Features() []Feature
• type DeleteRequest
  • func (r DeleteRequest) GetKey() string
  • func (r DeleteRequest) GetMetadata() map[string]string
• type DeleteStateOption
• type ETagError
  • func NewETagError(kind ETagErrorKind, err error) *ETagError
  • func (e *ETagError) Error() string
  • func (e *ETagError) Kind() ETagErrorKind
• type ETagErrorKind
• type Feature
  • func (f Feature) IsPresent(features []Feature) bool
• type GetRequest
• type GetResponse
• type GetStateOption
• type KeyInt
• type Metadata
• type OperationType
• type SetRequest
  • func (r SetRequest) GetKey() string
  • func (r SetRequest) GetMetadata() map[string]string
• type SetStateOption
• type Store
• type TransactionalStateOperation
• type TransactionalStateRequest
• type TransactionalStore

Constants

const (
	FirstWrite = "first-write"
	LastWrite  = "last-write"
	Strong     = "strong"
	Eventual   = "eventual"
)

Functions

func CheckRequestOptions

func CheckRequestOptions(options interface{}) error

CheckRequestOptions checks if request options use supported keywords

func DeleteWithOptions

func DeleteWithOptions(method func(req *DeleteRequest) error, req *DeleteRequest) error

DeleteWithOptions handles DeleteRequest with options

func SetWithOptions

func SetWithOptions(method func(req *SetRequest) error, req *SetRequest) error

SetWithOptions handles SetRequest with request options

Types

type BulkGetResponse

type BulkGetResponse struct {
	Key      string            `json:"key"`
	Data     []byte            `json:"data"`
	ETag     *string           `json:"etag,omitempty"`
	Metadata map[string]string `json:"metadata"`
	Error    string            `json:"error,omitempty"`
}

BulkGetResponse is the response object for bulk get response.

type BulkStore

type BulkStore interface {
	BulkDelete(req []DeleteRequest) error
	BulkGet(req []GetRequest) (bool, []BulkGetResponse, error)
	BulkSet(req []SetRequest) error
}

BulkStore is an interface to perform bulk operations on store

type DefaultBulkStore

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

DefaultBulkStore is a default implementation of BulkStore

func NewDefaultBulkStore

func NewDefaultBulkStore(store Store) DefaultBulkStore

NewDefaultBulkStore build a default bulk store

func (*DefaultBulkStore) BulkDelete

func (b *DefaultBulkStore) BulkDelete(req []DeleteRequest) error

BulkDelete performs a bulk delete operation

func (*DefaultBulkStore) BulkGet

func (b *DefaultBulkStore) BulkGet(req []GetRequest) (bool, []BulkGetResponse, error)

BulkGet performs a bulks get operations

func (*DefaultBulkStore) BulkSet

func (b *DefaultBulkStore) BulkSet(req []SetRequest) error

BulkSet performs a bulks save operation

func (*DefaultBulkStore) Features

func (b *DefaultBulkStore) Features() []Feature

Features returns the features of the encapsulated store

type DeleteRequest

type DeleteRequest struct {
	Key      string            `json:"key"`
	ETag     *string           `json:"etag,omitempty"`
	Metadata map[string]string `json:"metadata"`
	Options  DeleteStateOption `json:"options,omitempty"`
}

DeleteRequest is the object describing a delete state request

func (DeleteRequest) GetKey

func (r DeleteRequest) GetKey() string

Key gets the Key on a DeleteRequest

func (DeleteRequest) GetMetadata

func (r DeleteRequest) GetMetadata() map[string]string

Metadata gets the Metadata on a DeleteRequest

type DeleteStateOption

type DeleteStateOption struct {
	Concurrency string `json:"concurrency,omitempty"` // "concurrency"
	Consistency string `json:"consistency"`           // "eventual, strong"
}

DeleteStateOption controls how a state store reacts to a delete request

type ETagError

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

ETagError is a custom error type for etag exceptions

func NewETagError

func NewETagError(kind ETagErrorKind, err error) *ETagError

NewETagError returns an ETagError wrapping an existing context error

func (*ETagError) Error

func (e *ETagError) Error() string

func (*ETagError) Kind

func (e *ETagError) Kind() ETagErrorKind

type ETagErrorKind

type ETagErrorKind string

const (
	ETagInvalid  ETagErrorKind = "invalid"
	ETagMismatch ETagErrorKind = "mismatch"
)

type Feature

type Feature string

Feature names a feature that can be implemented by PubSub components.

const (
	// FeatureETag is the feature to etag metadata in state store.
	FeatureETag Feature = "ETAG"
	// FeatureTransactional is the feature that performs transactional operations.
	FeatureTransactional Feature = "TRANSACTIONAL"
)

func (Feature) IsPresent

func (f Feature) IsPresent(features []Feature) bool

IsPresent checks if a given feature is present in the list.

type GetRequest

type GetRequest struct {
	Key      string            `json:"key"`
	Metadata map[string]string `json:"metadata"`
	Options  GetStateOption    `json:"options,omitempty"`
}

GetRequest is the object describing a state fetch request

type GetResponse

type GetResponse struct {
	Data     []byte            `json:"data"`
	ETag     *string           `json:"etag,omitempty"`
	Metadata map[string]string `json:"metadata"`
}

GetResponse is the request object for getting state

type GetStateOption

type GetStateOption struct {
	Consistency string `json:"consistency"` // "eventual, strong"
}

GetStateOption controls how a state store reacts to a get request

type KeyInt

type KeyInt interface {
	GetKey() string
	GetMetadata() map[string]string
}

KeyInt is an interface that allows gets of the Key and Metadata inside requests

type Metadata

type Metadata struct {
	Properties map[string]string `json:"properties"`
}

Metadata contains a state store specific set of metadata properties

type OperationType

type OperationType string

OperationType describes a CRUD operation performed against a state store

const Delete OperationType = "delete"

Delete is a delete operation

const Upsert OperationType = "upsert"

Upsert is an update or create operation

type SetRequest

type SetRequest struct {
	Key      string            `json:"key"`
	Value    interface{}       `json:"value"`
	ETag     *string           `json:"etag,omitempty"`
	Metadata map[string]string `json:"metadata,omitempty"`
	Options  SetStateOption    `json:"options,omitempty"`
}

SetRequest is the object describing an upsert request

func (SetRequest) GetKey

func (r SetRequest) GetKey() string

GetKey gets the Key on a SetRequest

func (SetRequest) GetMetadata

func (r SetRequest) GetMetadata() map[string]string

GetMetadata gets the Key on a SetRequest

type SetStateOption

type SetStateOption struct {
	Concurrency string `json:"concurrency,omitempty"` // first-write, last-write
	Consistency string `json:"consistency"`           // "eventual, strong"
}

SetStateOption controls how a state store reacts to a set request

type Store

type Store interface {
	BulkStore
	Init(metadata Metadata) error
	Features() []Feature
	Delete(req *DeleteRequest) error
	Get(req *GetRequest) (*GetResponse, error)
	Set(req *SetRequest) error
	Ping() error
}

Store is an interface to perform operations on store

type TransactionalStateOperation

type TransactionalStateOperation struct {
	Operation OperationType `json:"operation"`
	Request   interface{}   `json:"request"`
}

TransactionalStateOperation describes operation type, key, and value for transactional operation.

type TransactionalStateRequest

type TransactionalStateRequest struct {
	Operations []TransactionalStateOperation `json:"operations"`
	Metadata   map[string]string             `json:"metadata,omitempty"`
}

TransactionalStateRequest describes a transactional operation against a state store that comprises multiple types of operations The Request field is either a DeleteRequest or SetRequest

type TransactionalStore

type TransactionalStore interface {
	Init(metadata Metadata) error
	Multi(request *TransactionalStateRequest) error
}

TransactionalStore is an interface for initialization and support multiple transactional requests