gcs_event_store

Documentation

Index

• func New(ctx context.Context, cfg *GCSConfig, options ...option.ClientOption) (storage.EventStore, error)
• type GCSConfig
  • func Config(bucket string) *GCSConfig
  • func (c *GCSConfig) NewContextWithTimeout(parent context.Context, operation Operation) (context.Context, context.CancelFunc)
  • func (c *GCSConfig) WithCompressor(compressor compression.Compressor) *GCSConfig
  • func (c *GCSConfig) WithFolder(folder string) *GCSConfig
  • func (c *GCSConfig) WithReadPolicy(readPolicy ReadPolicy) *GCSConfig
  • func (c *GCSConfig) WithTimeout(timeout Timeout) *GCSConfig
• type GCSEventStore
  • func (g *GCSEventStore) ListSourcesByKey(ctx context.Context, key string) ([]string, error)
  • func (g *GCSEventStore) LookUp(ctx context.Context, key, source string) (*event.Message, error)
  • func (g *GCSEventStore) LookUpByKey(ctx context.Context, key string) ([]*event.Message, error)
  • func (g *GCSEventStore) Persist(ctx context.Context, key, source, content string) error
• type ObjectIterator
• type Operation
• type ReadPolicy
  • func TakeFirstCreated() ReadPolicy
  • func TakeLastCreated() ReadPolicy
• type Timeout

Functions

func New

func New(ctx context.Context, cfg *GCSConfig, options ...option.ClientOption) (storage.EventStore, error)

Types

type GCSConfig

type GCSConfig struct {
	Bucket     string
	Compressor compression.Compressor
	Folder     *string
	ReadPolicy ReadPolicy
	Timeout    Timeout
}

func Config

func Config(bucket string) *GCSConfig

Config creates a default configuration, that - doesn't do compression/decompression when write & read to GCS - takes the earliest created object if there are multiple under the same key/source/ path - enforces universal 30s timeout in GCS requests

func (*GCSConfig) NewContextWithTimeout

func (c *GCSConfig) NewContextWithTimeout(
	parent context.Context,
	operation Operation) (context.Context, context.CancelFunc)

NewContextWithTimeout generates a new context.Context with timeout from the parent context. If neither operation-specific timeout nor default timeout is found, return parent context without any operation.

func (*GCSConfig) WithCompressor

func (c *GCSConfig) WithCompressor(compressor compression.Compressor) *GCSConfig

func (*GCSConfig) WithFolder

func (c *GCSConfig) WithFolder(folder string) *GCSConfig

func (*GCSConfig) WithReadPolicy

func (c *GCSConfig) WithReadPolicy(readPolicy ReadPolicy) *GCSConfig

func (*GCSConfig) WithTimeout

func (c *GCSConfig) WithTimeout(timeout Timeout) *GCSConfig

type GCSEventStore

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

GCSEventStore persists the contents in GCS, which requires consistent connections to Google Cloud.

func (*GCSEventStore) ListSourcesByKey

func (g *GCSEventStore) ListSourcesByKey(ctx context.Context, key string) ([]string, error)

func (*GCSEventStore) LookUp

func (g *GCSEventStore) LookUp(ctx context.Context, key, source string) (*event.Message, error)

LookUp returns a single message by looking up the path `folder/key/source`.

func (*GCSEventStore) LookUpByKey

func (g *GCSEventStore) LookUpByKey(ctx context.Context, key string) ([]*event.Message, error)

LookUpByKey returns a list of messages by looking up the prefix `folder/key/`.

func (*GCSEventStore) Persist

func (g *GCSEventStore) Persist(ctx context.Context, key, source, content string) error

Persist uploads the message as a file on the path `folder/key/source`.

type ObjectIterator

type ObjectIterator interface {
	Next() (*gcs.ObjectAttrs, error)
}

type Operation

type Operation string

const (
	ListContents Operation = "ListContents" // operation that lists all content associated with a given key
	ReadContent  Operation = "ReadContent"  // operation that reads content associated with a key-source pair
	WriteContent Operation = "WriteContent" //operation to writes content associated with a key-source pair
)

type ReadPolicy

type ReadPolicy interface {
	Apply(ObjectIterator) (*gcs.ObjectAttrs, error)
}

ReadPolicy defines the actions that the GCSEventStore takes when it got the object iterator via list query.

func TakeFirstCreated

func TakeFirstCreated() ReadPolicy

func TakeLastCreated

func TakeLastCreated() ReadPolicy

type Timeout

type Timeout struct {
	Default   *time.Duration              // the default timeout for all operations
	Operation map[Operation]time.Duration // the timeout of each operation - this overrides the default timeout
}