Как реализовать одноразовое использование заказанного ReplaySubject?

Как я могу иметь одного подписчика ReplaySubject, который:

  1. буферизует все события, полученные с помощью onNext(), пока кто-нибудь не подпишется на него,
  2. как только кто-то подписывается на него, все буферизованные события перенаправляются подписчику, стирая их из буфера ReplaySubject (по соображениям производительности), а последующие события пересылаются только подписчику (буферизация больше не выполняется),
  3. если на него подписывается более одного наблюдателя, выбрасывается исключение,
  4. все буферизованные события, переданные субъекту, упорядочены в зависимости от времени, когда каждое событие было создано, и они пересылаются подписчику в этом порядке во время подписки?

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


java observable rx-java rx-scala
person Eduardo Bezerra    schedule 11.05.2016    source источник

Ответы (1)


arrow_upward
3
arrow_downward

Для этой цели существует UnicastSubject, хотя он и не является частью официального API; это означает, что он может быть изменен или удален без предупреждения.

Изменить

Позвольте мне указать вам на функции, которые вы просили:

1) буферизует все события:

Если есть нет дочернего элемента или он не догнал: сохранить элемент в очереди:

2) При подписке буфер расходуется и воспроизводится подписчику.

Цикл слива будет опрашивать очередь и использовать ее и если Подписчик догнал, будет посылать ему напрямую.

3) если на него подписывается более одного наблюдателя, выбрасывается исключение

Последующий подписчик получит исключение

4) все буферизованные события, переданные субъекту, упорядочены в зависимости от времени, когда каждое событие было сгенерировано

Использует один -producer - очередь одного потребителя, которая гарантирует порядок FIFO

person akarnokd    schedule 11.05.2016
comment
Из 4-х пунктов, которые я упомянул, UnicastSubject выполняет только пункт 3 (исключение в случае более 1 абонента)... Возможно, был бы простой способ реализовать такую ​​тему с нуля, с выполнением всех 4-х пунктов? - person Eduardo Bezerra; 13.05.2016
comment
Что ты имеешь в виду? Посмотрите на эту строку: github.com/ReactiveX/RxJava/blob/1.x/src/main/java/rx/internal/ - person akarnokd; 13.05.2016
comment
Именно... Это пункт номер 3 из 4, которые я предложил... Субъект одноадресной рассылки не реализует 1, 2 или 4, поэтому он не отвечает на вопрос... - person Eduardo Bezerra; 16.05.2016
comment
Как вы можете сказать, что UnicastSubject не реализует все ваши пункты? - person akarnokd; 16.05.2016
comment
Этого нет в документации, и если вы читаете исходный код, вы не увидите там реализованных функций. - person Eduardo Bezerra; 17.05.2016
comment
Может быть, вы просто не признаете особенности. Смотрите мой обновленный ответ. - person akarnokd; 17.05.2016
comment
К сожалению, вы правы, и я только что увидел, что вы вносите наибольший вклад в rx. Прошу прощения, добрый человек :) - person Eduardo Bezerra; 17.05.2016