@SuppressWarnings("rawtypes")

private Func1<Single.OnSubscribe, Single.OnSubscribe> onCreate;

@SuppressWarnings("rawtypes")

private Func2<Single, Single.OnSubscribe, Single.OnSubscribe> onStart;

private Func1<Subscription, Subscription> onReturn;

@SuppressWarnings("rawtypes")

@Before

public void setUp() throws Exception {

onCreate = spy(new Func1<Single.OnSubscribe, Single.OnSubscribe>() {

@Override

public Single.OnSubscribe call(Single.OnSubscribe t) {

return t;

}

});

RxJavaHooks.setOnSingleCreate(onCreate);

onStart = spy(new Func2<Single, Single.OnSubscribe, Single.OnSubscribe>() {

@Override

public Single.OnSubscribe call(Single t1, Single.OnSubscribe t2) {

return t2;

}

});

RxJavaHooks.setOnSingleStart(onStart);

onReturn = spy(new Func1<Subscription, Subscription>() {

@Override

public Subscription call(Subscription t) {

return t;

}

});

RxJavaHooks.setOnSingleReturn(onReturn);

}

@After

public void after() {

RxJavaHooks.reset();

}

@Test

public void testHelloWorld() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.just("Hello World!").subscribe(ts);

ts.assertReceivedOnNext(Arrays.asList("Hello World!"));

}

@Test

public void testHelloWorld2() {

final AtomicReference<String> v = new AtomicReference<String>();

Single.just("Hello World!").subscribe(new SingleSubscriber<String>() {

@Override

public void onSuccess(String value) {

v.set(value);

}

@Override

public void onError(Throwable error) {

}

});

assertEquals("Hello World!", v.get());

}

@Test

public void testMap() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.just("A")

.map(new Func1<String, String>() {

@Override

public String call(String s) {

return s + "B";

}

})

.subscribe(ts);

ts.assertReceivedOnNext(Arrays.asList("AB"));

}

@Test

public void zip2Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single.zip(a, b, new Func2<Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b) {

return "" + a + b;

}

})

.subscribe(ts);

ts.assertValue("12");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zip3Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single<Integer> c = Single.just(3);

Single.zip(a, b, c, new Func3<Integer, Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b, Integer c) {

return "" + a + b + c;

}

})

.subscribe(ts);

ts.assertValue("123");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zip4Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single<Integer> c = Single.just(3);

Single<Integer> d = Single.just(4);

Single.zip(a, b, c, d, new Func4<Integer, Integer, Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b, Integer c, Integer d) {

return "" + a + b + c + d;

}

})

.subscribe(ts);

ts.assertValue("1234");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zip5Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single<Integer> c = Single.just(3);

Single<Integer> d = Single.just(4);

Single<Integer> e = Single.just(5);

Single.zip(a, b, c, d, e, new Func5<Integer, Integer, Integer, Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b, Integer c, Integer d, Integer e) {

return "" + a + b + c + d + e;

}

})

.subscribe(ts);

ts.assertValue("12345");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zip6Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single<Integer> c = Single.just(3);

Single<Integer> d = Single.just(4);

Single<Integer> e = Single.just(5);

Single<Integer> f = Single.just(6);

Single.zip(a, b, c, d, e, f, new Func6<Integer, Integer, Integer, Integer, Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b, Integer c, Integer d, Integer e, Integer f) {

return "" + a + b + c + d + e + f;

}

})

.subscribe(ts);

ts.assertValue("123456");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zip7Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single<Integer> c = Single.just(3);

Single<Integer> d = Single.just(4);

Single<Integer> e = Single.just(5);

Single<Integer> f = Single.just(6);

Single<Integer> g = Single.just(7);

Single.zip(a, b, c, d, e, f, g, new Func7<Integer, Integer, Integer, Integer, Integer, Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b, Integer c, Integer d, Integer e, Integer f, Integer g) {

return "" + a + b + c + d + e + f + g;

}

})

.subscribe(ts);

ts.assertValue("1234567");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zip8Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single<Integer> c = Single.just(3);

Single<Integer> d = Single.just(4);

Single<Integer> e = Single.just(5);

Single<Integer> f = Single.just(6);

Single<Integer> g = Single.just(7);

