@@ -309,16 +309,17 @@ public void call(Integer t1) {
309
309
@ Test
310
310
public void testNonBlockingOuterWhileBlockingOnNext () throws InterruptedException {
311
311
312
- final CountDownLatch latch = new CountDownLatch (1 );
312
+ final CountDownLatch completedLatch = new CountDownLatch (1 );
313
+ final CountDownLatch nextLatch = new CountDownLatch (1 );
313
314
final AtomicLong completeTime = new AtomicLong ();
314
315
// use subscribeOn to make async, observeOn to move
315
- Observable .range (1 , 1000 ).subscribeOn (Schedulers .newThread ()).observeOn (Schedulers .newThread ()).subscribe (new Observer <Integer >() {
316
+ Observable .range (1 , 2 ).subscribeOn (Schedulers .newThread ()).observeOn (Schedulers .newThread ()).subscribe (new Observer <Integer >() {
316
317
317
318
@ Override
318
319
public void onCompleted () {
319
320
System .out .println ("onCompleted" );
320
321
completeTime .set (System .nanoTime ());
321
- latch .countDown ();
322
+ completedLatch .countDown ();
322
323
}
323
324
324
325
@ Override
@@ -328,20 +329,27 @@ public void onError(Throwable e) {
328
329
329
330
@ Override
330
331
public void onNext (Integer t ) {
331
-
332
+ // don't let this thing finish yet
333
+ try {
334
+ if (!nextLatch .await (1000 , TimeUnit .MILLISECONDS )) {
335
+ throw new RuntimeException ("it shouldn't have timed out" );
336
+ }
337
+ } catch (InterruptedException e ) {
338
+ throw new RuntimeException ("it shouldn't have failed" );
339
+ }
332
340
}
333
341
334
342
});
335
343
336
344
long afterSubscribeTime = System .nanoTime ();
337
- System .out .println ("After subscribe: " + latch .getCount ());
338
- assertEquals (1 , latch .getCount ());
339
- latch .await ();
345
+ System .out .println ("After subscribe: " + completedLatch .getCount ());
346
+ assertEquals (1 , completedLatch .getCount ());
347
+ nextLatch .countDown ();
348
+ completedLatch .await (1000 , TimeUnit .MILLISECONDS );
340
349
assertTrue (completeTime .get () > afterSubscribeTime );
341
350
System .out .println ("onComplete nanos after subscribe: " + (completeTime .get () - afterSubscribeTime ));
342
351
}
343
352
344
-
345
353
private static int randomIntFrom0to100 () {
346
354
// XORShift instead of Math.random http://javamex.com/tutorials/random_numbers/xorshift.shtml
347
355
long x = System .nanoTime ();
0 commit comments