Java 类io.reactivex.observers.DefaultObserver 实例源码
项目：Demos
文件：ExpressViewModel.java
/**
* 获取快递信息
*
* @param type 快递类型
* @param postid 快递单号
*/
public void getExpressInfo(String type, String postid) {
isShowLoading.set(true);
dataManager.getExpressInfo(type, postid)
.subscribeOn(Schedulers.io()) // 在子线程中进行Http访问
.observeOn(AndroidSchedulers.mainThread()) // UI线程处理返回接口
.compose(getProvider().<ExpressInfo>bindUntilEvent(ActivityEvent.DESTROY)) // onDestroy取消订阅
.subscribe(new DefaultObserver<ExpressInfo>() { // 订阅
@Override
public void onNext(@NonNull ExpressInfo expressInfo) {
ExpressViewModel.this.expressInfo.setExpressInfo(expressInfo);
}
@Override
public void onError(@NonNull Throwable e) {
errorMessage.set(e.getMessage());
isShowLoading.set(false);
}
@Override
public void onComplete() {
isShowLoading.set(false);
}
});
}
项目：Demos
文件：ExpressPresenter.java
/**
* 获取快递信息
*
* @param type 快递类型
* @param postid 快递单号
*/
public void getExpressInfo(String type, String postid) {
expressView.showProgressDialog();
dataManager.getExpressInfo(type, postid)
.subscribeOn(Schedulers.io()) // 在子线程中进行Http访问
.observeOn(AndroidSchedulers.mainThread()) // UI线程处理返回接口
.compose(getProvider().<ExpressInfo>bindUntilEvent(ActivityEvent.DESTROY)) // onDestroy取消订阅
.subscribe(new DefaultObserver<ExpressInfo>() { // 订阅
@Override
public void onNext(@NonNull ExpressInfo expressInfo) {
expressView.updateView(expressInfo);
}
@Override
public void onError(@NonNull Throwable e) {
expressView.showError(e.getMessage());
expressView.hideProgressDialog();
}
@Override
public void onComplete() {
expressView.hideProgressDialog();
}
});
}
项目：RxRelay
文件：BehaviorRelayTest.java
@Test(timeout = 1000)
public void testUnsubscriptionCase() {
BehaviorRelay<String> src = BehaviorRelay.createDefault("null"); // FIXME was plain null which is not allowed
for (int i = 0; i < 10; i++) {
final Observer<Object> o = TestHelper.mockObserver();
InOrder inOrder = inOrder(o);
String v = "" + i;
src.accept(v);
System.out.printf("Turn: %d%n", i);
src.firstElement()
.toObservable()
.flatMap(new Function<String, Observable<String>>() {
@Override
public Observable<String> apply(String t1) {
return Observable.just(t1 + ", " + t1);
}
})
.subscribe(new DefaultObserver<String>() {
@Override
public void onNext(String t) {
o.onNext(t);
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onComplete() {
o.onComplete();
}
});
inOrder.verify(o).onNext(v + ", " + v);
inOrder.verify(o).onComplete();
verify(o, never()).onError(any(Throwable.class));
}
}
项目：RxRelay
文件：BehaviorRelayTest.java
@Test(timeout = 1000)
public void testUnsubscriptionCase() {
BehaviorRelay<String> src = BehaviorRelay.createDefault("null"); // FIXME was plain null which is not allowed
for (int i = 0; i < 10; i++) {
final Observer<Object> o = TestHelper.mockObserver();
InOrder inOrder = inOrder(o);
String v = "" + i;
src.accept(v);
System.out.printf("Turn: %d%n", i);
src.firstElement()
.toObservable()
.flatMap(new Function<String, Observable<String>>() {
@Override
public Observable<String> apply(String t1) {
return Observable.just(t1 + ", " + t1);
}
})
.subscribe(new DefaultObserver<String>() {
@Override
public void onNext(String t) {
o.onNext(t);
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onComplete() {
o.onComplete();
}
});
inOrder.verify(o).onNext(v + ", " + v);
inOrder.verify(o).onComplete();
verify(o, never()).onError(any(Throwable.class));
}
}
项目：RxRelay
文件：ReplayRelayTest.java
@Test(timeout = 1000)
public void testUnsubscriptionCase() {
ReplayRelay<String> src = ReplayRelay.create();
for (int i = 0; i < 10; i++) {
final Observer<Object> o = TestHelper.mockObserver();
InOrder inOrder = inOrder(o);
String v = "" + i;
src.accept(v);
System.out.printf("Turn: %d%n", i);
src.firstElement()
.toObservable()
.flatMap(new Function<String, Observable<String>>() {
@Override
public Observable<String> apply(String t1) {
return Observable.just(t1 + ", " + t1);
}
})
.subscribe(new DefaultObserver<String>() {
@Override
public void onNext(String t) {
System.out.println(t);
o.onNext(t);
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onComplete() {
o.onComplete();
}
});
inOrder.verify(o).onNext("0, 0");
inOrder.verify(o).onComplete();
verify(o, never()).onError(any(Throwable.class));
}
}
项目：RxRelay
文件：BehaviorRelayTest.java
@Test
@Ignore("OOMs")
public void testEmissionSubscriptionRace() throws Exception {
Scheduler s = Schedulers.io();
Scheduler.Worker worker = Schedulers.io().createWorker();
try {
for (int i = 0; i < 50000; i++) {
if (i % 1000 == 0) {
System.out.println(i);
}
final BehaviorRelay<Object> rs = BehaviorRelay.create();
final CountDownLatch finish = new CountDownLatch(1);
final CountDownLatch start = new CountDownLatch(1);
worker.schedule(new Runnable() {
@Override
public void run() {
try {
start.await();
} catch (Exception e1) {
e1.printStackTrace();
}
rs.accept(1);
}
});
final AtomicReference<Object> o = new AtomicReference<Object>();
rs.subscribeOn(s).observeOn(Schedulers.io())
.subscribe(new DefaultObserver<Object>() {
@Override
public void onComplete() {
o.set(-1);
finish.countDown();
}
@Override
public void onError(Throwable e) {
o.set(e);
finish.countDown();
}
@Override
public void onNext(Object t) {
o.set(t);
finish.countDown();
}
});
start.countDown();
if (!finish.await(5, TimeUnit.SECONDS)) {
System.out.println(o.get());
System.out.println(rs.hasObservers());
fail("Timeout @ " + i);
break;
} else {
Assert.assertEquals(1, o.get());
}
}
} finally {
worker.dispose();
}
}
项目：RxRelay
文件：ReplayRelayConcurrencyTest.java
@Test
public void testReplayRelayEmissionSubscriptionRace() throws Exception {
Scheduler s = Schedulers.io();
Scheduler.Worker worker = Schedulers.io().createWorker();
try {
for (int i = 0; i < 50000; i++) {
if (i % 1000 == 0) {
System.out.println(i);
}
final ReplayRelay<Object> rs = ReplayRelay.create();
final CountDownLatch finish = new CountDownLatch(1);
final CountDownLatch start = new CountDownLatch(1);
worker.schedule(new Runnable() {
@Override
public void run() {
try {
start.await();
} catch (Exception e1) {
e1.printStackTrace();
}
rs.accept(1);
}
});
final AtomicReference<Object> o = new AtomicReference<Object>();
rs.subscribeOn(s).observeOn(Schedulers.io())
.subscribe(new DefaultObserver<Object>() {
@Override
public void onComplete() {
o.set(-1);
finish.countDown();
}
@Override
public void onError(Throwable e) {
o.set(e);
finish.countDown();
}
@Override
public void onNext(Object t) {
o.set(t);
finish.countDown();
}
});
start.countDown();
if (!finish.await(5, TimeUnit.SECONDS)) {
System.out.println(o.get());
System.out.println(rs.hasObservers());
Assert.fail("Timeout @ " + i);
break;
} else {
Assert.assertEquals(1, o.get());
}
}
} finally {
worker.dispose();
}
}
项目：RxRelay
文件：PublishRelayTest.java
@Test(timeout = 1000)
public void testUnsubscriptionCase() {
PublishRelay<String> src = PublishRelay.create();
for (int i = 0; i < 10; i++) {
final Observer<Object> o = TestHelper.mockObserver();
InOrder inOrder = inOrder(o);
String v = "" + i;
System.out.printf("Turn: %d%n", i);
src.firstElement()
.toObservable()
.flatMap(new Function<String, Observable<String>>() {
@Override
public Observable<String> apply(String t1) {
return Observable.just(t1 + ", " + t1);
}
})
.subscribe(new DefaultObserver<String>() {
@Override
public void onNext(String t) {
o.onNext(t);
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onComplete() {
o.onComplete();
}
});
src.accept(v);
inOrder.verify(o).onNext(v + ", " + v);
inOrder.verify(o).onComplete();
verify(o, never()).onError(any(Throwable.class));
}
}
项目：RxRelay
文件：ReplayRelayBoundedConcurrencyTest.java
@Test
public void testReplaySubjectEmissionSubscriptionRace() throws Exception {
Scheduler s = Schedulers.io();
Scheduler.Worker worker = Schedulers.io().createWorker();
try {
for (int i = 0; i < 50000; i++) {
if (i % 1000 == 0) {
System.out.println(i);
}
final ReplayRelay<Object> rs = ReplayRelay.createWithSize(2);
final CountDownLatch finish = new CountDownLatch(1);
final CountDownLatch start = new CountDownLatch(1);
// int j = i;
worker.schedule(new Runnable() {
@Override
public void run() {
try {
start.await();
} catch (Exception e1) {
e1.printStackTrace();
}
// System.out.println("> " + j);
rs.accept(1);
}
});
final AtomicReference<Object> o = new AtomicReference<Object>();
rs
// .doOnSubscribe(v -> System.out.println("!! " + j))
// .doOnNext(e -> System.out.println(">> " + j))
.subscribeOn(s)
.observeOn(Schedulers.io())
// .doOnNext(e -> System.out.println(">>> " + j))
.subscribe(new DefaultObserver<Object>() {
@Override
protected void onStart() {
super.onStart();
}
@Override
public void onComplete() {
o.set(-1);
finish.countDown();
}
@Override
public void onError(Throwable e) {
o.set(e);
finish.countDown();
}
@Override
public void onNext(Object t) {
o.set(t);
finish.countDown();
}
});
start.countDown();
if (!finish.await(5, TimeUnit.SECONDS)) {
System.out.println(o.get());
System.out.println(rs.hasObservers());
Assert.fail("Timeout @ " + i);
break;
} else {
Assert.assertEquals(1, o.get());
}
}
} finally {
worker.dispose();
}
}
项目：RetrofitRxErrorHandler
文件：RealExampleTest.java
/**
* This test executes the real query to github server.
* Test created by Robert Zagorski on 19.10.2016
*/
@Test
public void main() throws IOException, InterruptedException {
// Create a very simple REST adapter which points the GitHub API.
RxCallAdapter rxCallAdapter = new RxCallAdapter.Builder()
.addBackoffStrategy(Exponential.init()
.addThrowable(UnknownHostException.class)
.addThrowable(SocketTimeoutException.class)
.setMaxRetries(3).build())
.build();
Retrofit retrofit = new Retrofit.Builder()
.baseUrl(API_URL)
.addConverterFactory(GsonConverterFactory.create())
.addCallAdapterFactory(new RxErrorHandingFactory(rxCallAdapter))
.build();
// Create an instance of our GitHub API interface.
GitHub github = retrofit.create(GitHub.class);
// Create a call instance for looking up Retrofit contributors.
Observable<List<Repository>> call = github.repos("square");
final CountDownLatch latch = new CountDownLatch(1);
// Fetch and print a list of the contributors to the retrofiterrorhandler.
call.subscribe(new DefaultObserver<List<Repository>>() {
@Override
public void onComplete() {
System.out.println(new GregorianCalendar().toInstant().toString() + " Finished");
latch.countDown();
}
@Override
public void onError(Throwable e) {
System.out.println(new GregorianCalendar().toInstant().toString() + " Finished with error: " + e);
onComplete();
}
@Override
public void onNext(List<Repository> repositories) {
for (Repository repository : repositories) {
System.out.println(repository.name + " (" + repository.description + ")");
}
}
});
latch.await();
}
项目：RxRelay
文件：ReplayRelayTest.java
@Test(timeout = 1000)
public void testUnsubscriptionCase() {
ReplayRelay<String> src = ReplayRelay.create();
for (int i = 0; i < 10; i++) {
final Observer<Object> o = TestHelper.mockObserver();
InOrder inOrder = inOrder(o);
String v = "" + i;
src.accept(v);
System.out.printf("Turn: %d%n", i);
src.firstElement()
.toObservable()
.flatMap(new Function<String, Observable<String>>() {
@Override
public Observable<String> apply(String t1) {
return Observable.just(t1 + ", " + t1);
}
})
.subscribe(new DefaultObserver<String>() {
@Override
public void onNext(String t) {
System.out.println(t);
o.onNext(t);
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onComplete() {
o.onComplete();
}
});
inOrder.verify(o).onNext("0, 0");
inOrder.verify(o).onComplete();
verify(o, never()).onError(any(Throwable.class));
}
}
项目：RxRelay
文件：BehaviorRelayTest.java
@Test
@Ignore("OOMs")
public void testEmissionSubscriptionRace() throws Exception {
Scheduler s = Schedulers.io();
Scheduler.Worker worker = Schedulers.io().createWorker();
try {
for (int i = 0; i < 50000; i++) {
if (i % 1000 == 0) {
System.out.println(i);
}
final BehaviorRelay<Object> rs = BehaviorRelay.create();
final CountDownLatch finish = new CountDownLatch(1);
final CountDownLatch start = new CountDownLatch(1);
worker.schedule(new Runnable() {
@Override
public void run() {
try {
start.await();
} catch (Exception e1) {
e1.printStackTrace();
}
rs.accept(1);
}
});
final AtomicReference<Object> o = new AtomicReference<Object>();
rs.subscribeOn(s).observeOn(Schedulers.io())
.subscribe(new DefaultObserver<Object>() {
@Override
public void onComplete() {
o.set(-1);
finish.countDown();
}
@Override
public void onError(Throwable e) {
o.set(e);
finish.countDown();
}
@Override
public void onNext(Object t) {
o.set(t);
finish.countDown();
}
});
start.countDown();
if (!finish.await(5, TimeUnit.SECONDS)) {
System.out.println(o.get());
System.out.println(rs.hasObservers());
fail("Timeout @ " + i);
break;
} else {
Assert.assertEquals(1, o.get());
}
}
} finally {
worker.dispose();
}
}
项目：RxRelay
文件：ReplayRelayConcurrencyTest.java
@Test
public void testReplayRelayEmissionSubscriptionRace() throws Exception {
Scheduler s = Schedulers.io();
Scheduler.Worker worker = Schedulers.io().createWorker();
try {
for (int i = 0; i < 50000; i++) {
if (i % 1000 == 0) {
System.out.println(i);
}
final ReplayRelay<Object> rs = ReplayRelay.create();
final CountDownLatch finish = new CountDownLatch(1);
final CountDownLatch start = new CountDownLatch(1);
worker.schedule(new Runnable() {
@Override
public void run() {
try {
start.await();
} catch (Exception e1) {
e1.printStackTrace();
}
rs.accept(1);
}
});
final AtomicReference<Object> o = new AtomicReference<Object>();
rs.subscribeOn(s).observeOn(Schedulers.io())
.subscribe(new DefaultObserver<Object>() {
@Override
public void onComplete() {
o.set(-1);
finish.countDown();
}
@Override
public void onError(Throwable e) {
o.set(e);
finish.countDown();
}
@Override
public void onNext(Object t) {
o.set(t);
finish.countDown();
}
});
start.countDown();
if (!finish.await(5, TimeUnit.SECONDS)) {
System.out.println(o.get());
System.out.println(rs.hasObservers());
Assert.fail("Timeout @ " + i);
break;
} else {
Assert.assertEquals(1, o.get());
}
}
} finally {
worker.dispose();
}
}
项目：RxRelay
文件：PublishRelayTest.java
@Test(timeout = 1000)
public void testUnsubscriptionCase() {
PublishRelay<String> src = PublishRelay.create();
for (int i = 0; i < 10; i++) {
final Observer<Object> o = TestHelper.mockObserver();
InOrder inOrder = inOrder(o);
String v = "" + i;
System.out.printf("Turn: %d%n", i);
src.firstElement()
.toObservable()
.flatMap(new Function<String, Observable<String>>() {
@Override
public Observable<String> apply(String t1) {
return Observable.just(t1 + ", " + t1);
}
})
.subscribe(new DefaultObserver<String>() {
@Override
public void onNext(String t) {
o.onNext(t);
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onComplete() {
o.onComplete();
}
});
src.accept(v);
inOrder.verify(o).onNext(v + ", " + v);
inOrder.verify(o).onComplete();
verify(o, never()).onError(any(Throwable.class));
}
}
项目：RxRelay
文件：ReplayRelayBoundedConcurrencyTest.java
@Test
public void testReplaySubjectEmissionSubscriptionRace() throws Exception {
Scheduler s = Schedulers.io();
Scheduler.Worker worker = Schedulers.io().createWorker();
try {
for (int i = 0; i < 50000; i++) {
if (i % 1000 == 0) {
System.out.println(i);
}
final ReplayRelay<Object> rs = ReplayRelay.createWithSize(2);
final CountDownLatch finish = new CountDownLatch(1);
final CountDownLatch start = new CountDownLatch(1);
// int j = i;
worker.schedule(new Runnable() {
@Override
public void run() {
try {
start.await();
} catch (Exception e1) {
e1.printStackTrace();
}
// System.out.println("> " + j);
rs.accept(1);
}
});
final AtomicReference<Object> o = new AtomicReference<Object>();
rs
// .doOnSubscribe(v -> System.out.println("!! " + j))
// .doOnNext(e -> System.out.println(">> " + j))
.subscribeOn(s)
.observeOn(Schedulers.io())
// .doOnNext(e -> System.out.println(">>> " + j))
.subscribe(new DefaultObserver<Object>() {
@Override
protected void onStart() {
super.onStart();
}
@Override
public void onComplete() {
o.set(-1);
finish.countDown();
}
@Override
public void onError(Throwable e) {
o.set(e);
finish.countDown();
}
@Override
public void onNext(Object t) {
o.set(t);
finish.countDown();
}
});
start.countDown();
if (!finish.await(5, TimeUnit.SECONDS)) {
System.out.println(o.get());
System.out.println(rs.hasObservers());
Assert.fail("Timeout @ " + i);
break;
} else {
Assert.assertEquals(1, o.get());
}
}
} finally {
worker.dispose();
}
}