Method serialize

observable.go:423–493

(parent context.Context, fromCh chan Item, identifier func(interface{}) int, opts ...Option)

}

func (o *ObservableImpl) serialize(parent context.Context, fromCh chan Item, identifier func(interface{}) int, opts ...Option) Observable {
	option := parseOptions(opts...)
	next := option.buildChannel()

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

	var from int
	var counter int64
	src := o.Observe(opts...)
	go func() {
		select {
		case <-ctx.Done():
			close(next)
			return
		case item := <-fromCh:
			if item.Error() {
				item.SendContext(ctx, next)
				close(next)
				return
			}
			from = item.V.(int)
			counter = int64(from)

			go func() {
				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
								}