-
-
Save benjchristensen/e4524a308456f3c21c0b to your computer and use it in GitHub Desktop.
import java.util.List; | |
import java.util.concurrent.TimeUnit; | |
import rx.Observable; | |
import rx.Subscriber; | |
import rx.schedulers.Schedulers; | |
public class DebounceBuffer { | |
public static void main(String args[]) { | |
// debounce to the last value in each burst | |
// intermittentBursts().debounce(10, TimeUnit.MILLISECONDS).toBlocking().forEach(System.out::println); | |
/* The following will emit a buffered list as it is debounced */ | |
// first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe | |
Observable<Integer> burstStream = intermittentBursts().take(20).publish().refCount(); | |
// then we get the debounced version | |
Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS); | |
// then the buffered one that uses the debounced stream to demark window start/stop | |
Observable<List<Integer>> buffered = burstStream.buffer(debounced); | |
// then we subscribe to the buffered stream so it does what we want | |
buffered.toBlocking().forEach(System.out::println); | |
} | |
/** | |
* This is an artificial source to demonstrate an infinite stream that bursts intermittently | |
*/ | |
public static Observable<Integer> intermittentBursts() { | |
return Observable.create((Subscriber<? super Integer> s) -> { | |
while (!s.isUnsubscribed()) { | |
// burst some number of items | |
for (int i = 0; i < Math.random() * 20; i++) { | |
s.onNext(i); | |
} | |
try { | |
// sleep for a random amount of time | |
// NOTE: Only using Thread.sleep here as an artificial demo. | |
Thread.sleep((long) (Math.random() * 1000)); | |
} catch (Exception e) { | |
// do nothing | |
} | |
} | |
}).subscribeOn(Schedulers.newThread()); // use newThread since we are using sleep to block | |
} | |
} |
import java.util.List; | |
import java.util.concurrent.TimeUnit; | |
import rx.Observable; | |
import rx.Subscriber; | |
import rx.schedulers.Schedulers; | |
/** | |
* Variant of above that uses `publish(Func1<? super Observable<T>, ? extends Observable<R>> selector)` | |
* which allows multicasting without the need to use `refcount()` which can result in race conditions. | |
* */ | |
public class DebounceBufferPublish { | |
public static void main(String args[]) { | |
/* The following will emit a buffered list as it is debounced */ | |
Observable<List<Integer>> buffered = intermittentBursts().take(20).publish(stream -> { | |
// inside the `publish` function we can access `stream` in a multicasted manner | |
return stream.buffer(stream.debounce(10, TimeUnit.MILLISECONDS)); | |
}); | |
buffered.toBlocking().forEach(System.out::println); | |
} | |
/** | |
* This is an artificial source to demonstrate an infinite stream that bursts intermittently | |
*/ | |
public static Observable<Integer> intermittentBursts() { | |
return Observable.create((Subscriber<? super Integer> s) -> { | |
while (!s.isUnsubscribed()) { | |
// burst some number of items | |
for (int i = 0; i < Math.random() * 20; i++) { | |
s.onNext(i); | |
} | |
try { | |
// sleep for a random amount of time | |
// NOTE: Only using Thread.sleep here as an artificial demo. | |
Thread.sleep((long) (Math.random() * 1000)); | |
} catch (Exception e) { | |
// do nothing | |
} | |
} | |
}).subscribeOn(Schedulers.newThread()); // use newThread since we are using sleep to block | |
} | |
} |
[0] | |
[0, 1, 2, 3, 4, 5, 6, 7] | |
[0, 1, 2, 3, 4, 5, 6] | |
[0, 1, 2, 3, 4, 5, 6] | |
[0, 1, 2, 3, 4, 5] | |
[0, 1, 2, 3] | |
[0, 1, 2, 3] | |
[0, 1, 2, 3, 4] |
The race condition is when the two consumers subscribe. Often on a hot stream it doesn't matter when subscribers come and go, and refCount
is perfect for that. The race condition refCount
protects against is having only 1 active subscription upstream. However, if 2 Subscribers subscribe to a refcounted stream that emits 1, 2, 3, 4, 5, the first may get 1, 2, 3, 4, 5 and the second may get 2, 3, 4, 5.
To ensure all subscribers start at exactly the same time and get the exact same values, refCount
can not be used. Either ConnectableObservable
with a manual, imperative invocation of connect
needs to be done, or the variant of publish(function)
which connects everything within the function before connecting the upstream.
Often on the buffered debounce it won't matter if there is a race, but it's worth understanding the tradeoff, and since the publish(function)
variant is actually more elegant it is often a good choice anytime the multicasting is constrained to a set of known subscribers, as in this case.
awesome! that makes perfect sense.
Vnice
What's the benefit of this over the Observable#buffer(long timespan, long timeshift, TimeUnit unit)
overload? Seems like they achieve the same thing, but considering this gist came later I'm sure there was a reason?
@hzsweers +1
@hzsweers Are you sure? My RX skills aren't the best but I don't they are the same.
That observable has an output every timespan
regardless if there is any input. There is no debouncing behavior. Debouncing behavior means that there will be no output if there is no input, and the output only happens if after there is input and then the input stops. One operates on periodic intervals and the other operates on aperiodic intervals.
The
.publish
via the Function param is definitely a cleaner api in the second variant.I'm trying to better understand a situation where this race condition could happen in the first
DebounceBuffer
example?burstStream
is essentially shareable (courtesy: publish().refcount()).does this mean that the stream is essentially the same, however the emission times could potentially be different between
debounced
andbuffered
?This intro to rx article says that
refcount
helps "avoid" the race condition (but i could be taking that out of context with regards to this particular example).