while (q.length > 0 && now - q[0].interval >= duration) {
                    observer.onNext(q.shift().value);
                }