create

observer/observer.go

@@ -1,7 +1,84 @@
 package observer
 
-type Observer interface {
-	Add(id string, chan interface{})
-	Remove(id string)
-	Notify(id string, arg interface{})
+import (
+	"errors"
+	"sync"
+)
+
+type observer struct {
+	mtx    sync.Mutex
+	events map[string][]chan interface{}
+}
+
+func (o *observer) Add(id string, ch chan interface{}) error {
+
+	o.mtx.Lock()
+	defer o.mtx.Unlock()
+
+	_, ok := o.events[id]
+	if ok {
+		o.events[id] = append(o.events[id], ch)
+	} else {
+		arr := make([]chan interface{}, 0)
+		arr = append(arr, ch)
+		o.events[id] = arr
+	}
+
+	return nil
+}
+func (o *observer) Remove(id string, rch chan interface{}) error {
+	o.mtx.Lock()
+	defer o.mtx.Unlock()
+
+	newArr := make([]chan interface{}, 0)
+	chans, ok := o.events[id]
+	if !ok {
+		return errors.New("event not found")
+	}
+
+	for _, ch := range chans {
+		if ch != rch {
+			newArr = append(newArr, ch)
+		} else {
+			close(ch)
+		}
+	}
+
+	o.events[id] = newArr
+	return nil
+}
+func (o *observer) Notify(id string, arg interface{}) error {
+	o.mtx.Lock()
+	defer o.mtx.Unlock()
+
+	chans, ok := o.events[id]
+	if !ok {
+		return errors.New("event not found")
+	}
+
+	for _, ch := range chans {
+		ch <- arg
+	}
+	return nil
+}
+
+var _observer *observer
+var once sync.Once
+
+func init() {
+	once.Do(func() {
+		_observer = &observer{
+			events: make(map[string][]chan interface{}),
+		}
+	})
+}
+
+func Add(id string, ch chan interface{}) error {
+	return _observer.Add(id, ch)
+}
+func Remove(id string, rch chan interface{}) error {
+	return _observer.Remove(id, rch)
+}
+func Notify(id string, arg interface{}) error {
+	return _observer.Notify(id, arg)
 }