MessageProcessor implements Processor<Message<T>, Message<T>>, Subscription { @Override public void onError(Throwable t) { if (subscriber != null) { subscriber.onError(t); } }  @Override void subscribe(Subscriber<? super Message<T>> subscriber); @Override void request(long n); @Override void cancel(); @Override void onNext(Message<T> t); @Override void onError(Throwable t); @Override void onSubscribe(Subscription s); @Override void onComplete();  }
@Test public void onError() throws Exception { Exception exception = new Exception(); underTest.subscribe(subscriber); underTest.onError(exception); underTest.onError(exception); verify(subscriber, times(1)).onError(exception); }