-
Notifications
You must be signed in to change notification settings - Fork 1.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
RxJava's PublishSubject equivalent in reactor-core #79
Comments
What about the bounded equivalent : EmitterProcessor<Float> randomNumberGenerator = EmitterProcessor.create()
randomNumberGenerator.log().subscribeOn(Computations.concurrent()).publishOn(Computations.concurrent()).subscribe();
SignalEmitter<Float> emitter = randomNumberGenerator.connectEmitter();
SecureRandom sr = new SecureRandom();
int i = 1;
while(true){
try {
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
Emission emission = emitter.emit(sr.nextFloat());
} |
Also note that unlike Schedulers equivalent in RxJava, we don't have static registries of Schedulers and Computations only create a specific kind of Scheduler with a new thread pool. You should share a ref to it and leave the operators use |
Note that |
@smaldini I take note of the correct way of using Regarding the use of |
Edited indeed : |
I'm afraid I still don't see the whole picture :-( What I am missing is how to subscribe some subscribers to the On the one hand I have:
On the other:
The problem here is that This is how I did something similar in RxJava:
|
Your code in the while(true){
...
} Indeed, it never will exit! Computations.single().schedule(() -> {
while(true){
...
}
}); |
@artembilan thanks, I just didn't know where to put that loop so that it didn't block the main thread. With the following modifications both RunApp instances receive the messages but, they keep receiving prefreched values: Generator class:
Test:
As you can see in the following logs, everything that is emitted between the
|
Not sure where is a problem with your code, but for me it works as expected: @Test
public void testPublishSubscribe() throws InterruptedException {
Scheduler concurrent = Computations.concurrent();
EmitterProcessor<Long> timeGenerator = EmitterProcessor.create();
timeGenerator
.subscribeOn(concurrent)
.publishOn(concurrent)
.subscribe(v -> System.out.println("0: " + v));
SignalEmitter<Long> emitter = timeGenerator.connectEmitter();
Computations.single().schedule(() -> {
while (true) {
try {
Thread.sleep(100);
}
catch (Exception e) {
e.printStackTrace();
}
System.out.println("NEW VALUE");
emitter.emit(System.currentTimeMillis());
}
});
Thread.sleep(100);
System.out.println("WAKE UP");
timeGenerator.subscribe(v -> System.out.println("1: " + v));
timeGenerator.subscribe(v -> System.out.println("2: " + v));
Thread.sleep(100);
System.out.println("ONE MORE SUBSCRIBER");
timeGenerator.subscribe(v -> System.out.println("3: " + v));
Thread.sleep(2000);
} The logs look like:
So, I really see a new value on each emission and all new subscribers don't receive any old (cached ?) values. Plus all subscribers received the same value. |
Just added a buffer-less Processor called |
@artembilan You subscribe with a consumer that prints every value that is emitted since the moment of the subscription:
In my case I subscribe with a Subscriber:
@smaldini What got me mixed up is that I was expecting that the subscriber would get the values from the moment that it invoked |
@codependent I think that's the same problem hit by operators such as cache etc and the classic solution is usually to attach a simple Now you want to deal with individual "suspect" behavior from a subscriber perspective not the processor itself, e.g. :
Or just return generator.onBackpressureDrop() as a consuming-side |
@smaldini That's it, with the additional FluxProcessor the subscribers won't get buffered values, Thank you everyone for looking into this. Final code:
And the log sequence:
|
@codependent if you intent to drop I advise the new |
@smaldini Since the requests won't happen immediately it fails and doesn't recover from it. In the end the subscribers never get anything. Maybe I'm not using it right:
After the faifure the subscribers request five values but don't get anything:
|
You would still need |
Got it, changed from keeping the dropping processor: Works like a charm :-) |
As seen on a Stackoverflow question, apparently there is not a equivalent Publisher in reactor-core.
So far, I haven't been able to migrate this behaviour from RxJava:
This is one of the ways I tried (among others):
In this case, the subscriber
xxx
never gets notified.The text was updated successfully, but these errors were encountered: