Merge pull request #20 from donutloop/feat/bus

- Added: Implementation of bus

Author: donutloop

bus/README.md

@@ -0,0 +1,35 @@
+# Usage
+
+Bus it is a simple but powerful publish-subscribe event system. It requires object to
+register themselves with the event bus to receive events.
+
+## Example 
+```go 
+package main 
+
+import (
+	"github.com/donutloop/toolkit/bus"
+	"log"
+)
+
+type msg struct {
+	Id      int64
+	counter int
+}
+
+func main() {
+	b := bus.New()
+
+    b.AddEventListener(func(m *msg) error {
+        m.counter++
+        return nil
+    })
+
+    b.AddEventListener(func(m *msg) error {
+        m.counter++
+        return nil
+    })
+
+    b.Dispatch(new(msg))
+}
+```

bus/bus.go

@@ -0,0 +1,130 @@
+package bus
+
+import (
+	"fmt"
+	"reflect"
+	"sync"
+)
+
+//The type of the function's first and only argument
+//declares the msg to listen for.
+type HandlerFunc interface{}
+
+type Msg interface{}
+
+//It is a simple but powerful publish-subscribe event system. It requires object to
+//register themselves with the event bus to receive events.
+type Bus interface {
+	Dispatch(msg Msg) error
+	AddHandler(handler HandlerFunc) error
+	AddEventListener(handler HandlerFunc)
+	Publish(msg Msg) error
+}
+
+type InProcBus struct {
+	sync.Mutex
+	handlers  map[string]reflect.Value
+	listeners map[string][]reflect.Value
+}
+
+func New() Bus {
+	return &InProcBus{
+		handlers:  make(map[string]reflect.Value),
+		listeners: make(map[string][]reflect.Value),
+	}
+}
+
+// Dispatch sends an msg to registered handler that were declared
+// to accept values of a msg
+func (b *InProcBus) Dispatch(msg Msg) error {
+	nameOfMsg := reflect.TypeOf(msg)
+
+	handler, ok := b.handlers[nameOfMsg.String()]
+	if !ok {
+		return fmt.Errorf("handler not found for %s", nameOfMsg)
+	}
+
+	params := make([]reflect.Value, 0, 1)
+	params = append(params, reflect.ValueOf(msg))
+
+	ret := handler.Call(params)
+	v := ret[0].Interface()
+	if err, ok := v.(error); ok && err != nil {
+		return err
+	}
+	return nil
+}
+
+// Dispatch sends an msg to all registered listeners that were declared
+// to accept values of a msg
+func (b *InProcBus) Publish(msg Msg) error {
+	nameOfMsg := reflect.TypeOf(msg)
+	listeners := b.listeners[nameOfMsg.String()]
+
+	params := make([]reflect.Value, 0, 1)
+	params = append(params, reflect.ValueOf(msg))
+
+	for _, listenerHandler := range listeners {
+		ret := listenerHandler.Call(params)
+		v := ret[0].Interface()
+		if err, ok := v.(error); ok && err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// AddHandler registers a handler function that will be called when a matching
+// msg is dispatched.
+func (b *InProcBus) AddHandler(handler HandlerFunc) error {
+	b.Mutex.Lock()
+	defer b.Mutex.Unlock()
+
+	handlerType := reflect.TypeOf(handler)
+	validateHandlerFunc(handlerType)
+
+	typeOfMsg := handlerType.In(0)
+	if _, ok := b.handlers[typeOfMsg.String()]; ok {
+		return fmt.Errorf("handler exists for %s", typeOfMsg)
+	}
+
+	b.handlers[typeOfMsg.String()] = reflect.ValueOf(handler)
+	return nil
+}
+
+// AddListener registers a listener function that will be called when a matching
+// msg is dispatched.
+func (b *InProcBus) AddEventListener(handler HandlerFunc) {
+	b.Mutex.Lock()
+	defer b.Mutex.Unlock()
+
+	handlerType := reflect.TypeOf(handler)
+	validateHandlerFunc(handlerType)
+	// the first input parameter is the msg
+	typOfMsg := handlerType.In(0)
+	_, ok := b.listeners[typOfMsg.String()]
+	if !ok {
+		b.listeners[typOfMsg.String()] = make([]reflect.Value, 0)
+	}
+	b.listeners[typOfMsg.String()] = append(b.listeners[typOfMsg.String()], reflect.ValueOf(handler))
+}
+
+// panic if conditions not met (this is a programming error)
+func validateHandlerFunc(handlerType reflect.Type) {
+	switch {
+	case handlerType.Kind() != reflect.Func:
+		panic(BadFuncError("handler func must be a function"))
+	case handlerType.NumIn() != 1:
+		panic(BadFuncError("handler func must take exactly one input argument"))
+	case handlerType.NumOut() != 1:
+		panic(BadFuncError("handler func must take exactly one output argument"))
+	}
+}
+
+// BadFuncError is raised via panic() when AddEventListener is called with an
+// invalid listener function.
+type BadFuncError string
+
+func (bhf BadFuncError) Error() string {
+	return fmt.Sprintf("bad handler func: %s", string(bhf))
+}

bus/bus_test.go

@@ -0,0 +1,150 @@
+package bus_test
+
+import (
+	"errors"
+	"fmt"
+	"github.com/donutloop/toolkit/bus"
+	"testing"
+)
+
+type msg struct {
+	Id   int64
+	body string
+}
+
+func TestHandlerReturnsError(t *testing.T) {
+	b := bus.New()
+
+	b.AddHandler(func(m *msg) error {
+		return errors.New("handler error")
+	})
+
+	err := b.Dispatch(new(msg))
+	if err == nil {
+		t.Fatal("Send query failed " + err.Error())
+	}
+}
+
+func TestHandlerReturn(t *testing.T) {
+	b := bus.New()
+
+	b.AddHandler(func(q *msg) error {
+		q.body = "Hello, world!"
+		return nil
+	})
+
+	msg := new(msg)
+	err := b.Dispatch(msg)
+
+	if err != nil {
+		t.Fatal("Send msg failed " + err.Error())
+	}
+
+	if msg.body != "Hello, world!" {
+		t.Fatal("Failed to get response from handler")
+	}
+}
+
+func TestEventListeners(t *testing.T) {
+	b := bus.New()
+	count := 0
+
+	b.AddEventListener(func(m *msg) error {
+		count += 1
+		return nil
+	})
+
+	b.AddEventListener(func(m *msg) error {
+		count += 10
+		return nil
+	})
+
+	err := b.Publish(new(msg))
+	if err != nil {
+		t.Fatal("Publish msg failed " + err.Error())
+	}
+
+	if count != 11 {
+		t.Fatal(fmt.Sprintf("Publish msg failed, listeners called: %v, expected: %v", count, 11))
+	}
+}
+
+func TestAddHandlerBadFunc(t *testing.T) {
+	defer func() {
+		if v := recover(); v != nil {
+			_, ok := v.(bus.BadFuncError)
+			if !ok {
+				t.Fatalf("unexpected object (%v)", v)
+			}
+		}
+	}()
+
+	b := bus.New()
+	b.AddHandler(func(q *msg, s string) error {
+		return nil
+	})
+}
+
+func TestAddListenerBadFunc(t *testing.T) {
+	defer func() {
+		if v := recover(); v != nil {
+			_, ok := v.(bus.BadFuncError)
+			if !ok {
+				t.Fatalf("unexpected object (%v)", v)
+			}
+		}
+	}()
+
+	b := bus.New()
+	b.AddEventListener(func(q *msg, s string) error {
+		return nil
+	})
+}
+
+func BenchmarkRun(b *testing.B) {
+	for n := 0; n < b.N; n++ {
+		b := bus.New()
+
+		b.AddEventListener(func(m *msg) error {
+			return nil
+		})
+
+		b.AddEventListener(func(m *msg) error {
+			return nil
+		})
+
+		b.AddEventListener(func(m *msg) error {
+			return nil
+		})
+
+		b.AddEventListener(func(m *msg) error {
+			return nil
+		})
+
+		b.AddEventListener(func(m *msg) error {
+			return nil
+		})
+
+		b.AddHandler(func(q *msg) error {
+			return nil
+		})
+
+		b.AddHandler(func(q *msg) error {
+			return nil
+		})
+
+		b.AddHandler(func(q *msg) error {
+			return nil
+		})
+
+		b.AddHandler(func(q *msg) error {
+			return nil
+		})
+
+		b.AddHandler(func(q *msg) error {
+			return nil
+		})
+
+		b.Dispatch(new(msg))
+	}
+}