hippo

README

hippo

Documentation

Index

• Constants
• func Subscribe(c chan<- *Event, topics ...Topic)
• func Unsubscribe(c chan<- *Event)
• type Aggregate
• type CacheService
• type Client
  • func NewClient(s StoreService) *Client
  • func (c *Client) CacheService() CacheService
  • func (c *Client) Dispatch(ctx context.Context, event *Event, buffer interface{}, hooks ...HookFn) (*Aggregate, error)
  • func (c *Client) Fetch(ctx context.Context, aggregateID string, buffer interface{}, opt FetchOptions) (*Aggregate, error)
  • func (c *Client) RegisterCacheService(cache CacheService)
  • func (c *Client) RegisterDomainRules(fn DomainTypeRulesFn, domainType interface{})
  • func (c *Client) Rules(domainType interface{}) DomainTypeRulesFn
• type DomainTypeRulesFn
• type DomainTypeRulesMap
• type Error
  • func (e Error) Error() string
• type Event
  • func NewEvent(topic, aggregateID string) *Event
  • func NewEventProto(topic, aggregateID string, data proto.Message) *Event
  • func NewEventWithMetadata(topic, aggregateID string, metadata map[string]string) *Event
  • func (e *Event) GetTopic() Topic
  • func (e *Event) Marshal(in interface{}) error
  • func (e *Event) MarshalProto(pb proto.Message) error
  • func (e *Event) SetVersion(version int64)
  • func (e *Event) Unmarshal(out interface{}) error
  • func (e *Event) UnmarshalProto(pb proto.Message) error
• type EventService
• type FetchOptions
• type Format
• type HookFn
• type Message
• type Params
• type StoreService
• type Topic
• type Version

Constants

const (
	ErrParamsIDRequired = Error("parameter id required")
	ErrNotImplemented   = Error("feature not implemented")
)

Params errors.

const (
	ErrAggregateIDCanNotBeEmpty = Error("aggregateID can not be empty")
	ErrBufferCanNotBeNil        = Error("buffer can not be nil")
	ErrFormatNotProvided        = Error("format is not provided")
	ErrConcurrencyException     = Error("concurrency exception")
	ErrAggregateIDWithoutEvents = Error("aggregateID without events")
	ErrEmptyState               = Error("aggregate with empty state")
	ErrInvalidEventFormat       = Error("event data is not encoded in the right format")
	ErrInvalidSchema            = Error("invalid schema schema to decode event data")
)

Event errors.

const (
	ErrKeyDoesNotExist          = Error("key does not exist")
	ErrVersionFieldDoesNotExist = Error("invalid aggregate key - version field does not exist")
	ErrStateFieldDoesNotExist   = Error("invalid aggregate key - state field does not exist")
)

Cache errors.

Functions

func Subscribe

func Subscribe(c chan<- *Event, topics ...Topic)

Subscribe causes package pubsub to relay incoming events to c. If no events are provided, all incoming events will be relayed to c. Otherwise, just the provided events will.

Package pubsub will not block sending to c: the caller must ensure that c has sufficient buffer space to keep up with the expected event rate. For a channel used for notification of just one event value, a buffer of size 1 is sufficient.

It is allowed to call Subscribe multiple times with the same channel: each call expands the set of events sent to that channel.

It is allowed to call Subscribe multiple times with different channels and the same events: each channel receives copies of incoming events independently.

func Unsubscribe

func Unsubscribe(c chan<- *Event)

Unsubscribe remove events from the map.

Types

type Aggregate

type Aggregate struct {
	State   interface{}
	Version int64
	// contains filtered or unexported fields
}

Aggregate represents an aggregator state and respective version

type CacheService

type CacheService interface {
	Get(ctx context.Context, aggregateID string, out *Aggregate) error
	Set(ctx context.Context, aggregateID string, in *Aggregate) error
	DB() interface{}
}

CacheService represents a service for managing cache.

type Client

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

Client to hold store service implementation

func NewClient

func NewClient(s StoreService) *Client

NewClient return client struct

func (*Client) CacheService

func (c *Client) CacheService() CacheService

CacheService assigns a cache service to the client store

func (*Client) Dispatch

func (c *Client) Dispatch(ctx context.Context, event *Event, buffer interface{}, hooks ...HookFn) (*Aggregate, error)

Dispatch returns an aggregate resource based on the event and domain rules defined

func (*Client) Fetch

func (c *Client) Fetch(ctx context.Context, aggregateID string, buffer interface{}, opt FetchOptions) (*Aggregate, error)

Fetch returns an aggregate resource based on the aggregateID and domain type

