title: "Bridging" Reactor's Flux from gRPC StreamObserver
body: <p>I want to create a <a href="https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html" rel="nofollow noreferrer">Reactor Flux</a> from a <a href="https://grpc.io/grpc-java/javadoc/io/grpc/stub/StreamObserver.html" rel="nofollow noreferrer">gRPC StreamObserver</a>. This needs to be done as long as StreamObserver does not implement the respective interfaces natively (see e.g. <a href="https://github.com/LogNet/grpc-spring-boot-starter/issues/3" rel="nofollow noreferrer">this issue</a>).</p>
  
  <p>What I came up with is roughly like the following:</p>
  
  <pre><code>final StreamObserver&lt;ProtoResponse&gt;[] streamObserverArray = new  StreamObserver[1];
  Flux&lt;Response&gt; myFlux Flux.create(sink -&gt; streamObserverArray[0] = new StreamObserver&lt;ProtoResponse&gt;() {
          @Override
          public void onNext(ProtoResponse value) {
              final Response response = convertFromProto(value);
              sink.next(response);
          }
  
          @Override
          public void onError(Throwable throwable) {
              sink.error(throwable);
          }
  
          @Override
          public void onCompleted() {
              sink.complete();
          }
      });
  myFlux
      .doOnError(throwable -&gt; {/* actual logic in here */}) //
      .doOnComplete(() -&gt; {/* actual logic in here */}) //
      .doOnCancel(() -&gt; {/* actual logic in here */}) //
      .parallel() //
      .runOn(Schedulers.parallel()) //
      .doOnNext(/* actual heavy lifting logic in here */) //
      .map(/* ... */) //
      .sequential() //
      .doOnNext(/* ...*/) //
      .subscribe(); // needed to start the actual processing of the events on this Flux
  
  MyGrpcService.newStub(channel).getResponses(protoRequest, streamObserverArray[0]);
  </code></pre>
  
  <p>The main idea why I want to use Reactor here is to distribute the "heavy lifting work" onto several threads in parallel and not do that on the gRPC request threads. </p>
  
  <p>I see several problems with the approach as it is done above:</p>
  
  <ul>
  <li>I really don't like the workaround with the <code>StreamObserver[]</code> array</li>
  <li>I need to create the complete flux first because if I don't finish it off with <code>.subscribe()</code> first, the <code>StreamObserver</code> might be <code>null</code> when gRPC starts communicating (aka race condition).</li>
  <li>I am not sure if backpressure works the way it is intended (although this is currently not my main concern).</li>
  </ul>
  
  <p>So my questions would be:
  What is the best/preferred way to bridge from a gRPC StreamObserver to a Reactor Flux? Are there any best practices?</p>