Calls StreamObserver.onStop only on termination.

This calls onStop once, when the processor is being stopped. It also
documents onCompleted as being called after a processor has finished
due to being configure ot fetch only a subset. Previously onStop
was being called multiple times, for example on every retry.

For #289.

nakadi-java-client/src/main/java/nakadi/StreamObserver.java

@@ -27,13 +27,18 @@ default void onBegin() {
   void onStart();
 
   /**
-   * Called after the stream either cannot be recovered and retried, or has completed.
+   * Called after the stream either cannot be recovered and retried, or has been stopped.
    */
   void onStop();
 
   /**
    * Notifies the Observer that the {@link StreamProcessorManaged} has finished sending batches.
    *
+   * <p>
+   *  This signals the end of a stream. which happens when a processor was configured to handle a
+   *  fixed number of items. Typically Nakadi streams are considered infinite.
+   *</p>
+   *
    * <p>Once the {@link StreamProcessorManaged} calls this method, it will not call
    * {@link #onNext}. </p>
    */

nakadi-java-client/src/main/java/nakadi/StreamProcessor.java

@@ -68,7 +68,8 @@ public class StreamProcessor implements StreamProcessorManaged {
               (t, e) -> handleUncaught(t, e, "stream_processor_err_compute"))
           .build());
   private final Scheduler monoComputeScheduler = Schedulers.from(monoComputeExecutor);
-
+  private Flowable<StreamBatchRecord<?>> _observable;
+  private StreamObserver streamObserver;
 
   @VisibleForTesting
   @SuppressWarnings("unused") StreamProcessor(NakadiClient client,
@@ -222,6 +223,8 @@ private void stopStreaming() {
           this.failedProcessorException.getMessage());
     }
 
+    streamObserver.onStop();
+
     subscriber.dispose();
     logger.debug("op=stream_processor_stop msg=stopping_executor name=monoIoScheduler");
     ExecutorServiceSupport.shutdown(monoIoExecutor);
@@ -233,6 +236,7 @@ private void stopStreaming() {
   private <T> void stream(StreamConfiguration sc, StreamObserverProvider<T> provider) {
 
     final StreamObserver<T> observer = provider.createStreamObserver();
+    streamObserver = observer;
     final TypeLiteral<T> literal = provider.typeLiteral();
     final Flowable<StreamBatchRecord<T>> observable = this.buildObservable(observer, sc, literal);
 
@@ -290,6 +294,7 @@ private <T> Flowable<StreamBatchRecord<T>> buildObservable(
         .doOnSubscribe(subscription -> streamObserver.onStart())
         .doOnComplete(streamObserver::onCompleted)
         .doOnCancel(() -> {
+          System.out.println("wwwwwwwwwwwwwwwwwww");
           if (retryAttemptsFinished.get()) {
 
             logger.info("op=stop_processor msg=stopping_after_stream_retry_finished");
@@ -299,8 +304,6 @@ private <T> Flowable<StreamBatchRecord<T>> buildObservable(
             // land in setupRxErrorHandler's RxJavaPlugins.setErrorHandler.
             stopStreaming();
 
-          } else {
-            streamObserver.onStop();
           }
         })
         .timeout(halfOpenKick, halfOpenUnit)