eda: github.com/chop-dbhi/eda Index | Files | Directories

package eda

import "github.com/chop-dbhi/eda"

The eda package is a library for implementing event-driven architectures. It provides a thin layer on top of backends that support ordered streams with a publish/subscribe interface. The current implementation uses NATS Streaming: https://github.com/nats-io/nats-streaming-server, but additional backends could be supported.

Use Case

The primary use case this library is being designed to support are applications involving "domain events". That is, these events carry information about something that occurred in a domain model that must be made available for other consumers.

One application of this is as a building block for systems using CQRS pattern where events produced on the write side (a result of handling a command) need to get published so the read side can consume and update their internal indexes.

Another related use case is Event Sourcing which are generally spoken of in the context of an "aggregate". The pattern requires each aggregate instance to maintain it's own stream of events acting as an internal changelog. This stream is generally "private" from other consumers and requires having a single handler to apply events in order to maintain a consistent internal state.

This library could be used for this, but the backends do not currently generalize well to 10's or 100's of thousands of streams. One strategy is "multi-plex" events from multiple aggregates on a single stream and have handlers that ignore events that are specific to the target aggregate. The basic trade-off are the number of streams (which may be limited by the backend) and the latency of reading events on a multi-plexed stream.


Package Files

conn.go doc.go encode.go stan.go

type Conn Uses

type Conn interface {
    // Publish publishes an event to the specified stream. It returns the ID of the event.
    Publish(stream string, evt *Event) (string, error)

    // Subscribe creates a subscription to the stream and associates the handler.
    Subscribe(stream string, handle Handler, opts *SubscriptionOptions) (Subscription, error)

    // Close closes the connection.
    Close() error

Conn is a connection interface to the underlying event streams backend.

func Connect Uses

func Connect(addr, cluster, client string, opts ...ConnectOption) (Conn, error)

Connect establishes a connection to the streaming backend.

type ConnectOption Uses

type ConnectOption func(o *ConnectOptions)

func WithLogger Uses

func WithLogger(l Logger) ConnectOption

type ConnectOptions Uses

type ConnectOptions struct {
    Logger Logger

func (*ConnectOptions) Apply Uses

func (o *ConnectOptions) Apply(opts ...ConnectOption)

type Data Uses

type Data interface {
    // Type returns the encoding type used to encode the data to bytes.
    Type() string

    // Decodes the underlying bytes into the passed value pointer.
    Decode(v interface{}) error

    // Encode encodes the underlying type into bytes.
    Encode() ([]byte, error)

Data encapsulates a value with a known encoding scheme.

func Bytes Uses

func Bytes(b []byte) Data

Bytes returns Data that encodes and decodes the raw bytes.

func JSON Uses

func JSON(v interface{}) Data

JSON returns Data that encodes and decodes the JSON-encodable value.

func Proto Uses

func Proto(m proto.Message) Data

Proto returns Data that encodes and decodes the proto message.

func String Uses

func String(s string) Data

String returns Data that encodes and decodes a string.

type Event Uses

type Event struct {
    // Stream is the stream this event was published on.
    Stream string `json:"stream"`

    // ID is the globally unique ID of the event.
    ID  string `json:"id"`

    // Type is the event type.
    Type string `json:"type"`

    // Time when the event was published.
    Time time.Time `json:"time"`

    // Time the event was acknowledged by the server.
    AckTime time.Time `json:"ack_time"`

    // Data is the event data.
    Data Data `json:"data"`

    // Schema is an identifier of the data schema.
    Schema string `json:"schema"`

    // Client is the ID of the client that produced this event.
    Client string `json:"client"`

    // Cause is the ID of the event that caused/resulted in this event
    // being produced.
    Cause string `json:"cause"`

    Aggregate string `json:"aggregate"`

    // Meta supports arbitrary key-value information associated with the event.
    Meta map[string]string `json:"meta,omitempty"`
    // contains filtered or unexported fields

Event is the top-level type that wraps the event data.

func (*Event) Is Uses

func (e *Event) Is(types ...string) bool

IsType returns true if the event is one of the passed types.

type Handler Uses

type Handler func(ctx context.Context, evt *Event) error

Handler is the event handler type for creating subscriptions.

type Logger Uses

type Logger interface {
    Print(v ...interface{})
    Printf(f string, v ...interface{})

Logger is a minimal interface required for internal logging. This is compatible with the stdlib log.Logger type.

type Subscription Uses

type Subscription interface {
    // Unsubscribe closes the subscription and resets the offset.
    Unsubscribe() error

    // Close closes the subscription and retains the offset.
    Close() error

type SubscriptionOptions Uses

type SubscriptionOptions struct {
    // Unique name of the subscriber. This is used to keep track of the
    // the offset of messages for a stream. This defaults to the stream name.
    Name string

    // If true, a new subscription will be send the entire backlog of events
    // in the stream. This useful for
    Backfill bool

    // If true, the stream offset will be tracked for the subscriber. Upon
    // reconnect, the next message from the offset will be received.
    Durable bool

    // If true and the subscriber had a durable subscription, this will reset the
    // durable subscription. The effect is that all events from the specified
    // start position or time will be replayed.
    Reset bool

    // If true, a new event will be processed only if/when the previous event
    // was handled successfully and acknowledged. If events should be processed
    // in order, one at a time, then this should be set to true.
    Serial bool

    // The maximum time to wait before acknowledging an event was handled.
    // If the timeout is reached, the server will redeliver the event.
    Timeout time.Duration


internal/pbPackage pb is a generated protocol buffer package.

Package eda imports 14 packages (graph). Updated 2017-12-28. Refresh now. Tools for package owners. This is an inactive package (no imports and no commits in at least two years).