title: Reactor choosing a sink/processor
body: <p>I have the following use case:</p>
  
  <p>Out of an application I am consuming with X threads some messages, where I have a Consumer implementation defined like that:</p>
  
  <pre><code>public interface Consumer {
       onMessage(Object message);
  }
  </code></pre>
  
  <p>The problem is that Consumer is not a different instance per thread, but a single instance, as it is a Spring bean and we also expect it not to have side effects per single call of onMessage.</p>
  
  <p>However, what I want to build is a duplicate message detection mechanism, which kind of looks like this:</p>
  
  <pre><code>public static &lt;T&gt; Flux&lt;OcurrenceCache&lt;T&gt;&gt; getExceedingRates(Flux&lt;T&gt; values, int maxHits, int bufferSize, Duration bufferTimeout) {
          return values.bufferTimeout(bufferSize, bufferTimeout)
              .map(vals -&gt; {
                  OcurrenceCache&lt;T&gt; occurrenceCache = new OcurrenceCache&lt;&gt;(maxHits);
                  for (T value : vals) {
                      occurrenceCache.incrementNrOccurrences(value);
                  }
                  return occurrenceCache;
              });
  }
  </code></pre>
  
  <p>Where basically from a Flux of values I am returning an occurrence cache with the elements that are encountered more than the max desired number of hits.</p>
  
  <p>Naively, I can implement things like that:</p>
  
  <pre><code>public class MyConsumer implements Consumer {
      private final EmitterProcessor&lt;Object&gt; emitterProcessor;
  
      public MyConsumer(Integer maxHits, Integer bufferSize, Long timeoutMillis){
         this.emitterProcessor = EmitterProcessor.create();
         this.emitterProcessor
                  .bufferTimeout(bufferSize, Duration.ofMillis(timeoutMillis))
                  .subscribe(integers -&gt; {
                      getExceedingRates(Flux.fromIterable(integers), maxHits, bufferSize, Duration.ofMillis(timeoutMillis))
                              .subscribe(integerOcurrenceCache -&gt; {
                                  System.out.println(integerOcurrenceCache.getExceedingValues());
                          });
                  });
      }
  
      @Override
      public void onMessage(Object message){
          emitterProcessor.onNext(message); 
      }
  }
  </code></pre>
  
  <p>However, this is far from optimal because I know that my messages from a specific thread will <strong>NEVER</strong> contain any the messages that came from another thread (they are pre-grouped as we use jms grouping and kinesis sharding). So, in a way, I'd like to use such a Processor that will:</p>
  
  <ul>
  <li>use the very same thread on which onMessage was called to kind of isolate the flux in such a way where values from it are isolated and not mixed up with the variables put from another thread.</li>
  </ul>