Поток потребления с уникальными значениями для каждого потребления

Question

Я определил это как глобальный:

Processor<Integer, Integer> p = RingBufferProcessor.create("test", 32);
Stream<List<Integer>> s = Streams.wrap(p).distinct().buffer(5, TimeUnit.SECONDS).log().unbounded();

На строителя:

s.consume(i -> System.err.println(Thread.currentThread() + " data=" + i));

Теперь я вызываю эту функцию дважды:

for (int i = 0; i < 1000; i++) {
    p.onNext(i % 3);
}

Отличный работает отлично, и я получаю в первый раз потреблять. Когда я снова вызываю этот метод, он все еще помнит отличное и не вызывает потребление.

Есть ли возможность чистить отдельно после каждого употребления. Идея, которую мне нужно достичь, заключается в том, что я буду буферизовать все входные данные и каждый раз, когда буду использовать только уникальные элементы.

У кого-нибудь есть идеи?

Tnx

Answer 1

После глубокого поиска я пришел к выводу, что я считаю, что это не вариант. То, что мне удалось сделать, это то, что когда я потребляю, я конвертирую в Set. Таким образом, каждый раз, когда расходует поток, я работаю только с уникальными значениями.