Function newEventSourceIterable

iterable_eventsource.go:15–58

(ctx context.Context, next <-chan Item, strategy BackpressureStrategy, opts ...Option)

}

func newEventSourceIterable(ctx context.Context, next <-chan Item, strategy BackpressureStrategy, opts ...Option) Iterable {
	it := &eventSourceIterable{
		observers: make([]chan Item, 0),
		opts:      opts,
	}

	go func() {
		for {
			select {
			case <-ctx.Done():
				it.closeAllObservers()
				return
			case item, ok := <-next:
				if !ok {
					it.closeAllObservers()
					return
				}
				it.RLock()
				switch strategy {
				default:
					fallthrough
				case Block:
					for _, observer := range it.observers {
						if !item.SendContext(ctx, observer) {
							return
						}
					}
				case Drop:
					for _, observer := range it.observers {
						select {
						default:
						case <-ctx.Done():
							return
						case observer <- item:
						}
					}
				}
				it.RUnlock()
			}
		}
	}()

	return it
}

func (i *eventSourceIterable) closeAllObservers() {
	i.Lock()