feat: initial commit

pkg/bus/bus.go

@@ -0,0 +1,13 @@
+package bus
+
+import "context"
+
+type Bus interface {
+	Subscribe(ctx context.Context, ns MessageNamespace) (<-chan Message, error)
+	Unsubscribe(ctx context.Context, ns MessageNamespace, ch <-chan Message)
+	Publish(ctx context.Context, msg Message) error
+	Request(ctx context.Context, msg Message) (Message, error)
+	Reply(ctx context.Context, ns MessageNamespace, h RequestHandler) error
+}
+
+type RequestHandler func(msg Message) (Message, error)

pkg/bus/error.go

@@ -0,0 +1,9 @@
+package bus
+
+import "github.com/pkg/errors"
+
+var (
+	ErrPublishTimeout    = errors.New("publish timeout")
+	ErrUnexpectedMessage = errors.New("unexpected message")
+	ErrNoResponse        = errors.New("no response")
+)

pkg/bus/memory/bus.go

@@ -0,0 +1,91 @@
+package memory
+
+import (
+	"context"
+
+	"forge.cadoles.com/arcad/edge/pkg/bus"
+	cmap "github.com/orcaman/concurrent-map"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+type Bus struct {
+	opt           *Option
+	dispatchers   cmap.ConcurrentMap
+	nextRequestID uint64
+}
+
+func (b *Bus) Subscribe(ctx context.Context, ns bus.MessageNamespace) (<-chan bus.Message, error) {
+	logger.Debug(
+		ctx, "subscribing to messages",
+		logger.F("messageNamespace", ns),
+	)
+
+	dispatchers := b.getDispatchers(ns)
+	d := newEventDispatcher(b.opt.BufferSize)
+
+	go d.Run()
+
+	dispatchers.Add(d)
+
+	return d.Out(), nil
+}
+
+func (b *Bus) Unsubscribe(ctx context.Context, ns bus.MessageNamespace, ch <-chan bus.Message) {
+	logger.Debug(
+		ctx, "unsubscribing from messages",
+		logger.F("messageNamespace", ns),
+	)
+
+	dispatchers := b.getDispatchers(ns)
+	dispatchers.RemoveByOutChannel(ch)
+}
+
+func (b *Bus) Publish(ctx context.Context, msg bus.Message) error {
+	dispatchers := b.getDispatchers(msg.MessageNamespace())
+	dispatchersList := dispatchers.List()
+
+	logger.Debug(
+		ctx, "publishing message",
+		logger.F("dispatchers", len(dispatchersList)),
+		logger.F("messageNamespace", msg.MessageNamespace()),
+	)
+
+	for _, d := range dispatchersList {
+		if err := d.In(msg); err != nil {
+			return errors.WithStack(err)
+		}
+	}
+
+	return nil
+}
+
+func (b *Bus) getDispatchers(namespace bus.MessageNamespace) *eventDispatcherSet {
+	strNamespace := string(namespace)
+
+	rawDispatchers, exists := b.dispatchers.Get(strNamespace)
+	dispatchers, ok := rawDispatchers.(*eventDispatcherSet)
+
+	if !exists || !ok {
+		dispatchers = newEventDispatcherSet()
+		b.dispatchers.Set(strNamespace, dispatchers)
+	}
+
+	return dispatchers
+}
+
+func NewBus(funcs ...OptionFunc) *Bus {
+	opt := DefaultOption()
+
+	for _, fn := range funcs {
+		fn(opt)
+	}
+
+	return &Bus{
+		opt:         opt,
+		dispatchers: cmap.New(),
+	}
+}
+
+// Check bus implementation.
+var _ bus.Bus = NewBus()

pkg/bus/memory/bus_test.go

@@ -0,0 +1,29 @@
+package memory
+
+import (
+	"testing"
+
+	busTesting "forge.cadoles.com/arcad/edge/pkg/bus/testing"
+)
+
+func TestMemoryBus(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Test disabled when -short flag is set")
+	}
+
+	t.Parallel()
+
+	t.Run("PublishSubscribe", func(t *testing.T) {
+		t.Parallel()
+
+		b := NewBus()
+		busTesting.TestPublishSubscribe(t, b)
+	})
+
+	t.Run("RequestReply", func(t *testing.T) {
+		t.Parallel()
+
+		b := NewBus()
+		busTesting.TestRequestReply(t, b)
+	})
+}

