Fix: refactor observable to solve a small-probability crash

common/observable/observable.go

@@ -7,60 +7,58 @@ import (
 
 type Observable struct {
 	iterable Iterable
-	listener *sync.Map
+	listener map[Subscription]*Subscriber
+	mux      sync.Mutex
 	done     bool
-	doneLock sync.RWMutex
 }
 
 func (o *Observable) process() {
 	for item := range o.iterable {
-		o.listener.Range(func(key, value interface{}) bool {
-			elm := value.(*Subscriber)
-			elm.Emit(item)
-			return true
-		})
+		o.mux.Lock()
+		for _, sub := range o.listener {
+			sub.Emit(item)
+		}
+		o.mux.Unlock()
 	}
 	o.close()
 }
 
 func (o *Observable) close() {
-	o.doneLock.Lock()
-	o.done = true
-	o.doneLock.Unlock()
+	o.mux.Lock()
+	defer o.mux.Unlock()
 
-	o.listener.Range(func(key, value interface{}) bool {
-		elm := value.(*Subscriber)
-		elm.Close()
-		return true
-	})
+	o.done = true
+	for _, sub := range o.listener {
+		sub.Close()
+	}
 }
 
 func (o *Observable) Subscribe() (Subscription, error) {
-	o.doneLock.RLock()
-	done := o.done
-	o.doneLock.RUnlock()
-	if done == true {
+	o.mux.Lock()
+	defer o.mux.Unlock()
+	if o.done {
 		return nil, errors.New("Observable is closed")
 	}
 	subscriber := newSubscriber()
-	o.listener.Store(subscriber.Out(), subscriber)
+	o.listener[subscriber.Out()] = subscriber
 	return subscriber.Out(), nil
 }
 
 func (o *Observable) UnSubscribe(sub Subscription) {
-	elm, exist := o.listener.Load(sub)
+	o.mux.Lock()
+	defer o.mux.Unlock()
+	subscriber, exist := o.listener[sub]
 	if !exist {
 		return
 	}
-	subscriber := elm.(*Subscriber)
-	o.listener.Delete(subscriber.Out())
+	delete(o.listener, sub)
 	subscriber.Close()
 }
 
 func NewObservable(any Iterable) *Observable {
 	observable := &Observable{
 		iterable: any,
-		listener: &sync.Map{},
+		listener: map[Subscription]*Subscriber{},
 	}
 	go observable.process()
 	return observable