@Test @Ignore("Pending implementation of backpressure") public void multipleOperatorMultipleThreadSizeException() throws InterruptedException { List<Integer> expected1List = Arrays.asList( 10, 4, 3, 2, 1 ); Observable<Integer> expected1 = Observable.from( expected1List ).subscribeOn( Schedulers.io() ); List<Integer> expected2List = Arrays.asList( 9, 8, 7 ); Observable<Integer> expected2 = Observable.from( expected2List ).subscribeOn( Schedulers.io() ); List<Integer> expected3List = Arrays.asList( 6, 5, 0 ); Observable<Integer> expected3 = Observable.from( expected3List ).subscribeOn( Schedulers.io() ); Observable<Integer> ordered = OrderedMerge.orderedMerge( new IntegerComparator(), 2, expected1, expected2, expected3 ); final CountDownLatch latch = new CountDownLatch( 1 ); final boolean[] errorThrown = new boolean[1]; ordered.subscribe( new Subscriber<Integer>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError( final Throwable e ) { logger.error("Expected error thrown", e); if ( e.getMessage().contains( "The maximum queue size of 2 has been reached" ) ) { errorThrown[0] = true; } latch.countDown(); } @Override public void onNext( final Integer integer ) { logger.info("onNext invoked with {}", integer); } } ); latch.await(); assertTrue("An exception was thrown", errorThrown[0]); }
private OrderedMerge( final Comparator<T> comparator, final int maxBufferSize, Observable<? extends T>... observables ) { this.comparator = comparator; this.maxBufferSize = maxBufferSize; this.observables = observables; }
OrderedMerge implements Observable.OnSubscribe<T> { private OrderedMerge( final Comparator<T> comparator, final int maxBufferSize, Observable<? extends T>... observables ) { this.comparator = comparator; this.maxBufferSize = maxBufferSize; this.observables = observables; } }
OrderedMerge implements Observable.OnSubscribe<T> { private OrderedMerge( final Comparator<T> comparator, final int maxBufferSize, Observable<? extends T>... observables ) { this.comparator = comparator; this.maxBufferSize = maxBufferSize; this.observables = observables; } private  OrderedMerge( final Comparator<T> comparator, final int maxBufferSize,
                          Observable<? extends T>... observables ); }
OrderedMerge implements Observable.OnSubscribe<T> { private OrderedMerge( final Comparator<T> comparator, final int maxBufferSize, Observable<? extends T>... observables ) { this.comparator = comparator; this.maxBufferSize = maxBufferSize; this.observables = observables; } private  OrderedMerge( final Comparator<T> comparator, final int maxBufferSize,
                          Observable<? extends T>... observables ); @Override void call( final Subscriber<? super T> outerOperation ); static Observable<T> orderedMerge( Comparator<T> comparator, int maxBufferSize,
                                                  Observable<? extends T>... observables ); }
OrderedMerge implements Observable.OnSubscribe<T> { private OrderedMerge( final Comparator<T> comparator, final int maxBufferSize, Observable<? extends T>... observables ) { this.comparator = comparator; this.maxBufferSize = maxBufferSize; this.observables = observables; } private  OrderedMerge( final Comparator<T> comparator, final int maxBufferSize,
                          Observable<? extends T>... observables ); @Override void call( final Subscriber<? super T> outerOperation ); static Observable<T> orderedMerge( Comparator<T> comparator, int maxBufferSize,
                                                  Observable<? extends T>... observables );  }