pkg/bus/memory/event_dispatcher.go

@@ -0,0 +1,117 @@
+package memory
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"forge.cadoles.com/arcad/edge/pkg/bus"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+type eventDispatcherSet struct {
+	mutex sync.Mutex
+	items map[*eventDispatcher]struct{}
+}
+
+func (s *eventDispatcherSet) Add(d *eventDispatcher) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	s.items[d] = struct{}{}
+}
+
+func (s *eventDispatcherSet) RemoveByOutChannel(out <-chan bus.Message) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	for d := range s.items {
+		if d.IsOut(out) {
+			d.Close()
+			delete(s.items, d)
+		}
+	}
+}
+
+func (s *eventDispatcherSet) List() []*eventDispatcher {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	dispatchers := make([]*eventDispatcher, 0, len(s.items))
+
+	for d := range s.items {
+		dispatchers = append(dispatchers, d)
+	}
+
+	return dispatchers
+}
+
+func newEventDispatcherSet() *eventDispatcherSet {
+	return &eventDispatcherSet{
+		items: make(map[*eventDispatcher]struct{}),
+	}
+}
+
+type eventDispatcher struct {
+	in     chan bus.Message
+	out    chan bus.Message
+	mutex  sync.RWMutex
+	closed bool
+}
+
+func (d *eventDispatcher) Close() {
+	d.mutex.Lock()
+	defer d.mutex.Unlock()
+
+	d.closed = true
+	close(d.in)
+}
+
+func (d *eventDispatcher) In(msg bus.Message) (err error) {
+	d.mutex.RLock()
+	defer d.mutex.RUnlock()
+
+	if d.closed {
+		return
+	}
+
+	d.in <- msg
+
+	return nil
+}
+
+func (d *eventDispatcher) Out() <-chan bus.Message {
+	return d.out
+}
+
+func (d *eventDispatcher) IsOut(out <-chan bus.Message) bool {
+	return d.out == out
+}
+
+func (d *eventDispatcher) Run() {
+	ctx := context.Background()
+
+	for {
+		msg, ok := <-d.in
+		if !ok {
+			close(d.out)
+
+			return
+		}
+
+		timeout := time.After(2 * time.Second)
+		select {
+		case d.out <- msg:
+		case <-timeout:
+			logger.Error(ctx, "message out chan timed out", logger.F("message", msg))
+		}
+	}
+}
+
+func newEventDispatcher(bufferSize int64) *eventDispatcher {
+	return &eventDispatcher{
+		in:     make(chan bus.Message, bufferSize),
+		out:    make(chan bus.Message, bufferSize),
+		closed: false,
+	}
+}

pkg/bus/memory/option.go

@@ -0,0 +1,19 @@
+package memory
+
+type Option struct {
+	BufferSize int64
+}
+
+type OptionFunc func(*Option)
+
+func DefaultOption() *Option {
+	return &Option{
+		BufferSize: 16, // nolint: gomnd
+	}
+}
+
+func WithBufferSize(size int64) OptionFunc {
+	return func(o *Option) {
+		o.BufferSize = size
+	}
+}

pkg/bus/memory/request_reply.go

