Handle default datasource management

pkg/bus/bus.go

@@ -10,11 +10,14 @@ type Msg interface{}
 
 type Bus interface {
 	Dispatch(msg Msg) error
+	Publish(msg Msg) error
 	AddHandler(handler HandlerFunc)
+	AddEventListener(handler HandlerFunc)
 }
 
 type InProcBus struct {
-	handlers map[string]HandlerFunc
+	handlers  map[string]HandlerFunc
+	listeners map[string][]HandlerFunc
 }
 
 // temp stuff, not sure how to handle bus instance, and init yet
@@ -23,6 +26,7 @@ var globalBus = New()
 func New() Bus {
 	bus := &InProcBus{}
 	bus.handlers = make(map[string]HandlerFunc)
+	bus.listeners = make(map[string][]HandlerFunc)
 	return bus
 }
 
@@ -46,17 +50,55 @@ func (b *InProcBus) Dispatch(msg Msg) error {
 	}
 }
 
+func (b *InProcBus) Publish(msg Msg) error {
+	var msgName = reflect.TypeOf(msg).Elem().Name()
+
+	var listeners = b.listeners[msgName]
+	if len(listeners) == 0 {
+		return nil
+	}
+
+	var params = make([]reflect.Value, 1)
+	params[0] = reflect.ValueOf(msg)
+
+	for listenerHandler := range listeners {
+		ret := reflect.ValueOf(listenerHandler).Call(params)
+		err := ret[0].Interface()
+		if err != nil {
+			return err.(error)
+		}
+	}
+
+	return nil
+}
+
 func (b *InProcBus) AddHandler(handler HandlerFunc) {
 	handlerType := reflect.TypeOf(handler)
 	queryTypeName := handlerType.In(0).Elem().Name()
 	b.handlers[queryTypeName] = handler
 }
 
+func (b *InProcBus) AddEventListener(handler HandlerFunc) {
+	handlerType := reflect.TypeOf(handler)
+	eventName := handlerType.In(0).Elem().Name()
+	list, exists := b.listeners[eventName]
+	if !exists {
+		list = make([]HandlerFunc, 0)
+		b.listeners[eventName] = list
+	}
+	list = append(list, handler)
+}
+
 // Package level functions
 func AddHandler(implName string, handler HandlerFunc) {
 	globalBus.AddHandler(handler)
 }
 
+// Package level functions
+func AddEventListener(handler HandlerFunc) {
+	globalBus.AddEventListener(handler)
+}
+
 func Dispatch(msg Msg) error {
 	return globalBus.Dispatch(msg)
 }

pkg/bus/bus_test.go

@@ -43,3 +43,26 @@ func TestQueryHandlerReturn(t *testing.T) {
 		t.Fatal("Failed to get response from handler")
 	}
 }
+
+func TestEventListeners(t *testing.T) {
+	bus := New()
+	count := 0
+
+	bus.AddEventListener(func(query *TestQuery) error {
+		count += 1
+		return nil
+	})
+
+	bus.AddEventListener(func(query *TestQuery) error {
+		count += 10
+		return nil
+	})
+
+	err := bus.Publish(&TestQuery{})
+
+	if err != nil {
+		t.Fatal("Publish event failed " + err.Error())
+	} else if count != 0 {
+		t.Fatal("Publish event failed, listeners called: %v, expected: %v", count, 11)
+	}
+}