ObservableInteractor implements Interactor<DisposableObserver, P> { @SuppressWarnings("unchecked") @Override public void execute(DisposableObserver observer, P params) { DisposableObserver disposableObserver = buildReactiveStream(params) .subscribeOn(Schedulers.from(threadExecutor)) .observeOn(postExecutionThread.getScheduler()) .subscribeWith(observer); disposables.add(disposableObserver); } ObservableInteractor(ThreadExecutor threadExecutor,
      PostExecutionThread postExecutionThread); @SuppressWarnings("unchecked") @Override void execute(DisposableObserver observer, P params); @Override void dispose();  }
@Test public void shouldReturnEmptyStream() throws Exception { emptyObservableInteractor.execute(testDisposableObserver, null); assertEquals(0, testDisposableObserver.valuesCount); }
@Test(expected = NullPointerException.class) public void shouldFailWhenNullObserver() throws Exception { observableInteractor.execute(null, null); }