Single<Integer> h = Single.just(8);

Single.zip(a, b, c, d, e, f, g, h, new Func8<Integer, Integer, Integer, Integer, Integer, Integer, Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b, Integer c, Integer d, Integer e, Integer f, Integer g, Integer h) {

return "" + a + b + c + d + e + f + g + h;

}

})

.subscribe(ts);

ts.assertValue("12345678");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zip9Singles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<Integer> a = Single.just(1);

Single<Integer> b = Single.just(2);

Single<Integer> c = Single.just(3);

Single<Integer> d = Single.just(4);

Single<Integer> e = Single.just(5);

Single<Integer> f = Single.just(6);

Single<Integer> g = Single.just(7);

Single<Integer> h = Single.just(8);

Single<Integer> i = Single.just(9);

Single.zip(a, b, c, d, e, f, g, h, i, new Func9<Integer, Integer, Integer, Integer, Integer, Integer, Integer, Integer, Integer, String>() {

@Override

public String call(Integer a, Integer b, Integer c, Integer d, Integer e, Integer f, Integer g, Integer h, Integer i) {

return "" + a + b + c + d + e + f + g + h + i;

}

})

.subscribe(ts);

ts.assertValue("123456789");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void zipIterableShouldZipListOfSingles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

@SuppressWarnings("unchecked")

Iterable<Single<Integer>> singles = Arrays.asList(Single.just(1), Single.just(2), Single.just(3));

Single

.zip(singles, new FuncN<String>() {

@Override

public String call(Object... args) {

StringBuilder stringBuilder = new StringBuilder();

for (Object arg : args) {

stringBuilder.append(arg);

}

return stringBuilder.toString();

}

}).subscribe(ts);

ts.assertValue("123");

ts.assertNoErrors();

ts.assertCompleted();

}

@Test

public void zipIterableShouldZipSetOfSingles() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Set<Single<String>> singlesSet = Collections.newSetFromMap(new LinkedHashMap<Single<String>, Boolean>(2));

Single<String> s1 = Single.just("1");

Single<String> s2 = Single.just("2");

Single<String> s3 = Single.just("3");

singlesSet.add(s1);

singlesSet.add(s2);

singlesSet.add(s3);

Single

.zip(singlesSet, new FuncN<String>() {

@Override

public String call(Object... args) {

StringBuilder stringBuilder = new StringBuilder();

for (Object arg : args) {

stringBuilder.append(arg);

}

return stringBuilder.toString();

}

}).subscribe(ts);

ts.assertValue("123");

ts.assertNoErrors();

ts.assertCompleted();

}

@Test

public void zipEmptyIterableShouldThrow() {

TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();

Iterable<Single<Object>> singles = Collections.emptyList();

Single

.zip(singles, new FuncN<Object>() {

@Override

public Object call(Object... args) {

throw new IllegalStateException("Should not be called");

}

})

.subscribe(testSubscriber);

testSubscriber.assertNoValues();

testSubscriber.assertNotCompleted();

testSubscriber.assertError(NoSuchElementException.class);

assertEquals("Can't zip 0 Singles.", testSubscriber.getOnErrorEvents().get(0).getMessage());

}

@Test

public void testZipWith() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.just("A").zipWith(Single.just("B"), new Func2<String, String, String>() {

@Override

public String call(String a, String b) {

return a + b;

}

})

.subscribe(ts);

ts.assertReceivedOnNext(Arrays.asList("AB"));

}

@Test

public void testMerge() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> a = Single.just("A");

Single<String> b = Single.just("B");

Single.merge(a, b).subscribe(ts);

ts.assertReceivedOnNext(Arrays.asList("A", "B"));

}

@Test

public void testMergeWith() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.just("A").mergeWith(Single.just("B")).subscribe(ts);

ts.assertReceivedOnNext(Arrays.asList("A", "B"));

}

@Test

public void testHookCreate() {

@SuppressWarnings("unchecked")

OnSubscribe<Object> subscriber = mock(OnSubscribe.class);

Single.create(subscriber);

verify(onCreate, times(1)).call(subscriber);

}

@Test

