Method WindowWithTime

observable_operator.go:2775–2857

WindowWithTime periodically subdivides items from an Observable into Observables based on timed windows and emit them rather than emitting the items one at a time.

(timespan Duration, opts ...Option)

// WindowWithTime periodically subdivides items from an Observable into Observables based on timed windows
// and emit them rather than emitting the items one at a time.
func (o *ObservableImpl) WindowWithTime(timespan Duration, opts ...Option) Observable {
	if timespan == nil {
		return Thrown(IllegalInputError{error: "timespan must no be nil"})
	}

	f := func(ctx context.Context, next chan Item, option Option, opts ...Option) {
		observe := o.Observe(opts...)
		ch := option.buildChannel()
		done := make(chan struct{})
		empty := true
		mutex := sync.Mutex{}
		if !Of(FromChannel(ch)).SendContext(ctx, next) {
			return
		}

		go func() {
			defer func() {
				mutex.Lock()
				close(ch)
				mutex.Unlock()
			}()
			defer close(next)
			for {
				select {
				case <-ctx.Done():
					return
				case <-done:
					return
				case <-time.After(timespan.duration()):
					mutex.Lock()
					if empty {
						mutex.Unlock()
						continue
					}
					close(ch)
					empty = true
					ch = option.buildChannel()
					if !Of(FromChannel(ch)).SendContext(ctx, next) {
						close(done)
						return
					}
					mutex.Unlock()
				}
			}
		}()

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