Method Serialize

observable_operator.go:2079–2132

Serialize forces an Observable to make serialized calls and to be well-behaved.

(from int, identifier func(interface{}) int, opts ...Option)

// Serialize forces an Observable to make serialized calls and to be well-behaved.
func (o *ObservableImpl) Serialize(from int, identifier func(interface{}) int, opts ...Option) Observable {
	option := parseOptions(opts...)
	next := option.buildChannel()

	ctx := option.buildContext(o.parent)
	minHeap := binaryheap.NewWith(func(a, b interface{}) int {
		return a.(int) - b.(int)
	})
	counter := int64(from)
	items := make(map[int]interface{})

	go func() {
		src := o.Observe(opts...)
		defer close(next)

		for {
			select {
			case <-ctx.Done():
				return
			case item, ok := <-src:
				if !ok {
					return
				}
				if item.Error() {
					next <- item
					return
				}

				id := identifier(item.V)
				minHeap.Push(id)
				items[id] = item.V

				for !minHeap.Empty() {
					v, _ := minHeap.Peek()
					id := v.(int)
					if atomic.LoadInt64(&counter) == int64(id) {
						if itemValue, contains := items[id]; contains {
							minHeap.Pop()
							delete(items, id)
							Of(itemValue).SendContext(ctx, next)
							counter++
							continue
						}
					}
					break
				}
			}
		}
	}()

	return &ObservableImpl{
		iterable: newChannelIterable(next),
	}
}

// Skip suppresses the first n items in the original Observable and
// returns a new Observable with the rest items.