Method Sample

observable_operator.go:1851–1920

Sample returns an Observable that emits the most recent items emitted by the source Iterable whenever the input Iterable emits an item.

(iterable Iterable, opts ...Option)

// Sample returns an Observable that emits the most recent items emitted by the source
// Iterable whenever the input Iterable emits an item.
func (o *ObservableImpl) Sample(iterable Iterable, opts ...Option) Observable {
	option := parseOptions(opts...)
	next := option.buildChannel()
	ctx := option.buildContext(o.parent)
	itCh := make(chan Item)
	obsCh := make(chan Item)

	go func() {
		defer close(obsCh)
		observe := o.Observe(opts...)
		for {
			select {
			case <-ctx.Done():
				return
			case i, ok := <-observe:
				if !ok {
					return
				}
				i.SendContext(ctx, obsCh)
			}
		}
	}()

	go func() {
		defer close(itCh)
		observe := iterable.Observe(opts...)
		for {
			select {
			case <-ctx.Done():
				return
			case i, ok := <-observe:
				if !ok {
					return
				}
				i.SendContext(ctx, itCh)
			}
		}
	}()

	go func() {
		defer close(next)
		var lastEmittedItem Item
		isItemWaitingToBeEmitted := false

		for {
			select {
			case _, ok := <-itCh:
				if ok {
					if isItemWaitingToBeEmitted {
						next <- lastEmittedItem
						isItemWaitingToBeEmitted = false
					}
				} else {
					return
				}
			case item, ok := <-obsCh:
				if ok {
					lastEmittedItem = item