@Test public void shouldFillInObservableThreadInfo() throws Throwable { loggableObservable.get(observableRule.stringType()) .subscribeOn(Schedulers.immediate()) .observeOn(Schedulers.immediate()) .subscribe(subscriber); final ObservableInfo observableInfo = loggableObservable.getInfo(); final Optional<String> subscribeOnThread = observableInfo.getSubscribeOnThread(); final Optional<String> observeOnThread = observableInfo.getObserveOnThread(); final String currentThreadName = Thread.currentThread().getName(); assertThat(subscribeOnThread.isPresent()).isTrue(); assertThat(observeOnThread.isPresent()).isTrue(); assertThat(subscribeOnThread.get()).isEqualTo(currentThreadName); assertThat(observeOnThread.get()).isEqualTo(currentThreadName); }
@Override <T> Observable<T> get(T type) throws Throwable { return ((Observable<T>) joinPoint.proceed()) .doOnEach(new Action1<Notification<? super T>>() { @Override public void call(Notification<? super T> notification) { if (!observableInfo.getSubscribeOnThread().isPresent() && (notification.isOnNext() || notification.isOnError())) { observableInfo.setSubscribeOnThread(Thread.currentThread().getName()); } } }) .doOnUnsubscribe(new Action0() { @Override public void call() { if (!observableInfo.getObserveOnThread().isPresent()) { observableInfo.setObserveOnThread(Thread.currentThread().getName()); } messageManager.printObservableThreadInfo(observableInfo); } }); }
LogSchedulersObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { return ((Observable<T>) joinPoint.proceed()) .doOnEach(new Action1<Notification<? super T>>() { @Override public void call(Notification<? super T> notification) { if (!observableInfo.getSubscribeOnThread().isPresent() && (notification.isOnNext() || notification.isOnError())) { observableInfo.setSubscribeOnThread(Thread.currentThread().getName()); } } }) .doOnUnsubscribe(new Action0() { @Override public void call() { if (!observableInfo.getObserveOnThread().isPresent()) { observableInfo.setObserveOnThread(Thread.currentThread().getName()); } messageManager.printObservableThreadInfo(observableInfo); } }); } }
LogSchedulersObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { return ((Observable<T>) joinPoint.proceed()) .doOnEach(new Action1<Notification<? super T>>() { @Override public void call(Notification<? super T> notification) { if (!observableInfo.getSubscribeOnThread().isPresent() && (notification.isOnNext() || notification.isOnError())) { observableInfo.setSubscribeOnThread(Thread.currentThread().getName()); } } }) .doOnUnsubscribe(new Action0() { @Override public void call() { if (!observableInfo.getObserveOnThread().isPresent()) { observableInfo.setObserveOnThread(Thread.currentThread().getName()); } messageManager.printObservableThreadInfo(observableInfo); } }); } LogSchedulersObservable(FrodoProceedingJoinPoint joinPoint,
      MessageManager messageManager, ObservableInfo observableInfo); }
LogSchedulersObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { return ((Observable<T>) joinPoint.proceed()) .doOnEach(new Action1<Notification<? super T>>() { @Override public void call(Notification<? super T> notification) { if (!observableInfo.getSubscribeOnThread().isPresent() && (notification.isOnNext() || notification.isOnError())) { observableInfo.setSubscribeOnThread(Thread.currentThread().getName()); } } }) .doOnUnsubscribe(new Action0() { @Override public void call() { if (!observableInfo.getObserveOnThread().isPresent()) { observableInfo.setObserveOnThread(Thread.currentThread().getName()); } messageManager.printObservableThreadInfo(observableInfo); } }); } LogSchedulersObservable(FrodoProceedingJoinPoint joinPoint,
      MessageManager messageManager, ObservableInfo observableInfo);  }
LogSchedulersObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { return ((Observable<T>) joinPoint.proceed()) .doOnEach(new Action1<Notification<? super T>>() { @Override public void call(Notification<? super T> notification) { if (!observableInfo.getSubscribeOnThread().isPresent() && (notification.isOnNext() || notification.isOnError())) { observableInfo.setSubscribeOnThread(Thread.currentThread().getName()); } } }) .doOnUnsubscribe(new Action0() { @Override public void call() { if (!observableInfo.getObserveOnThread().isPresent()) { observableInfo.setObserveOnThread(Thread.currentThread().getName()); } messageManager.printObservableThreadInfo(observableInfo); } }); } LogSchedulersObservable(FrodoProceedingJoinPoint joinPoint,
      MessageManager messageManager, ObservableInfo observableInfo);   }