@@ -0,0 +1,151 @@
+package memory
+
+import (
+	"context"
+	"strconv"
+	"sync/atomic"
+
+	"forge.cadoles.com/arcad/edge/pkg/bus"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+const (
+	MessageNamespaceRequest bus.MessageNamespace = "reqrep/request"
+	MessageNamespaceReply   bus.MessageNamespace = "reqrep/reply"
+)
+
+type RequestMessage struct {
+	RequestID uint64
+
+	Message bus.Message
+
+	ns bus.MessageNamespace
+}
+
+func (m *RequestMessage) MessageNamespace() bus.MessageNamespace {
+	return m.ns
+}
+
+type ReplyMessage struct {
+	RequestID uint64
+	Message   bus.Message
+	Error     error
+
+	ns bus.MessageNamespace
+}
+
+func (m *ReplyMessage) MessageNamespace() bus.MessageNamespace {
+	return m.ns
+}
+
+func (b *Bus) Request(ctx context.Context, msg bus.Message) (bus.Message, error) {
+	requestID := atomic.AddUint64(&b.nextRequestID, 1)
+
+	req := &RequestMessage{
+		RequestID: requestID,
+		Message:   msg,
+		ns:        msg.MessageNamespace(),
+	}
+
+	replyNamespace := createReplyNamespace(requestID)
+
+	replies, err := b.Subscribe(ctx, replyNamespace)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	defer func() {
+		b.Unsubscribe(ctx, replyNamespace, replies)
+	}()
+
+	logger.Debug(ctx, "publishing request", logger.F("request", req))
+
+	if err := b.Publish(ctx, req); err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, errors.WithStack(ctx.Err())
+
+		case msg, ok := <-replies:
+			if !ok {
+				return nil, errors.WithStack(bus.ErrNoResponse)
+			}
+
+			reply, ok := msg.(*ReplyMessage)
+			if !ok {
+				return nil, errors.WithStack(bus.ErrUnexpectedMessage)
+			}
+
+			if reply.Error != nil {
+				return nil, errors.WithStack(err)
+			}
+
+			return reply.Message, nil
+		}
+	}
+}
+
+type RequestHandler func(evt bus.Message) (bus.Message, error)
+
+func (b *Bus) Reply(ctx context.Context, msgNamespace bus.MessageNamespace, h bus.RequestHandler) error {
+	requests, err := b.Subscribe(ctx, msgNamespace)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	defer func() {
+		b.Unsubscribe(ctx, msgNamespace, requests)
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return errors.WithStack(ctx.Err())
+
+		case msg, ok := <-requests:
+			if !ok {
+				return nil
+			}
+
+			request, ok := msg.(*RequestMessage)
+			if !ok {
+				return errors.WithStack(bus.ErrUnexpectedMessage)
+			}
+
+			logger.Debug(ctx, "handling request", logger.F("request", request))
+
+			msg, err := h(request.Message)
+
+			reply := &ReplyMessage{
+				RequestID: request.RequestID,
+				Message:   nil,
+				Error:     nil,
+
+				ns: createReplyNamespace(request.RequestID),
+			}
+
+			if err != nil {
+				reply.Error = errors.WithStack(err)
+			} else {
+				reply.Message = msg
+			}
+
+			logger.Debug(ctx, "publishing reply", logger.F("reply", reply))
+
+			if err := b.Publish(ctx, reply); err != nil {
+				return errors.WithStack(err)
+			}
+		}
+	}
+}
+
+func createReplyNamespace(requestID uint64) bus.MessageNamespace {
+	return bus.NewMessageNamespace(
+		MessageNamespaceReply,
+		bus.MessageNamespace(strconv.FormatUint(requestID, 10)),
+	)
+}

pkg/bus/message.go

@@ -0,0 +1,33 @@
+package bus
+
+import (
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+type (
+	MessageNamespace string
+)
+
+type Message interface {
+	MessageNamespace() MessageNamespace
+}
+
+func NewMessageNamespace(namespaces ...MessageNamespace) MessageNamespace {
+	var sb strings.Builder
+
+	for i, ns := range namespaces {
+		if i != 0 {
+			if _, err := sb.WriteString(":"); err != nil {
+				panic(errors.Wrap(err, "could not build new message namespace"))
+			}
+		}
+
+		if _, err := sb.WriteString(string(ns)); err != nil {
+			panic(errors.Wrap(err, "could not build new message namespace"))
+		}
+	}
+
+	return MessageNamespace(sb.String())
+}

pkg/bus/testing/publish_subscribe.go

@@ -0,0 +1,96 @@
+package testing
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"forge.cadoles.com/arcad/edge/pkg/bus"
+	"github.com/pkg/errors"
+)
+
+const (
+	testNamespace bus.MessageNamespace = "testNamespace"
+)
+
+type testMessage struct{}
+
+func (e *testMessage) MessageNamespace() bus.MessageNamespace {
+	return testNamespace
+}
+
+func TestPublishSubscribe(t *testing.T, b bus.Bus) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+
+	t.Log("subscribe")
+
+	messages, err := b.Subscribe(ctx, testNamespace)
+	if err != nil {
+		t.Fatal(errors.WithStack(err))
+	}
+
+	var wg sync.WaitGroup
+
+	wg.Add(5)
+
+	go func() {
+		// 5 events should be received
+		t.Log("publish 0")
+
+		if err := b.Publish(ctx, &testMessage{}); err != nil {
+			t.Error(errors.WithStack(err))
+		}
+
+		t.Log("publish 1")
+
+		if err := b.Publish(ctx, &testMessage{}); err != nil {
+			t.Error(errors.WithStack(err))
+		}
+
+		t.Log("publish 2")
+
+		if err := b.Publish(ctx, &testMessage{}); err != nil {
+			t.Error(errors.WithStack(err))
+		}
+
+		t.Log("publish 3")
+
+		if err := b.Publish(ctx, &testMessage{}); err != nil {
+			t.Error(errors.WithStack(err))
+		}
+
+		t.Log("publish 4")
+
+		if err := b.Publish(ctx, &testMessage{}); err != nil {
+			t.Error(errors.WithStack(err))
+		}
+	}()
+
+	var count int32 = 0
+
+	go func() {
+		t.Log("range for events")
+
+		for msg := range messages {
+			t.Logf("received msg %d", atomic.LoadInt32(&count))
+			atomic.AddInt32(&count, 1)
+
+			if e, g := testNamespace, msg.MessageNamespace(); e != g {
+				t.Errorf("evt.MessageNamespace(): expected '%v', got '%v'", e, g)
+			}
+
+			wg.Done()
+		}
+	}()
+
+	wg.Wait()
+
+	b.Unsubscribe(ctx, testNamespace, messages)
+
+	if e, g := int32(5), count; e != g {
+		t.Errorf("message received count: expected '%v', got '%v'", e, g)
+	}
+}

