cqrs

Documentation

Index

• Constants
• Variables
• type Broker
• type Caller
• type Envelope
  • func NewDefaultEnvelope(options ...EnvelopeOption) (envelope Envelope)
  • func NewEnvelope(options ...EnvelopeOption) (envelope Envelope)
  • func (e *Envelope) Ack()
  • func (e *Envelope) Reject(err error)
  • func (e *Envelope) SourceKey() RouteKey
  • func (e *Envelope) SubjectKey() RouteKey
  • func (e *Envelope) TypeKey() RouteKey
  • func (e *Envelope) Wait() <-chan error
• type EnvelopeOption
  • func FromData(data []byte) EnvelopeOption
  • func WithData(encoding string, data interface{}) EnvelopeOption
  • func WithHeaders(header http.Header) EnvelopeOption
  • func WithID(messageID string) EnvelopeOption
  • func WithNewUUID() EnvelopeOption
  • func WithSource(source RouteKey) EnvelopeOption
  • func WithSubject(subject RouteKey) EnvelopeOption
  • func WithTime(ts time.Time) EnvelopeOption
  • func WithType(messageType RouteKey) EnvelopeOption
• type Message
  • func NewMessage() Message
• type MockBroker
  • func NewMockBroker() *MockBroker
  • func (m *MockBroker) Close() error
  • func (m *MockBroker) Publish(topic RouteKey, messages ...Message) error
  • func (m *MockBroker) Subscribe(_ context.Context, topic RouteKey, _ *time.Duration) (<-chan Envelope, error)
• type NoOpBroker
  • func (m NoOpBroker) Close() error
  • func (m NoOpBroker) Publish(_ RouteKey, _ ...Message) error
  • func (m NoOpBroker) Subscribe(_ context.Context, _ RouteKey, _ *time.Duration) (<-chan Envelope, error)
• type OnMessageErrorFunc
• type OnMessageFunc
• type Receiver
• type RetryableError
  • func NewRetryableError(err error) RetryableError
  • func (r RetryableError) Cause() error
  • func (r RetryableError) Error() string
  • func (r RetryableError) Unwrap() error
  • func (r RetryableError) Wrap(err error) error
• type RouteKey
  • func (r RouteKey) String() string
  • func (r RouteKey) StringPtr() *string
  • func (r RouteKey) With(keys ...RouteKey) RouteKey
• type Sender
• type Validator

Constants

const (
	TextPlain       = ceEvent.TextPlain
	ApplicationJSON = ceEvent.ApplicationJSON
)

Variables

var PublishTimeoutErr = errors.New("failed to publish message within timeout")

Types

type Broker

type Broker interface {
	Sender
	Receiver
}

Broker an interface that describes a system that can handle inbound and outbound Message(s)

type Caller

type Caller interface {
	Request(topic RouteKey, message Message) (Envelope, error)
}

Caller is an interface that describes a system that expects to make a synchronous request to another system

type Envelope

type Envelope struct {
	Message
	Error error

	Header http.Header
	// contains filtered or unexported fields
}

Envelope a value object (container) for Message Allows a recipient to check if a message contained an error

func NewDefaultEnvelope

func NewDefaultEnvelope(options ...EnvelopeOption) (envelope Envelope)

NewDefaultEnvelope accepts raw message bytes, attempts to unmarshall the event into a Message If there is a problem decoding the message the error will be stored on the Envelope Sets message.ID and Time

func NewEnvelope

func NewEnvelope(options ...EnvelopeOption) (envelope Envelope)

NewEnvelope accepts raw message bytes, attempts to unmarshall the event into a Message If there is a problem decoding the message the error will be stored on the Envelope

func (*Envelope) Ack

func (e *Envelope) Ack()

Ack closes the ack channel without an error

func (*Envelope) Reject

func (e *Envelope) Reject(err error)

Reject rejects the message with an error

func (*Envelope) SourceKey

func (e *Envelope) SourceKey() RouteKey

SourceKey returns the RouteKey of the event source

func (*Envelope) SubjectKey

func (e *Envelope) SubjectKey() RouteKey

SubjectKey returns the RouteKey of the event subject

func (*Envelope) TypeKey

func (e *Envelope) TypeKey() RouteKey

TypeKey returns the RouteKey of the event type

func (*Envelope) Wait

func (e *Envelope) Wait() <-chan error

