feat: observer and event bus

Author: chenlumin

go/structural/observer/event_bus.go

@@ -7,8 +7,8 @@ import (
 )
 
 type IEventBus interface {
-	// on(eventName string, handler interface{})
-	// off(evnetName string, handler interface{})
+	// On(eventName string, handler interface{})
+	// Off(evnetName string, handler interface{})
 	// 订阅事件
 	Subscribe(eventName string, handler interface{}) error
 	// 取消订阅事件
@@ -50,6 +50,20 @@ func (b *AsyncEventBus) Subscribe(eventName string, handler interface{}) error {
 }
 
 func (b *AsyncEventBus) UnSubscribe(eventName string, handler interface{}) error {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	handlers, ok := b.handlers[eventName]
+	if !ok {
+		return fmt.Errorf("the eventName dosent's exist")
+	}
+	handlerFunc := reflect.ValueOf(handler)
+	for i, v := range handlers {
+		if v == handlerFunc {
+			handlers = append(handlers[:i], handlers[i+1:]...)
+			b.handlers[eventName] = handlers
+		}
+	}
 	return nil
 }
 
@@ -61,16 +75,15 @@ func (b *AsyncEventBus) Publish(eventName string, args ...interface{}) {
 	}
 
 	params := make([]reflect.Value, len(args))
-	for _, v := range args {
-		arg := reflect.ValueOf(v)
-		params = append(params, arg)
+	for i, v := range args {
+		params[i] = reflect.ValueOf(v)
 	}
 
-	// 	// handler := reflect.ValueOf(v)
-	// 	go v.Call(params)
-	// }
-
-	for i := range handlers {
-		go handlers[i].Call(params)
+	for _, v := range handlers {
+		go v.Call(params)
 	}
+
+	// for i := range handlers {
+	// 	go handlers[i].Call(params)
+	// }
 }

go/structural/observer/event_bus_test.go

@@ -0,0 +1,44 @@
+package observer
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func TestEventBus(t *testing.T) {
+	eb := NewAsyncEventBus()
+	eb.Subscribe("read", func(msg string) {
+		fmt.Println("read1 !", msg)
+	})
+	eb.Subscribe("read", func(msg string) {
+		fmt.Println("read2 !", msg)
+	})
+	fmt.Println(eb.handlers)
+	eb.Publish("read", "hi")
+
+	time.Sleep(1 * time.Second)
+
+}
+
+func handler1(msg string) {
+	fmt.Println("handler1", msg)
+}
+
+func TestEventBusUnscribe(t *testing.T) {
+
+	eb := NewAsyncEventBus()
+	eb.Subscribe("read", func(msg string) {
+		fmt.Println("read1 !", msg)
+	})
+	eb.Subscribe("read", handler1)
+
+	fmt.Println(eb.handlers)
+	eb.Publish("read", "hi")
+
+	time.Sleep(3 * time.Microsecond)
+	t.Log("remove handler1")
+	eb.UnSubscribe("read", handler1)
+	eb.Publish("read", "h2")
+	time.Sleep(1 * time.Second)
+}

go/structural/observer/observer.go

@@ -26,7 +26,7 @@ func (s *Subject) Register(observer IObserver) {
 func (s *Subject) Remove(observer IObserver) {
 	for i, v := range s.observers {
 		if v == observer {
-			s.observers = append(s.observers[:i], s.observers[i:]...)
+			s.observers = append(s.observers[:i], s.observers[i+1:]...)
 		}
 	}
 }

go/structural/observer/observer_test.go

@@ -11,7 +11,25 @@ func TestSubjectAndObserver(t *testing.T) {
 
 	s.Register(&o1)
 	s.Register(&o2)
+	
+
+	s.Notify("hello")
+	
+}
+
+
+func TestObserverRemove(t *testing.T) {
+	s := Subject{}
+
+	o1 := Observer1{}
+	o2 := Observer2{}
+
+
+	s.Register(&o1)
+	s.Register(&o2)
+	s.Remove(&o2)
 
 	s.Notify("hello")
 
 }
+