Simplify serialize option

observable.go

@@ -427,12 +427,10 @@ func (o *ObservableImpl) serialize(fromCh chan Item, identifier func(interface{}
 	next := option.buildChannel()
 
 	ctx := option.buildContext()
-	mutex := sync.Mutex{}
 	minHeap := binaryheap.NewWith(func(a, b interface{}) int {
 		return a.(int) - b.(int)
 	})
-	status := make(map[int]interface{})
-	notif := make(chan struct{})
+	items := make(map[int]interface{})
 
 	var from int
 	var counter int64
@@ -449,12 +447,10 @@ func (o *ObservableImpl) serialize(fromCh chan Item, identifier func(interface{}
 				return
 			}
 			from = item.V.(int)
-			minHeap.Push(from)
 			counter = int64(from)
 
-			// Scatter
 			go func() {
-				defer close(notif)
+				defer close(next)
 
 				for {
 					select {
@@ -470,52 +466,23 @@ func (o *ObservableImpl) serialize(fromCh chan Item, identifier func(interface{}
 						}
 
 						id := identifier(item.V)
-						mutex.Lock()
-						if id != from {
-							minHeap.Push(id)
-						}
-						status[id] = item.V
-						mutex.Unlock()
-						select {
-						case <-ctx.Done():
-							return
-						case notif <- struct{}{}:
-						}
-					}
-				}
-			}()
-
-			// Gather
-			go func() {
-				defer close(next)
-
-				for {
-					select {
-					case <-ctx.Done():
-						return
-					case _, ok := <-notif:
-						if !ok {
-							return
-						}
+						minHeap.Push(id)
+						items[id] = item.V
 
-						mutex.Lock()
 						for !minHeap.Empty() {
 							v, _ := minHeap.Peek()
 							id := v.(int)
 							if atomic.LoadInt64(&counter) == int64(id) {
-								if itemValue, contains := status[id]; contains {
+								if itemValue, contains := items[id]; contains {
 									minHeap.Pop()
-									delete(status, id)
-									mutex.Unlock()
+									delete(items, id)
 									Of(itemValue).SendContext(ctx, next)
-									mutex.Lock()
-									atomic.AddInt64(&counter, 1)
+									counter++
 									continue
 								}
 							}
 							break
 						}
-						mutex.Unlock()
 					}
 				}
 			}()

observable_operator_option_test.go

@@ -73,6 +73,24 @@ func Test_Observable_Option_ContextPropagation(t *testing.T) {
 }
 
 func Test_Observable_Option_Serialize(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	idx := 1
+	<-testObservable(ctx, 1, 3, 2, 6, 4, 5).Map(func(_ context.Context, i interface{}) (interface{}, error) {
+		return i, nil
+	}, WithBufferedChannel(10), WithCPUPool(), WithContext(ctx), Serialize(func(i interface{}) int {
+		return i.(int)
+	})).DoOnNext(func(i interface{}) {
+		v := i.(int)
+		if v != idx {
+			assert.FailNow(t, "not sequential", "expected=%d, got=%d", idx, v)
+		}
+		idx++
+	})
+}
+
+func Test_Observable_Option_Serialize_Range(t *testing.T) {
 	defer goleak.VerifyNone(t)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()