@Test @Ignore( "Pending implementation of backpressure" ) public void multipleOperatorSingleThreadSizeException() throws InterruptedException { List<Integer> expected1List = Arrays.asList( 5, 3, 2, 0 ); Observable<Integer> expected1 = Observable.from(expected1List); List<Integer> expected2List = Arrays.asList(10, 7, 6, 4); Observable<Integer> expected2 = Observable.from(expected2List); List<Integer> expected3List = Arrays.asList(9, 8, 1); Observable<Integer> expected3 = Observable.from(expected3List); Observable<Integer> ordered = OrderedMerge.orderedMerge(new ReverseIntegerComparator(), 2, expected1, expected2, expected3); final CountDownLatch latch = new CountDownLatch( 1 ); final List<Integer> results = new ArrayList(); 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 ); results.add( integer ); } } ); latch.await(); assertEquals(0, results.size()); 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 );  }