SubscriptionLimiter { public static <T> Observable<T> limitSubscriptions(int maxNumber, Observable<T> source) { State state = new State(); return Observable.create(subscriber -> { synchronized (state) { if (state.getNumberOfSubscriptions() == maxNumber) { System.out.println("Subscription not allowed"); subscriber.onError(new RuntimeException("Number of subscription limited to " + maxNumber)); } else { Subscription subscription = source.subscribe(subscriber); state.add(subscription); } } }); }  static Observable<T> limitSubscriptions(int maxNumber, Observable<T> source);  }
@Test public void should_allow_one_subscription() { PublishSubject<Integer> subject = PublishSubject.create(); Observable<Integer> limitedObservable = SubscriptionLimiter.limitSubscriptions(1, subject); TestSubscriber<Integer> subscriber = new TestSubscriber<>(); limitedObservable.subscribe(subscriber); subject.onNext(123); assertThat(subscriber.getOnNextEvents()).hasSize(1).contains(123); }
@Test public void should_fail_on_second_subscription() { PublishSubject<Integer> subject = PublishSubject.create(); Observable<Integer> limitedObservable = SubscriptionLimiter.limitSubscriptions(1, subject); TestSubscriber<Integer> subscriber = new TestSubscriber<>(); TestSubscriber<Integer> subscriber2 = new TestSubscriber<>(); limitedObservable.subscribe(subscriber); limitedObservable.subscribe(subscriber2); subject.onNext(123); assertThat(subscriber2.getOnNextEvents()).isEmpty(); assertThat(subscriber2.getOnErrorEvents()).hasSize(1); }
@Test public void should_allow_a_subscription_after_an_unsubscription() { PublishSubject<Integer> subject = PublishSubject.create(); Observable<Integer> limitedObservable = SubscriptionLimiter.limitSubscriptions(1, subject); TestSubscriber<Integer> subscriber = new TestSubscriber<>(); TestSubscriber<Integer> subscriber2 = new TestSubscriber<>(); Subscription subscription = limitedObservable.subscribe(subscriber); subscription.unsubscribe(); limitedObservable.subscribe(subscriber2); subject.onNext(123); assertThat(subscriber2.getOnNextEvents()).hasSize(1).contains(123); }