rangedb

README

RangeDB

An event store database in Go. This package includes a stand-alone database and web server along with a library for embedding event sourced applications.

Examples are provided here.

Docker

docker run -p 8080:8080 inklabs/rangedb

Documentation

Index

• Constants
• func GetAllEventsByAggregateTypes(store Store, aggregateTypes ...string) []<-chan *Record
• func GetEventStream(message AggregateMessage) string
• func GetStream(aggregateType, aggregateID string) string
• func MergeRecordChannelsInOrder(channels []<-chan *Record) <-chan *Record
• func ReplayEvents(store Store, subscribers ...RecordSubscriber)
• type AggregateMessage
• type Event
• type Record
  • func RecordChannelToSlice(records <-chan *Record) []*Record
• type RecordIoStream
• type RecordSerializer
• type RecordSubscriber
• type Store

Constants

const Version = "0.2.1"

Functions

func GetAllEventsByAggregateTypes

func GetAllEventsByAggregateTypes(store Store, aggregateTypes ...string) []<-chan *Record

GetAllEventsByAggregateTypes returns a slice of Record channels by aggregateType.

func GetEventStream

func GetEventStream(message AggregateMessage) string

GetEventStream returns the stream name for an event.

func GetStream

func GetStream(aggregateType, aggregateID string) string

GetStream returns the stream name for an aggregateType and aggregateID.

func MergeRecordChannelsInOrder

func MergeRecordChannelsInOrder(channels []<-chan *Record) <-chan *Record

MergeRecordChannelsInOrder combines record channels ordered by record.GlobalSequenceNumber.

func ReplayEvents

func ReplayEvents(store Store, subscribers ...RecordSubscriber)

ReplayEvents applies all events to each subscriber.

Types

type AggregateMessage

type AggregateMessage interface {
	AggregateID() string
	AggregateType() string
}

AggregateMessage is the interface that supports building an event stream name.

type Event

type Event interface {
	AggregateMessage
	EventType() string
}

Event is the interface that defines the required event methods.

type Record

type Record struct {
	AggregateType        string      `msgpack:"a" json:"aggregateType"`
	AggregateID          string      `msgpack:"i" json:"aggregateID"`
	GlobalSequenceNumber uint64      `msgpack:"g" json:"globalSequenceNumber"`
	StreamSequenceNumber uint64      `msgpack:"s" json:"sequenceNumber"`
	InsertTimestamp      uint64      `msgpack:"u" json:"insertTimestamp"`
	EventID              string      `msgpack:"e" json:"eventID"`
	EventType            string      `msgpack:"t" json:"eventType"`
	Data                 interface{} `msgpack:"d" json:"data"`
	Metadata             interface{} `msgpack:"m" json:"metadata"`
}

Record contains event data and metadata.

func RecordChannelToSlice

func RecordChannelToSlice(records <-chan *Record) []*Record

RecordChannelToSlice reads all records from the channel into a slice

type RecordIoStream

type RecordIoStream interface {
	Read(io.Reader) (<-chan *Record, <-chan error)
	Write(io.Writer, <-chan *Record) <-chan error
	Bind(events ...Event)
}

RecordIoStream is the interface that (de)serializes a stream of Records.

type RecordSerializer

type RecordSerializer interface {
	Serialize(record *Record) ([]byte, error)
	Deserialize(data []byte) (*Record, error)
	Bind(events ...Event)
}

RecordSerializer is the interface that (de)serializes Records.

type RecordSubscriber

type RecordSubscriber interface {
	Accept(record *Record)
}

RecordSubscriber is the interface that defines how a projection receives Records.

type Store

type Store interface {
	AllEvents() <-chan *Record
	AllEventsByAggregateType(aggregateType string) <-chan *Record
	AllEventsByAggregateTypes(aggregateTypes ...string) <-chan *Record
	AllEventsByStream(stream string) <-chan *Record
	EventsByAggregateType(pagination paging.Pagination, aggregateType string) <-chan *Record
	EventsByAggregateTypeStartingWith(aggregateType string, eventNumber uint64) <-chan *Record
	EventsByStream(pagination paging.Pagination, streamName string) <-chan *Record
	EventsByStreamStartingWith(streamName string, eventNumber uint64) <-chan *Record
	Save(event Event, metadata interface{}) error
	SaveEvent(aggregateType, aggregateID, eventType, eventID string, event, metadata interface{}) error
	Subscribe(subscribers ...RecordSubscriber)
	SubscribeAndReplay(subscribers ...RecordSubscriber)
	TotalEventsInStream(streamName string) uint64
}

Store is the interface that stores and retrieves event records.