func (*Client) RegisterCacheService

func (c *Client) RegisterCacheService(cache CacheService)

RegisterCacheService assigns a cache service to the client store

func (*Client) RegisterDomainRules

func (c *Client) RegisterDomainRules(fn DomainTypeRulesFn, domainType interface{})

RegisterDomainRules assigns a cache service to the client store

func (*Client) Rules

func (c *Client) Rules(domainType interface{}) DomainTypeRulesFn

Rules returns domain function rules for a specific domain type if domain types not previously registered log and return template function

type DomainTypeRulesFn

type DomainTypeRulesFn func(topic string, buffer, previous interface{}) (next interface{})

DomainTypeRulesFn represents a function type to define how data in buffer could change the previous State into the next State for a specific topic.

type DomainTypeRulesMap

type DomainTypeRulesMap map[string]DomainTypeRulesFn

DomainTypeRulesMap a map from domain type names to map domain type rules function

type Error

type Error string

Error represents a HIPPO error.

func (Error) Error

func (e Error) Error() string

Error returns the error message.

type Event

type Event struct {
	// Topic (name) of the event. These should be written in the past tense (event_created)
	Topic string
	// Aggregate ID is the primary key of the aggregate to which the event refers to.
	AggregateID string
	// Version of the aggregate, useful when using concurrency writes. (read-only)
	Version int64
	// Schema of the aggregate.
	Schema string
	// Format of the encoded type of the aggregate data
	Format Format
	// Data raw object data.
	Data []byte
	// Priority of the event, where 0 is the highest priority.
	Priority int32
	// TODO: Signature includes SHA1 signature computed against it's contents and signature
	// of the previous event. (not implemented yet)
	Signature string
	// TODO: Origin of the event. e.g. service name. (not implemented yet)
	OriginName string
	// TODO: Origin of the event. e.g. service ip address / browser. (not implemented yet)
	OriginIP string
	// Metadata
	Metadata map[string]string
	// CreateTime timestamp when event ocurred, location should be set to UTC.
	CreateTime time.Time
}

Event resource.

func NewEvent

func NewEvent(topic, aggregateID string) *Event

NewEvent returns an event resource.

func NewEventProto

func NewEventProto(topic, aggregateID string, data proto.Message) *Event

NewEventProto returns an event resource.

func NewEventWithMetadata

func NewEventWithMetadata(topic, aggregateID string, metadata map[string]string) *Event

NewEventWithMetadata returns an event resource.

func (*Event) GetTopic

func (e *Event) GetTopic() Topic

GetTopic returns event topic casted as Topic

func (*Event) Marshal

func (e *Event) Marshal(in interface{}) error

Marshal data to event data

func (*Event) MarshalProto

func (e *Event) MarshalProto(pb proto.Message) error

MarshalProto takes a protocol buffer message and encodes it into the wire format. Also sets the event schema as the underlying proto message type and encoded format

func (*Event) SetVersion

func (e *Event) SetVersion(version int64)

SetVersion assign event version

func (*Event) Unmarshal

func (e *Event) Unmarshal(out interface{}) error

Unmarshal event data to out

func (*Event) UnmarshalProto

func (e *Event) UnmarshalProto(pb proto.Message) error

UnmarshalProto parses the protocol buffer representation in buf and places the decoded result in pb. If the struct underlying pb does not match the data in buf, the results can be unpredictable.

type EventService

type EventService interface {
	Create(ctx context.Context, e *Event) error
	GetLastVersion(ctx context.Context, aggregateID string) (int64, error)
	List(ctx context.Context, p Params) ([]*Event, error)
}

EventService represents a service for managing an aggregate store.

type FetchOptions

type FetchOptions struct {
	SkipOptimisticConcurrency bool
}

FetchOptions is a configurable object for fetch func

type Format

type Format int32

Format enumerator

const (
	PROTOBUF Format = 0
	JSON     Format = 1
	STRING   Format = 2
)

type HookFn

type HookFn func(*Aggregate) error

HookFn represents a function type that will be called after the store is loaded. Please note that the new event is not yet dispatched / persisted when the HookFn is called.

type Message

type Message struct {
	ID    string
	Event *Event
}

Message represents ID and event to be dispatched

type Params

type Params struct {
	// ID is required
	ID string
	// FromVersion (optional)
	FromVersion int64
	// ToVersion (optional)
	ToVersion int64
}

Params represents parameters to load a store

type StoreService

type StoreService interface {
	EventService() EventService
}

StoreService represents a service for managing an aggregate store.

type Topic

type Topic string

Topic string name

type Version

type Version int64

Version type.