events: implement package with memory & nats streams (#1942)

events/events.go

@@ -0,0 +1,46 @@
+// Package events contains interfaces for managing events within distributed systems
+package events
+
+import (
+	"encoding/json"
+	"errors"
+	"time"
+)
+
+var (
+	// ErrMissingTopic is returned if a blank topic was provided to publish
+	ErrMissingTopic = errors.New("Missing topic")
+	// ErrEncodingMessage is returned from publish if there was an error encoding the message option
+	ErrEncodingMessage = errors.New("Error encoding message")
+)
+
+// Stream of events
+type Stream interface {
+	Publish(topic string, opts ...PublishOption) error
+	Subscribe(opts ...SubscribeOption) (<-chan Event, error)
+}
+
+// Store of events
+type Store interface {
+	Read(opts ...ReadOption) ([]*Event, error)
+	Write(event *Event, opts ...WriteOption) error
+}
+
+// Event is the object returned by the broker when you subscribe to a topic
+type Event struct {
+	// ID to uniquely identify the event
+	ID string
+	// Topic of event, e.g. "registry.service.created"
+	Topic string
+	// Timestamp of the event
+	Timestamp time.Time
+	// Metadata contains the encoded event was indexed by
+	Metadata map[string]string
+	// Payload contains the encoded message
+	Payload []byte
+}
+
+// Unmarshal the events message into an object
+func (e *Event) Unmarshal(v interface{}) error {
+	return json.Unmarshal(e.Payload, v)
+}

events/memory/memory.go

@@ -0,0 +1,181 @@
+package memory
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/micro/go-micro/v3/events"
+	"github.com/micro/go-micro/v3/logger"
+	"github.com/micro/go-micro/v3/store"
+	"github.com/micro/go-micro/v3/store/memory"
+	"github.com/pkg/errors"
+)
+
+// NewStream returns an initialized memory stream
+func NewStream(opts ...Option) (events.Stream, error) {
+	// parse the options
+	var options Options
+	for _, o := range opts {
+		o(&options)
+	}
+	if options.Store == nil {
+		options.Store = memory.NewStore()
+	}
+
+	return &mem{store: options.Store}, nil
+}
+
+type subscriber struct {
+	Queue   string
+	Topic   string
+	Channel chan events.Event
+}
+
+type mem struct {
+	store store.Store
+
+	subs []*subscriber
+	sync.RWMutex
+}
+
+func (m *mem) Publish(topic string, opts ...events.PublishOption) error {
+	// validate the topic
+	if len(topic) == 0 {
+		return events.ErrMissingTopic
+	}
+
+	// parse the options
+	options := events.PublishOptions{
+		Timestamp: time.Now(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// encode the message if it's not already encoded
+	var payload []byte
+	if p, ok := options.Payload.([]byte); ok {
+		payload = p
+	} else {
+		p, err := json.Marshal(options.Payload)
+		if err != nil {
+			return events.ErrEncodingMessage
+		}
+		payload = p
+	}
+
+	// construct the event
+	event := &events.Event{
+		ID:        uuid.New().String(),
+		Topic:     topic,
+		Timestamp: options.Timestamp,
+		Metadata:  options.Metadata,
+		Payload:   payload,
+	}
+
+	// serialize the event to bytes
+	bytes, err := json.Marshal(event)
+	if err != nil {
+		return errors.Wrap(err, "Error encoding event")
+	}
+
+	// write to the store
+	key := fmt.Sprintf("%v/%v", event.Topic, event.ID)
+	if err := m.store.Write(&store.Record{Key: key, Value: bytes}); err != nil {
+		return errors.Wrap(err, "Error writing event to store")
+	}
+
+	// send to the subscribers async
+	go m.handleEvent(event)
+
+	return nil
+}
+
+func (m *mem) Subscribe(opts ...events.SubscribeOption) (<-chan events.Event, error) {
+	// parse the options
+	options := events.SubscribeOptions{
+		Queue: uuid.New().String(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// setup the subscriber
+	sub := &subscriber{
+		Channel: make(chan events.Event),
+		Topic:   options.Topic,
+		Queue:   options.Queue,
+	}
+
+	// register the subscriber
+	m.Lock()
+	m.subs = append(m.subs, sub)
+	m.Unlock()
+
+	// lookup previous events if the start time option was passed
+	if options.StartAtTime.Unix() > 0 {
+		go m.lookupPreviousEvents(sub, options.StartAtTime)
+	}
+
+	// return the channel
+	return sub.Channel, nil
+}
+
+// lookupPreviousEvents finds events for a subscriber which occured before a given time and sends
+// them into the subscribers channel
+func (m *mem) lookupPreviousEvents(sub *subscriber, startTime time.Time) {
+	var prefix string
+	if len(sub.Topic) > 0 {
+		prefix = sub.Topic + "/"
+	}
+
+	// lookup all events which match the topic (a blank topic will return all results)
+	recs, err := m.store.Read(prefix, store.ReadPrefix())
+	if err != nil && logger.V(logger.ErrorLevel, logger.DefaultLogger) {
+		logger.Errorf("Error looking up previous events: %v", err)
+		return
+	} else if err != nil {
+		return
+	}
+
+	// loop through the records and send it to the channel if it matches
+	for _, r := range recs {
+		var ev events.Event
+		if err := json.Unmarshal(r.Value, &ev); err != nil {
+			continue
+		}
+		if ev.Timestamp.Unix() < startTime.Unix() {
+			continue
+		}
+
+		sub.Channel <- ev
+	}
+}
+
+// handleEvents sends the event to any registered subscribers.
+func (m *mem) handleEvent(ev *events.Event) {
+	m.RLock()
+	subs := m.subs
+	m.RUnlock()
+
+	// filteredSubs is a KV map of the queue name and subscribers. This is used to prevent a message
+	// being sent to two subscribers with the same queue.
+	filteredSubs := map[string]*subscriber{}
+
+	// filter down to subscribers who are interested in this topic
+	for _, sub := range subs {
+		if len(sub.Topic) == 0 || sub.Topic == ev.Topic {
+			filteredSubs[sub.Queue] = sub
+		}
+	}
+
+	// send the message to each channel async (since one channel might be blocked)
+	for _, sub := range subs {
+		go func(s *subscriber) {
+			s.Channel <- *ev
+		}(sub)
+	}
+}

events/memory/memory_test.go

@@ -0,0 +1,185 @@
+package memory
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/micro/go-micro/v3/events"
+	"github.com/stretchr/testify/assert"
+)
+
+type testPayload struct {
+	Message string
+}
+
+func TestStream(t *testing.T) {
+	stream, err := NewStream()
+	assert.Nilf(t, err, "NewStream should not return an error")
+	assert.NotNilf(t, stream, "NewStream should return a stream object")
+
+	// TestMissingTopic will test the topic validation on publish
+	t.Run("TestMissingTopic", func(t *testing.T) {
+		err := stream.Publish("")
+		assert.Equalf(t, err, events.ErrMissingTopic, "Publishing to a blank topic should return an error")
+	})
+
+	// TestFirehose will publish a message to the test topic. The subscriber will subscribe to the
+	// firehose topic (indicated by a lack of the topic option).
+	t.Run("TestFirehose", func(t *testing.T) {
+		payload := &testPayload{Message: "HelloWorld"}
+		metadata := map[string]string{"foo": "bar"}
+
+		// create the subscriber
+		evChan, err := stream.Subscribe()
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		// setup the subscriber async
+		var wg sync.WaitGroup
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		err = stream.Publish("test",
+			events.WithPayload(payload),
+			events.WithMetadata(metadata),
+		)
+		assert.Nil(t, err, "Publishing a valid message should not return an error")
+		wg.Add(1)
+
+		// wait for the subscriber to recieve the message or timeout
+		wg.Wait()
+	})
+
+	// TestSubscribeTopic will publish a message to the test topic. The subscriber will subscribe to the
+	// same test topic.
+	t.Run("TestSubscribeTopic", func(t *testing.T) {
+		payload := &testPayload{Message: "HelloWorld"}
+		metadata := map[string]string{"foo": "bar"}
+
+		// create the subscriber
+		evChan, err := stream.Subscribe(events.WithTopic("test"))
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		// setup the subscriber async
+		var wg sync.WaitGroup
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		err = stream.Publish("test",
+			events.WithPayload(payload),
+			events.WithMetadata(metadata),
+		)
+		assert.Nil(t, err, "Publishing a valid message should not return an error")
+		wg.Add(1)
+
+		// wait for the subscriber to recieve the message or timeout
+		wg.Wait()
+	})
+
+	// TestSubscribeQueue will publish a message to a random topic. Two subscribers will then consume
+	// the message from the firehose topic with different queues. The second subscriber will be registered
+	// after the message is published to test durability.
+	t.Run("TestSubscribeQueue", func(t *testing.T) {
+		topic := uuid.New().String()
+		payload := &testPayload{Message: "HelloWorld"}
+		metadata := map[string]string{"foo": "bar"}
+
+		// create the first subscriber
+		evChan1, err := stream.Subscribe(events.WithTopic(topic))
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		// setup the subscriber async
+		var wg sync.WaitGroup
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan1:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		err = stream.Publish(topic,
+			events.WithPayload(payload),
+			events.WithMetadata(metadata),
+		)
+		assert.Nil(t, err, "Publishing a valid message should not return an error")
+		wg.Add(2)
+
+		// create the second subscriber
+		evChan2, err := stream.Subscribe(
+			events.WithTopic(topic),
+			events.WithQueue("second_queue"),
+			events.WithStartAtTime(time.Now().Add(time.Minute*-1)),
+		)
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan2:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		// wait for the subscriber to recieve the message or timeout
+		wg.Wait()
+	})
+}

events/memory/options.go

@@ -0,0 +1,18 @@
+package memory
+
+import "github.com/micro/go-micro/v3/store"
+
+// Options which are used to configure the in-memory stream
+type Options struct {
+	Store store.Store
+}
+
+// Option is a function which configures options
+type Option func(o *Options)
+
+// Store sets the store to use
+func Store(s store.Store) Option {
+	return func(o *Options) {
+		o.Store = s
+	}
+}

events/nats/nats.go

@@ -0,0 +1,154 @@
+package nats
+
+import (
+	"encoding/json"
+	"time"
+
+	"github.com/google/uuid"
+	stan "github.com/nats-io/stan.go"
+	"github.com/pkg/errors"
+
+	"github.com/micro/go-micro/v3/events"
+	"github.com/micro/go-micro/v3/logger"
+)
+
+const (
+	defaultClusterID = "micro"
+	eventsTopic      = "events"
+)
+
+// NewStream returns an initialized nats stream or an error if the connection to the nats
+// server could not be established
+func NewStream(opts ...Option) (events.Stream, error) {
+	// parse the options
+	options := Options{
+		ClientID:  uuid.New().String(),
+		ClusterID: defaultClusterID,
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// pass the address as an option if it was set
+	var cOpts []stan.Option
+	if len(options.Address) > 0 {
+		cOpts = append(cOpts, stan.NatsURL(options.Address))
+	}
+
+	// connect to the cluster
+	conn, err := stan.Connect(options.ClusterID, options.ClientID, cOpts...)
+	if err != nil {
+		return nil, errors.Wrap(err, "Error connecting to nats")
+	}
+
+	return &stream{conn}, nil
+}
+
+type stream struct {
+	conn stan.Conn
+}
+
+// Publish a message to a topic
+func (s *stream) Publish(topic string, opts ...events.PublishOption) error {
+	// validate the topic
+	if len(topic) == 0 {
+		return events.ErrMissingTopic
+	}
+
+	// parse the options
+	options := events.PublishOptions{
+		Timestamp: time.Now(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// encode the message if it's not already encoded
+	var payload []byte
+	if p, ok := options.Payload.([]byte); ok {
+		payload = p
+	} else {
+		p, err := json.Marshal(options.Payload)
+		if err != nil {
+			return events.ErrEncodingMessage
+		}
+		payload = p
+	}
+
+	// construct the event
+	event := &events.Event{
+		ID:        uuid.New().String(),
+		Topic:     topic,
+		Timestamp: options.Timestamp,
+		Metadata:  options.Metadata,
+		Payload:   payload,
+	}
+
+	// serialize the event to bytes
+	bytes, err := json.Marshal(event)
+	if err != nil {
+		return errors.Wrap(err, "Error encoding event")
+	}
+
+	// publish the event to the events channel
+	if _, err := s.conn.PublishAsync(eventsTopic, bytes, nil); err != nil {
+		return errors.Wrap(err, "Error publishing message to events")
+	}
+
+	// publish the event to the topic's channel
+	if _, err := s.conn.PublishAsync(event.Topic, bytes, nil); err != nil {
+		return errors.Wrap(err, "Error publishing message to topic")
+	}
+
+	return nil
+}
+
+// Subscribe to a topic
+func (s *stream) Subscribe(opts ...events.SubscribeOption) (<-chan events.Event, error) {
+	// parse the options
+	options := events.SubscribeOptions{
+		Topic: eventsTopic,
+		Queue: uuid.New().String(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// setup the subscriber
+	c := make(chan events.Event)
+	handleMsg := func(m *stan.Msg) {
+		// decode the message
+		var evt events.Event
+		if err := json.Unmarshal(m.Data, &evt); err != nil {
+			if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
+				logger.Errorf("Error decoding message: %v", err)
+			}
+			// not ackknowledging the message is the way to indicate an error occured
+			return
+		}
+
+		// push onto the channel and wait for the consumer to take the event off before we acknowledge it.
+		c <- evt
+
+		if err := m.Ack(); err != nil && logger.V(logger.ErrorLevel, logger.DefaultLogger) {
+			logger.Errorf("Error acknowledging message: %v", err)
+		}
+	}
+
+	// setup the options
+	subOpts := []stan.SubscriptionOption{
+		stan.DurableName(options.Topic),
+		stan.SetManualAckMode(),
+	}
+	if options.StartAtTime.Unix() > 0 {
+		stan.StartAtTime(options.StartAtTime)
+	}
+
+	// connect the subscriber
+	_, err := s.conn.QueueSubscribe(options.Topic, options.Queue, handleMsg, subOpts...)
+	if err != nil {
+		return nil, errors.Wrap(err, "Error subscribing to topic")
+	}
+
+	return c, nil
+}

events/nats/nats_test.go

@@ -0,0 +1,200 @@
+package nats
+
+import (
+	"net"
+	"os/exec"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/micro/go-micro/v3/events"
+	"github.com/stretchr/testify/assert"
+)
+
+type testPayload struct {
+	Message string
+}
+
+func TestStream(t *testing.T) {
+	_, err := exec.LookPath("nats-streaming-server")
+	if err != nil {
+		t.Skipf("Skipping nats test, nats-streaming-server binary is not detected")
+	}
+
+	conn, err := net.DialTimeout("tcp", ":4222", time.Millisecond*100)
+	if err != nil {
+		t.Skipf("Skipping nats test, could not connect to cluster on port 4222: %v", err)
+	}
+	if err := conn.Close(); err != nil {
+		t.Fatalf("Error closing test tcp connection to nats cluster")
+	}
+
+	stream, err := NewStream(ClusterID("test-cluster"))
+	assert.Nilf(t, err, "NewStream should not return an error")
+	assert.NotNilf(t, stream, "NewStream should return a stream object")
+
+	// TestMissingTopic will test the topic validation on publish
+	t.Run("TestMissingTopic", func(t *testing.T) {
+		err := stream.Publish("")
+		assert.Equalf(t, err, events.ErrMissingTopic, "Publishing to a blank topic should return an error")
+	})
+
+	// TestFirehose will publish a message to the test topic. The subscriber will subscribe to the
+	// firehose topic (indicated by a lack of the topic option).
+	t.Run("TestFirehose", func(t *testing.T) {
+		payload := &testPayload{Message: "HelloWorld"}
+		metadata := map[string]string{"foo": "bar"}
+
+		// create the subscriber
+		evChan, err := stream.Subscribe()
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		// setup the subscriber async
+		var wg sync.WaitGroup
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		err = stream.Publish("test",
+			events.WithPayload(payload),
+			events.WithMetadata(metadata),
+		)
+		assert.Nil(t, err, "Publishing a valid message should not return an error")
+		wg.Add(1)
+
+		// wait for the subscriber to recieve the message or timeout
+		wg.Wait()
+	})
+
+	// TestSubscribeTopic will publish a message to the test topic. The subscriber will subscribe to the
+	// same test topic.
+	t.Run("TestSubscribeTopic", func(t *testing.T) {
+		payload := &testPayload{Message: "HelloWorld"}
+		metadata := map[string]string{"foo": "bar"}
+
+		// create the subscriber
+		evChan, err := stream.Subscribe(events.WithTopic("test"))
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		// setup the subscriber async
+		var wg sync.WaitGroup
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		err = stream.Publish("test",
+			events.WithPayload(payload),
+			events.WithMetadata(metadata),
+		)
+		assert.Nil(t, err, "Publishing a valid message should not return an error")
+		wg.Add(1)
+
+		// wait for the subscriber to recieve the message or timeout
+		wg.Wait()
+	})
+
+	// TestSubscribeQueue will publish a message to a random topic. Two subscribers will then consume
+	// the message from the firehose topic with different queues. The second subscriber will be registered
+	// after the message is published to test durability.
+	t.Run("TestSubscribeQueue", func(t *testing.T) {
+		topic := uuid.New().String()
+		payload := &testPayload{Message: "HelloWorld"}
+		metadata := map[string]string{"foo": "bar"}
+
+		// create the first subscriber
+		evChan1, err := stream.Subscribe(events.WithTopic(topic))
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		// setup the subscriber async
+		var wg sync.WaitGroup
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan1:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		err = stream.Publish(topic,
+			events.WithPayload(payload),
+			events.WithMetadata(metadata),
+		)
+		assert.Nil(t, err, "Publishing a valid message should not return an error")
+		wg.Add(2)
+
+		// create the second subscriber
+		evChan2, err := stream.Subscribe(
+			events.WithTopic(topic),
+			events.WithQueue("second_queue"),
+			events.WithStartAtTime(time.Now().Add(time.Minute*-1)),
+		)
+		assert.Nilf(t, err, "Subscribe should not return an error")
+
+		go func() {
+			timeout := time.NewTimer(time.Millisecond * 250)
+
+			select {
+			case event, _ := <-evChan2:
+				assert.NotNilf(t, event, "The message was nil")
+				assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
+
+				var result testPayload
+				err = event.Unmarshal(&result)
+				assert.Nil(t, err, "Error decoding result")
+				assert.Equal(t, result, *payload, "Payload didn't match")
+
+				wg.Done()
+			case <-timeout.C:
+				t.Fatalf("Event was not recieved")
+			}
+		}()
+
+		// wait for the subscriber to recieve the message or timeout
+		wg.Wait()
+	})
+}

events/nats/options.go

@@ -0,0 +1,32 @@
+package nats
+
+// Options which are used to configure the nats stream
+type Options struct {
+	ClusterID string
+	ClientID  string
+	Address   string
+}
+
+// Option is a function which configures options
+type Option func(o *Options)
+
+// ClusterID sets the cluster id for the nats connection
+func ClusterID(id string) Option {
+	return func(o *Options) {
+		o.ClusterID = id
+	}
+}
+
+// ClientID sets the client id for the nats connection
+func ClientID(id string) Option {
+	return func(o *Options) {
+		o.ClientID = id
+	}
+}
+
+// Address of the nats cluster
+func Address(addr string) Option {
+	return func(o *Options) {
+		o.Address = addr
+	}
+}

events/options.go

@@ -0,0 +1,140 @@
+package events
+
+import "time"
+
+// PublishOptions contains all the options which can be provided when publishing an event
+type PublishOptions struct {
+	// Metadata contains any keys which can be used to query the data, for example a customer id
+	Metadata map[string]string
+	// Payload contains any additonal data which is relevent to the event but does not need to be
+	// indexed such as structured data
+	Payload interface{}
+	// Timestamp to set for the event, if the timestamp is a zero value, the current time will be used
+	Timestamp time.Time
+}
+
+// PublishOption sets attributes on PublishOptions
+type PublishOption func(o *PublishOptions)
+
+// WithMetadata sets the Metadata field on PublishOptions
+func WithMetadata(md map[string]string) PublishOption {
+	return func(o *PublishOptions) {
+		o.Metadata = md
+	}
+}
+
+// WithPayload sets the payload field on PublishOptions
+func WithPayload(p interface{}) PublishOption {
+	return func(o *PublishOptions) {
+		o.Payload = p
+	}
+}
+
+// WithTimestamp sets the timestamp field on PublishOptions
+func WithTimestamp(t time.Time) PublishOption {
+	return func(o *PublishOptions) {
+		o.Timestamp = t
+	}
+}
+
+// SubscribeOptions contains all the options which can be provided when subscribing to a topic
+type SubscribeOptions struct {
+	// Queue is the name of the subscribers queue, if two subscribers have the same queue the message
+	// should only be published to one of them
+	Queue string
+	// Topic to subscribe to, if left blank the consumer will be subscribed to the firehouse topic which
+	// recieves all events
+	Topic string
+	// StartAtTime is the time from which the messages should be consumed from. If not provided then
+	// the messages will be consumed starting from the moment the Subscription starts.
+	StartAtTime time.Time
+}
+
+// SubscribeOption sets attributes on SubscribeOptions
+type SubscribeOption func(o *SubscribeOptions)
+
+// WithQueue sets the Queue fielf on SubscribeOptions to the value provided
+func WithQueue(q string) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.Queue = q
+	}
+}
+
+// WithTopic sets the topic to subscribe to
+func WithTopic(t string) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.Topic = t
+	}
+}
+
+// WithStartAtTime sets the StartAtTime field on SubscribeOptions to the value provided
+func WithStartAtTime(t time.Time) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.StartAtTime = t
+	}
+}
+
+// WriteOptions contains all the options which can be provided when writing an event to a store
+type WriteOptions struct {
+	// TTL is the duration the event should be recorded for, a zero value TTL indicates the event should
+	// be stored indefinately
+	TTL time.Duration
+}
+
+// WriteOption sets attributes on WriteOptions
+type WriteOption func(o *WriteOptions)
+
+// WithTTL sets the TTL attribute on WriteOptions
+func WithTTL(d time.Duration) WriteOption {
+	return func(o *WriteOptions) {
+		o.TTL = d
+	}
+}
+
+// ReadOptions contains all the options which can be provided when reading events from a store
+type ReadOptions struct {
+	// Topic to read events from, if no topic is provided events from all topics will be returned
+	Topic string
+	// Query to filter the results using. The store will query the metadata provided when the event
+	// was written to the store
+	Query map[string]string
+	// Limit the number of results to return
+	Limit int
+	// Offset the results by this number, useful for paginated queries
+	Offset int
+}
+
+// ReadOption sets attributes on ReadOptions
+type ReadOption func(o *ReadOptions)
+
+// ReadTopic sets the topic attribute on ReadOptions
+func ReadTopic(t string) ReadOption {
+	return func(o *ReadOptions) {
+		o.Topic = t
+	}
+}
+
+// ReadFilter sets a key and value in the query
+func ReadFilter(key, value string) ReadOption {
+	return func(o *ReadOptions) {
+		if o.Query == nil {
+			o.Query = map[string]string{key: value}
+		} else {
+			o.Query[key] = value
+		}
+	}
+}
+
+// ReadLimit sets the limit attribute on ReadOptions
+func ReadLimit(l int) ReadOption {
+	return func(o *ReadOptions) {
+		o.Limit = 1
+	}
+}
+
+// ReadOffset sets the offset attribute on ReadOptions
+func ReadOffset(l int) ReadOption {
+	return func(o *ReadOptions) {
+		o.Offset = 1
+	}
+}