public void testHookSubscribeStart() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> single = Single.create(new OnSubscribe<String>() {

@Override public void call(SingleSubscriber<? super String> s) {

s.onSuccess("Hello");

}

});

single.subscribe(ts);

verify(onStart, times(1)).call(eq(single), any(Single.OnSubscribe.class));

}

@Test

public void testHookUnsafeSubscribeStart() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> single = Single.create(new OnSubscribe<String>() {

@Override public void call(SingleSubscriber<? super String> s) {

s.onSuccess("Hello");

}

});

single.unsafeSubscribe(ts);

verify(onStart, times(1)).call(eq(single), any(Single.OnSubscribe.class));

}

@Test

public void testHookSubscribeReturn() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> single = Single.create(new OnSubscribe<String>() {

@Override public void call(SingleSubscriber<? super String> s) {

s.onSuccess("Hello");

}

});

single.subscribe(ts);

verify(onReturn, times(1)).call(any(SafeSubscriber.class));

}

@Test

public void testHookUnsafeSubscribeReturn() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> single = Single.create(new OnSubscribe<String>() {

@Override public void call(SingleSubscriber<? super String> s) {

s.onSuccess("Hello");

}

});

single.unsafeSubscribe(ts);

verify(onReturn, times(1)).call(ts);

}

@Test

public void testReturnUnsubscribedWhenHookThrowsError() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> single = Single.create(new OnSubscribe<String>() {

@Override public void call(SingleSubscriber<? super String> s) {

throw new RuntimeException("Exception");

}

});

Subscription subscription = single.unsafeSubscribe(ts);

assertTrue(subscription.isUnsubscribed());

}

@Test

public void testCache() throws InterruptedException {

final AtomicInteger counter = new AtomicInteger();

Single<String> s = Single.create(new OnSubscribe<String>() {

@Override

public void call(final SingleSubscriber<? super String> observer) {

new Thread(new Runnable() {

@Override

public void run() {

counter.incrementAndGet();

observer.onSuccess("one");

}

}).start();

}

}).cache();

// we then expect the following 2 subscriptions to get that same value

final CountDownLatch latch = new CountDownLatch(2);

// subscribe once

s.subscribe(new Action1<String>() {

@Override

public void call(String v) {

assertEquals("one", v);

latch.countDown();

}

});

// subscribe again

s.subscribe(new Action1<String>() {

@Override

public void call(String v) {

assertEquals("one", v);

latch.countDown();

}

});

if (!latch.await(1000, TimeUnit.MILLISECONDS)) {

fail("subscriptions did not receive values");

}

assertEquals(1, counter.get());

}

@Test

public void testCreateSuccess() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.create(new OnSubscribe<String>() {

@Override

public void call(SingleSubscriber<? super String> s) {

s.onSuccess("Hello");

}

}).subscribe(ts);

ts.assertReceivedOnNext(Arrays.asList("Hello"));

}

@Test

public void testCreateError() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.create(new OnSubscribe<String>() {

@Override

public void call(SingleSubscriber<? super String> s) {

s.onError(new RuntimeException("fail"));

}

}).subscribe(ts);

assertEquals(1, ts.getOnErrorEvents().size());

}

@Test

public void testAsync() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.just("Hello")

.subscribeOn(Schedulers.io())

.map(new Func1<String, String>() {

@Override

public String call(String v) {

System.out.println("SubscribeOn Thread: " + Thread.currentThread());

return v;

}

})

.observeOn(Schedulers.computation())

.map(new Func1<String, String>() {

@Override

public String call(String v) {

System.out.println("ObserveOn Thread: " + Thread.currentThread());

return v;

}

})

.subscribe(ts);

ts.awaitTerminalEvent();

ts.assertReceivedOnNext(Arrays.asList("Hello"));

}

@Test

public void testFlatMap() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single.just("Hello").flatMap(new Func1<String, Single<String>>() {

@Override

public Single<String> call(String s) {

return Single.just(s + " World!").subscribeOn(Schedulers.computation());

}

}).subscribe(ts);

ts.awaitTerminalEvent();

ts.assertReceivedOnNext(Arrays.asList("Hello World!"));

}

@Test

