@Test public void testDuplicateOrderingCorrectComparator() throws InterruptedException { List<Integer> expected1List = Arrays.asList( 1, 2, 3, 4, 5, 10 ); Observable<Integer> expected1 = Observable.from( expected1List ).subscribeOn( Schedulers.io() ); List<Integer> expected2List = Arrays.asList( 5, 6, 7, 8, 9 ); Observable<Integer> expected2 = Observable.from( expected2List ).subscribeOn( Schedulers.io() ); List<Integer> expected3List = Arrays.asList( 0, 1, 2, 3, 5, 6, 9 ); 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 List<Integer> results = new ArrayList(); ordered.subscribe( new Subscriber<Integer>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError( final Throwable e ) { e.printStackTrace(); fail( "An error was thrown " ); } @Override public void onNext( final Integer integer ) { logger.info( "onNext invoked with {}", integer ); results.add( integer ); } } ); latch.await(); List<Integer> expected = Arrays.asList( 0, 1, 1,2, 2, 3, 3,4, 5, 5, 5, 6, 6, 7,8, 9, 9,10 ); assertEquals( expected.size(), results.size() ); for ( int i = 0; i < expected.size(); i++ ) { assertEquals( "Same element expected", expected.get( i ), results.get( i ) ); } }
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 );  }