I have an observable that performs time consuming network operations. The client code might subscribe frequently to the observable which leads to a high network load.
Since we can't control when a subscriber appears this has to be done on the observable side.
When the number of concurrent subscribers is at a maximum you want further subscribers to receive an empty stream.
Given a source that you want to limit subscriptions for, do this:
Observable<T> limited = source.compose(
new TransformerLimitSubscribers<T>(
new AtomicInteger(), maxSubscribers))
.onErrorResumeNext(Observable.<T>empty());
...
limited.subscribe(s1);
...
limited.subscribe(s2);
where the transfomer is defined by this class:
public final class TransformerLimitSubscribers<T> implements Transformer<T, T> {
private final AtomicInteger subscriberCount;
private final int maxSubscribers;
public TransformerLimitSubscribers(AtomicInteger subscriberCount, int maxSubscribers) {
this.subscriberCount = subscriberCount;
this.maxSubscribers = maxSubscribers;
}
@Override
public Observable<T> call(Observable<T> o) {
return o.doOnSubscribe(onSubscribe()).doOnUnsubscribe(onUnsubscribe());
}
private Action0 onSubscribe() {
return new Action0() {
@Override
public void call() {
if (subscriberCount.incrementAndGet() > maxSubscribers)
throw new TooManySubscribersException();
}
};
}
private Action0 onUnsubscribe() {
return new Action0() {
@Override
public void call() {
subscriberCount.decrementAndGet();
}
};
}
public static class TooManySubscribersException extends RuntimeException {
}
}
Collected from the Internet
Please contact [email protected] to delete if infringement.
Comments