public void testTimeout() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> s = Single.create(new OnSubscribe<String>() {

@Override

public void call(SingleSubscriber<? super String> s) {

try {

Thread.sleep(5000);

} catch (InterruptedException e) {

// ignore as we expect this for the test

}

s.onSuccess("success");

}

}).subscribeOn(Schedulers.io());

s.timeout(100, TimeUnit.MILLISECONDS).subscribe(ts);

ts.awaitTerminalEvent();

ts.assertError(TimeoutException.class);

}

@Test

public void testTimeoutWithFallback() {

TestSubscriber<String> ts = new TestSubscriber<String>();

Single<String> s = Single.create(new OnSubscribe<String>() {

@Override

public void call(SingleSubscriber<? super String> s) {

try {

Thread.sleep(5000);

} catch (InterruptedException e) {

// ignore as we expect this for the test

}

s.onSuccess("success");

}

}).subscribeOn(Schedulers.io());

s.timeout(100, TimeUnit.MILLISECONDS, Single.just("hello")).subscribe(ts);

ts.awaitTerminalEvent();

ts.assertNoErrors();

ts.assertValue("hello");

}

@Test

public void testToBlocking() {

Single<String> s = Single.just("one");

BlockingSingle<String> blocking = s.toBlocking();

assertNotNull(blocking);

assertEquals("one", blocking.value());

}

@Test

public void testUnsubscribe() throws InterruptedException {

TestSubscriber<String> ts = new TestSubscriber<String>();

final AtomicBoolean unsubscribed = new AtomicBoolean();

final AtomicBoolean interrupted = new AtomicBoolean();

final CountDownLatch latch = new CountDownLatch(2);

Single<String> s = Single.create(new OnSubscribe<String>() {

@Override

public void call(final SingleSubscriber<? super String> s) {

final Thread t = new Thread(new Runnable() {

@Override

public void run() {

try {

Thread.sleep(5000);

s.onSuccess("success");

} catch (InterruptedException e) {

interrupted.set(true);

latch.countDown();

}

}

});

s.add(Subscriptions.create(new Action0() {

@Override

public void call() {

unsubscribed.set(true);

t.interrupt();

latch.countDown();

}

}));

t.start();

}

});

s.subscribe(ts);

Thread.sleep(100);

ts.unsubscribe();

if (latch.await(1000, TimeUnit.MILLISECONDS)) {

assertTrue(unsubscribed.get());

assertTrue(interrupted.get());

} else {

fail("timed out waiting for latch");

}

}

