title: How to sort an infinite Flux?
body: <p>I have an infinite <code>Flux</code> instance produced by a dedicated <code>Processor</code> meaning that each element is emitted via <code>sink.next</code> (elements come from reactive Kafka <code>Receiver</code> if that matters). 
  The issues is that every time I try to do something useful like <code>sort</code> on this <code>Flux</code> it just gives an empty result. 
  This applies to <code>reduce</code> as well. </p>
  
  <p>Am I doing something wrong?</p>
  
  <p><strong>EDIT</strong></p>
  
  <p>Here's a more concrete example, it gives an empty <code>Flux</code>:</p>
  
  <pre><code>Flux.&lt;Integer&gt;create(sink -&gt; {
    sink.next(1);
    sink.next(2);
    sink.next(3);
    sink.next(4);
  })
      .sort() // If I remove this everthing works as expected
      .log()
      .subscribe();
  </code></pre>
  
  <p>Talking about my specific case, here's the simplified version of what I have:</p>
  
  <pre><code>    FluxProcessor&lt;Message, Message&gt; processor = ReplayProcessor.&lt;Message&gt;createTimeout(Duration.ofDays(1)).serialize();
      FluxSink&lt;Message&gt; sink = processor.sink();
      Flux&lt;Message&gt; pipeline = processor;
  
      kafka.receive()
          .log()
          .map(ReceiverRecord::value)
          .subscribe(sink::next);
  
      return pipeline; // Work with the pipeline later on
  </code></pre>
  
  <p>Then if I try to either <code>.sort</code> or <code>.reduce</code> the <code>pipeline</code> it always results in empty <code>Flux</code>.</p>