Last active
March 1, 2024 15:46
-
-
Save benjchristensen/e4524a308456f3c21c0b to your computer and use it in GitHub Desktop.
DebounceBuffer: Use publish(), debounce() and buffer() together to capture bursts of events.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 | |
} | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 | |
} | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
[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] |
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.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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 conditionrefCount
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. EitherConnectableObservable
with a manual, imperative invocation ofconnect
needs to be done, or the variant ofpublish(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.