/**

@Test

public void testUnsubscribe2() throws InterruptedException {

SingleSubscriber<String> ts = new SingleSubscriber<String>() {

@Override

public void onSuccess(String value) {

// not interested in value

}

@Override

public void onError(Throwable error) {

// not interested in value

}

};

final AtomicBoolean unsubscribed = new AtomicBoolean();

final AtomicBoolean interrupted = new AtomicBoolean();

final CountDownLatch latch = new CountDownLatch(2);

Single<String> s = Single.create(new OnSubscribe<String>() {

@Override

public void call(final SingleSubscriber<? super String> s) {

final Thread t = new Thread(new Runnable() {

@Override

public void run() {

try {

Thread.sleep(5000);

s.onSuccess("success");

} catch (InterruptedException e) {

interrupted.set(true);

latch.countDown();

}

}

});

s.add(Subscriptions.create(new Action0() {

@Override

public void call() {

unsubscribed.set(true);

t.interrupt();

latch.countDown();

}

}));

t.start();

}

});

s.subscribe(ts);

Thread.sleep(100);

ts.unsubscribe();

if (latch.await(1000, TimeUnit.MILLISECONDS)) {

assertTrue(unsubscribed.get());

assertTrue(interrupted.get());

} else {

fail("timed out waiting for latch");

}

}

/**

@Test

public void testUnsubscribeViaReturnedSubscription() throws InterruptedException {

final AtomicBoolean unsubscribed = new AtomicBoolean();

final AtomicBoolean interrupted = new AtomicBoolean();

final CountDownLatch latch = new CountDownLatch(2);

Single<String> s = Single.create(new OnSubscribe<String>() {

@Override

public void call(final SingleSubscriber<? super String> s) {

final Thread t = new Thread(new Runnable() {

@Override

public void run() {

try {

Thread.sleep(5000);

s.onSuccess("success");

} catch (InterruptedException e) {

interrupted.set(true);

latch.countDown();

}

}

});

s.add(Subscriptions.create(new Action0() {

@Override

public void call() {

unsubscribed.set(true);

t.interrupt();

latch.countDown();

}

}));

t.start();

}

});

Subscription subscription = s.subscribe();

Thread.sleep(100);

subscription.unsubscribe();

if (latch.await(1000, TimeUnit.MILLISECONDS)) {

assertTrue(unsubscribed.get());

assertTrue(interrupted.get());

} else {

fail("timed out waiting for latch");

}

}

@Test

public void testBackpressureAsObservable() {

Single<String> s = Single.create(new OnSubscribe<String>() {

@Override

public void call(SingleSubscriber<? super String> t) {

t.onSuccess("hello");

}

});

TestSubscriber<String> ts = new TestSubscriber<String>() {

@Override

public void onStart() {

request(0);

}

};

s.subscribe(ts);

ts.assertNoValues();

ts.requestMore(1);

ts.assertValue("hello");

}

@Test

public void testToObservable() {

Observable<String> a = Single.just("a").toObservable();

TestSubscriber<String> ts = TestSubscriber.create();

a.subscribe(ts);

ts.assertValue("a");

ts.assertCompleted();

}

@Test

public void toCompletableSuccess() {

Completable completable = Single.just("value").toCompletable();

TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();

completable.unsafeSubscribe(testSubscriber);

testSubscriber.assertCompleted();

testSubscriber.assertNoValues();

testSubscriber.assertNoErrors();

}

@Test

public void toCompletableError() {

TestException exception = new TestException();

Completable completable = Single.error(exception).toCompletable();

TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();

completable.unsafeSubscribe(testSubscriber);

testSubscriber.assertError(exception);

testSubscriber.assertNoValues();

testSubscriber.assertNotCompleted();

}

@Test(expected = IllegalArgumentException.class)

public void doOnErrorNull() {

Single.just(1).doOnError(null);

}

@Test

public void doOnErrorShouldNotCallActionIfNoErrorHasOccurred() {

@SuppressWarnings("unchecked")

Action1<Throwable> action = mock(Action1.class);

TestSubscriber<String> testSubscriber = new TestSubscriber<String>();

Single

.just("value")

.doOnError(action)

.subscribe(testSubscriber);

testSubscriber.assertValue("value");

testSubscriber.assertNoErrors();

verifyZeroInteractions(action);

}

@Test

public void doOnErrorShouldCallActionIfErrorHasOccurred() {

@SuppressWarnings("unchecked")

Action1<Throwable> action = mock(Action1.class);

TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();

Throwable error = new IllegalStateException();

Single

.error(error)

.doOnError(action)

.subscribe(testSubscriber);

testSubscriber.assertNoValues();

testSubscriber.assertError(error);

verify(action).call(error);

}

@Test

public void doOnErrorShouldThrowCompositeExceptionIfOnErrorActionThrows() {

@SuppressWarnings("unchecked")

Action1<Throwable> action = mock(Action1.class);

Throwable error = new RuntimeException();

Throwable exceptionFromOnErrorAction = new IllegalStateException();

doThrow(exceptionFromOnErrorAction).when(action).call(error);

TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();

Single

.error(error)

.doOnError(action)

.subscribe(testSubscriber);

testSubscriber.assertNoValues();

CompositeException compositeException = (CompositeException) testSubscriber.getOnErrorEvents().get(0);

assertEquals(2, compositeException.getExceptions().size());

assertSame(error, compositeException.getExceptions().get(0));

assertSame(exceptionFromOnErrorAction, compositeException.getExceptions().get(1));

verify(action).call(error);

}

@Test

public void shouldEmitValueFromCallable() throws Exception {

@SuppressWarnings("unchecked")

Callable<String> callable = mock(Callable.class);

when(callable.call()).thenReturn("value");

TestSubscriber<String> testSubscriber = new TestSubscriber<String>();

Single

.fromCallable(callable)

.subscribe(testSubscriber);