Wait waits until the ack channel has been closed

type EnvelopeOption

type EnvelopeOption func(envelope *Envelope) error

EnvelopeOption a higher order function that modifies properties of a message

func FromData

func FromData(data []byte) EnvelopeOption

FromData accepts raw byte data an attempts to json.Unmarshal into the enveloped Message If an error occurs while unmarshalling it will be accessible from Envelope.Error

func WithData

func WithData(encoding string, data interface{}) EnvelopeOption

WithData encodes the given payload with the given content type. If the provided payload is a byte array, when marshalled to json it will be encoded as base64. If the provided payload is different from byte array, datacodec.Encode is invoked to attempt a marshalling to byte array.

func WithHeaders

func WithHeaders(header http.Header) EnvelopeOption

func WithID

func WithID(messageID string) EnvelopeOption

WithID sets the ID property of the enveloped Message

func WithNewUUID

func WithNewUUID() EnvelopeOption

WithNewUUID sets the ID property of the enveloped Message to a random UUIDv4

func WithSource

func WithSource(source RouteKey) EnvelopeOption

WithSource sets the source of the enveloped Message

func WithSubject

func WithSubject(subject RouteKey) EnvelopeOption

WithSubject sets the subject of the enveloped Message

func WithTime

func WithTime(ts time.Time) EnvelopeOption

WithTime sets the time of the enveloped Message

func WithType

func WithType(messageType RouteKey) EnvelopeOption

WithType sets the type property of the enveloped Message

type Message

type Message interface {
	Validator
	json.Marshaler
	json.Unmarshaler
	event.EventReader
	event.EventWriter
}

Message an interface

func NewMessage

func NewMessage() Message

NewMessage creates a new instance compatible Message

type MockBroker

type MockBroker struct {
	mock.Mock
	// contains filtered or unexported fields
}

func NewMockBroker

func NewMockBroker() *MockBroker

func (*MockBroker) Close

func (m *MockBroker) Close() error

func (*MockBroker) Publish

func (m *MockBroker) Publish(topic RouteKey, messages ...Message) error

func (*MockBroker) Subscribe

func (m *MockBroker) Subscribe(
	_ context.Context,
	topic RouteKey,
	_ *time.Duration,
) (<-chan Envelope, error)

type NoOpBroker

type NoOpBroker struct{}

func (NoOpBroker) Close

func (m NoOpBroker) Close() error

func (NoOpBroker) Publish

func (m NoOpBroker) Publish(_ RouteKey, _ ...Message) error

func (NoOpBroker) Subscribe

func (m NoOpBroker) Subscribe(_ context.Context, _ RouteKey, _ *time.Duration) (<-chan Envelope, error)

type OnMessageErrorFunc

type OnMessageErrorFunc func(ctx context.Context, msg Envelope, err error) error

OnMessageErrorFunc a function that processes errors that are returned from OnMessageFunc

type OnMessageFunc

type OnMessageFunc func(ctx context.Context, msg Envelope) error

OnMessageFunc a function that processes an Envelope message

type Receiver

type Receiver interface {
	Subscribe(ctx context.Context, topic RouteKey, ackDeadline *time.Duration) (<-chan Envelope, error)
	Close() error
}

Receiver an interface that describes a system that expects to subscribe to a topic and receive Message(s)

type RetryableError

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

RetryableError ...

func NewRetryableError

func NewRetryableError(err error) RetryableError

func (RetryableError) Cause

func (r RetryableError) Cause() error

func (RetryableError) Error

func (r RetryableError) Error() string

func (RetryableError) Unwrap

func (r RetryableError) Unwrap() error

func (RetryableError) Wrap

func (r RetryableError) Wrap(err error) error

type RouteKey

type RouteKey string

RouteKey a stringer used for constructing conventional topic, source, and event names

func (RouteKey) String

func (r RouteKey) String() string

func (RouteKey) StringPtr

func (r RouteKey) StringPtr() *string

func (RouteKey) With

func (r RouteKey) With(keys ...RouteKey) RouteKey

With creates a copy of the route key with the additional keys applied Example:

RouteKey("com.service").With("events")

type Sender

type Sender interface {
	Publish(topic RouteKey, messages ...Message) error
}

Sender is an interface that describes a system that expects to publish (n) Message(s)

type Validator

type Validator interface {
	Validate() error
}

Validator describes an interface for validating a data structure