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