pkg/bus/testing/request_reply.go

@@ -0,0 +1,110 @@
+package testing
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"forge.cadoles.com/arcad/edge/pkg/bus"
+	"github.com/pkg/errors"
+)
+
+const (
+	testTypeReqRes bus.MessageNamespace = "testNamspaceReqRes"
+)
+
+type testReqResMessage struct {
+	i int
+}
+
+func (m *testReqResMessage) MessageNamespace() bus.MessageNamespace {
+	return testNamespace
+}
+
+func TestRequestReply(t *testing.T, b bus.Bus) {
+	expectedRoundTrips := 256
+	timeout := time.Now().Add(time.Duration(expectedRoundTrips) * time.Second)
+
+	var (
+		initWaitGroup sync.WaitGroup
+		resWaitGroup  sync.WaitGroup
+	)
+
+	initWaitGroup.Add(1)
+
+	go func() {
+		repondCtx, cancelRespond := context.WithDeadline(context.Background(), timeout)
+		defer cancelRespond()
+
+		initWaitGroup.Done()
+
+		err := b.Reply(repondCtx, testNamespace, func(msg bus.Message) (bus.Message, error) {
+			defer resWaitGroup.Done()
+
+			req, ok := msg.(*testReqResMessage)
+			if !ok {
+				return nil, errors.WithStack(bus.ErrUnexpectedMessage)
+			}
+
+			result := &testReqResMessage{req.i}
+
+			// Simulate random work
+			time.Sleep(time.Millisecond * 100)
+
+			t.Logf("[RES] sending res #%d", req.i)
+
+			return result, nil
+		})
+		if err != nil {
+			t.Error(err)
+		}
+	}()
+
+	initWaitGroup.Wait()
+
+	var reqWaitGroup sync.WaitGroup
+
+	for i := 0; i < expectedRoundTrips; i++ {
+		resWaitGroup.Add(1)
+		reqWaitGroup.Add(1)
+
+		go func(i int) {
+			defer reqWaitGroup.Done()
+
+			requestCtx, cancelRequest := context.WithDeadline(context.Background(), timeout)
+			defer cancelRequest()
+
+			req := &testReqResMessage{i}
+
+			t.Logf("[REQ] sending req #%d", i)
+
+			result, err := b.Request(requestCtx, req)
+			if err != nil {
+				t.Error(err)
+			}
+
+			t.Logf("[REQ] received req #%d reply", i)
+
+			if result == nil {
+				t.Error("result should not be nil")
+
+				return
+			}
+
+			res, ok := result.(*testReqResMessage)
+			if !ok {
+				t.Error(errors.WithStack(bus.ErrUnexpectedMessage))
+
+				return
+			}
+
+			if e, g := req.i, res.i; e != g {
+				t.Errorf("res.i: expected '%v', got '%v'", e, g)
+			}
+		}(i)
+	}
+
+	reqWaitGroup.Wait()
+	resWaitGroup.Wait()
+}