title: Non-blocking streaming of data between two Quarkus services (Vert.x with Mutiny in Java)
body: <p><strong>Update!</strong></p>
  <p>I have fixed minor bugs in the sample code after solving some of the problems that were irrelevant to the main question which is still about non-blocking streaming between services.</p>
  <p><strong>Background info</strong>:</p>
  <p>I'm porting a Spring WebFlux service under Quarkus. The service runs long searches on multiple huge data sets and returns the partial results in a Flux (text/event-stream) as they become available.</p>
  <p><strong>Problem</strong>:</p>
  <p>Right now I'm trying to use Mutiny Multi with Vert.x under Quarkus but cannot figure out how the consumer service could receive this stream without blocking.</p>
  <p>In all examples the consumer is either a JS front end page or the producer's content type is application/json that seems to bluck until the Multi completes before sending it over in one JSON object (which makes no sense in my application).</p>
  <p><strong>Questions</strong>:</p>
  <ol>
  <li>How to receive text/event-stream with the Mutiny-flavoured Vert.x WebClient?</li>
  <li>If the problem would be that the WebClient is not able to receive continuous steams: What is the standard way to stream data between two Quarkus services?</li>
  </ol>
  <p><strong>Here is a simplified example</strong></p>
  <p><strong>Test entity</strong></p>
  <pre class="lang-java prettyprint-override"><code>public class SearchResult implements Serializable {
  
      private String content;
  
      public SearchResult(String content) {
          this.content = content;
      }
  
  
      //.. toString, getters and setters
      
  }
  </code></pre>
  <p><strong>Producer 1. simple infinite stream -&gt; hangs</strong></p>
  <pre class="lang-java prettyprint-override"><code>@GET
  @Path(&quot;/search&quot;)
  @Produces(MediaType.SERVER_SENT_EVENTS)
  @SseElementType(MediaType.APPLICATION_JSON)
  public Multi&lt;SearchResult&gt; getResults() {
          return Multi.createFrom().ticks().every(Duration.ofSeconds(2)              .onItem().transform(n -&gt; new SearchResult(n.toString()));
  }
  </code></pre>
  <p><strong>Producer 2. with Vertx Paths infinite stream -&gt; hangs</strong></p>
  <pre class="lang-java prettyprint-override"><code>@Route(path = &quot;/routed&quot;, methods = HttpMethod.GET)
  public Multi&lt;SearchResult&gt; getSrStreamRouted(RoutingContext context) {
          log.info(&quot;routed run&quot;);
          return ReactiveRoutes.asEventStream(Multi.createFrom().ticks().every(Duration.ofSeconds(2))
                  .onItem().transform(n -&gt; new SearchResult(n.toString()));
  }
  
  </code></pre>
  <p><strong>Producer 3. simple finite stream -&gt; blocks until completion</strong></p>
  <pre class="lang-java prettyprint-override"><code>@GET
  @Path(&quot;/search&quot;)
  @Produces(MediaType.SERVER_SENT_EVENTS)
  @SseElementType(MediaType.APPLICATION_JSON)
  public Multi&lt;SearchResult&gt; getResults() {
      return Multi.createFrom().ticks().every(Duration.ofSeconds(2))
          .transform().byTakingFirstItems(5)
          .onItem().transform(n -&gt; new SearchResult(n.toString()));
  }
  </code></pre>
  <p><strong>Consumer</strong>:</p>
  <p>Tried multiple different solutions on both producer and consumer sides, but <strong>in every case the the stream blocks until it is complete</strong> or hangs indefinitely without transferring data for infinite streams. I get the same results with httpie. Here is the latest iteration:</p>
  <pre class="lang-java prettyprint-override"><code>WebClientOptions webClientOptions = new WebClientOptions().setDefaultHost(&quot;localhost&quot;).setDefaultPort(8182);
  WebClient client = WebClient.create(vertx, webClientOptions);
          
  client.get(&quot;/string&quot;)
                  .send()
                  .onFailure().invoke(resp -&gt; log.error(&quot;error: &quot; + resp))
                  .onItem().invoke(resp -&gt; log.info(&quot;result: &quot; + resp.statusCode()))
                  .toMulti()
                  .subscribe().with(r -&gt; log.info(String.format(&quot;Subscribe: code:%d body:%s&quot;,r.statusCode(), r.bodyAsString())));
  </code></pre>