4

How can I have a single-subscriber ReplaySubject that:

  1. buffers all events received with onNext() until someone subscribes to it,
  2. once someone subscribes to it, all buffered events are forwarded to the subscriber, erasing them from the ReplaySubject buffer (for performance reasons) and subsequent events are only forwarded to the subscriber (no more buffering),
  3. if more than one observer subscribes to it, an exception is thrown,
  4. all buffered events given to the subject are ordered based on the time when each event was generated, and they are forwarded to the subscriber in that order during subscription?

Also, does this make sense? I think there would be good use cases for this subject, for cases when it forwards events from, e.g., a file system...

David Chand
  • 45
  • 5
Eduardo Bezerra
  • 1,923
  • 1
  • 17
  • 32

1 Answers1

4

There is the UnicastSubject for this purpose, although it's not part of the official API; that means it can change or get removed without warning.

Edit

Let me point you to the features you requested:

1) buffers all events:

If there is no child or it hasn't caught up: store the element in the queue:

2) When subscribed to, the buffer is consumed and replayed to the Subscriber

The drain loop will poll from the queue and consume it and if the Subscriber caught up, will emit to it directly.

3) if more than one observer subscribes to it, an exception is thrown

A subsequent subscriber will get an exception

4) all buffered events given to the subject are ordered based on the time when each event was generated

Uses a single-producer single-consumer queue that guarantees FIFO order

akarnokd
  • 69,132
  • 14
  • 157
  • 192
  • Of the 4 items that I mentioned, the UnicastSubject fulfills only item 3 (exception in case of more than 1 subscriber)... Perhaps there would be an easy way to implement such a subject from scratch, with all 4 items fulfilled? – Eduardo Bezerra May 12 '16 at 22:09
  • What do you mean? Look at this line: https://github.com/ReactiveX/RxJava/blob/1.x/src/main/java/rx/internal/operators/UnicastSubject.java#L256 – akarnokd May 13 '16 at 09:47
  • Precisely... That's item number 3, of the 4 that I proposed... The unicast subject does *not* implement 1, 2, or 4, so it doesn't answer the question... – Eduardo Bezerra May 16 '16 at 16:27
  • How can you tell UnicastSubject doesn't implement all of your points? – akarnokd May 16 '16 at 19:36
  • It's not in the documentation, and if you read the source code, you won't see those features implemented there either. – Eduardo Bezerra May 16 '16 at 21:08
  • Maybe you just don't recognize the features. See my updated answer. – akarnokd May 16 '16 at 22:18
  • Oops, you're right, and I just saw that you are the biggest contributor to rx. I apologize, my good sir :) – Eduardo Bezerra May 17 '16 at 14:32