title: Have a Reactor Flux that behaves like a queue
body: <p>The need is to have a Flux that acts similar to a queue that must be shared between different subscribers, where each subscribers can take it's own unique and unshared value from the queue.</p>
  <p>Each call to next() should return a different value and the sequence should always be contiguous as expressed in this test and not start over.</p>
  <p>Also parallel calls from parallel flux should't get duplicate values.</p>
  <p>The sequence dao is a mock of methods that calls r2dbc api, so it has to be reactive methods like this. Therefore using the generate functions doesn't seems a possible option, unless there are tricks I probably don't know.</p>
  <pre class="lang-java prettyprint-override"><code>
  public class SequenceServiceTest {
  
      static class SequenceDaoMock {
  
          private static final long DEV_SEQ_INCREMENT = 5;
  
          private final AtomicLong databaseSequenceMockAtomicLong = new AtomicLong(10);
  
          public Mono&lt;Long&gt; selectNextVal(){
              return Mono.fromSupplier(() -&gt; databaseSequenceMockAtomicLong.getAndAdd(DB_SQ_INCREMENT);
          }
      
          public Mono&lt;Long&gt; getSequenceIncrement(){
              return Mono.fromSupplier(() -&gt; DB_SQ_INCREMENT);
          }
  
      }
  
      @Test
      void test1(){
          SequenceDaoMock sequenceDao = new SequenceDaoMock(); 
          Mono&lt;Long&gt; sequenceIncrementMono = sequenceDao.getSequenceIncrement().cache();
          Flux&lt;Long&gt; originalSequence = Mono.zip(sequenceDao.selectNextVal(),sequenceIncrementMono)
                                    .flatMapMany(t -&gt; Flux.range(0, t.getT2().intValue()).map(n -&gt; t.getT1() + n))
                                    .repeat();
          //code that should be changed because tests fails
          WorkQueueProcessor&lt;Long&gt; queue = WorkQueueProcessor.&lt;Long&gt;builder().share(false).bufferSize(4).build();
          sequence.subscribe(queue); 
          Flux&lt;Long&gt; sequence = queue.drain().publish().autoconnect(1);
          //end of code that should be changed because tests fails
  
          Mono.zip(sequence.next(),sequence.next(),sequence.next())
              .flatMapMany(t -&gt; Flux.just(t.getT1(),t.getT2(),t.getT3()))
              .zipWith(seqence.next().repeat())
              .flatMap(t -&gt; Flux.just(t.getT1(),t.getT2()))
              .distinct()
              .sort()
              .as(StepVerifier::create)
              .expectNext(10L)
              .expectNext(11L)
              .expectNext(12L)
              .expectNext(13L)
              .expectNext(14L)
              .expectNext(15L)
              .verifyComplete();
  
              Flux.just(&quot;a&quot;,&quot;b&quot;,&quot;c&quot;,&quot;d&quot;,&quot;e&quot;,&quot;f&quot;,&quot;g&quot;,&quot;h&quot;,&quot;i&quot;,&quot;j&quot;,&quot;k&quot;,&quot;l&quot;,&quot;m&quot;,&quot;n&quot;,&quot;o&quot;,&quot;p&quot;,&quot;q&quot;,&quot;r&quot;,&quot;s&quot;,&quot;t&quot;,&quot;&quot;,&quot;u&quot;,&quot;v&quot;,&quot;w&quot;,&quot;x&quot;,&quot;y&quot;,&quot;z&quot;)
              .parallel(3)
              .runOn(Schedulers.newParallel(&quot;para&quot;,3))
              .flatMap(letter -&gt; sequence.next().map(seq -&gt; Tuple.of(seq,letter)))
              .sequential()
              .collectList()
              .as(StepVerifier::create)
              .assertNext( list -&gt; {
                   SortedSet set = list.stream().map(t -&gt; t.getT1()).collect(Collectors.toCollection(TreeSet::new));
                   Collectors.toCollection(TreeSet::new) 
                   assertTrue(set.size(26));
                   assertEquals(set.first(),16L);
                   assertEquals(set.last(),41L);
               })
              .verifyComplete();
  
  
      }
  
  }
  </code></pre>
  <p>The test fails because I get duplicate values in the parallel test, so there must be a concurrency problem.</p>
  <p>Some code could be refactored like, switching back from .parallel().flatMap( ... ).sequential().zipwith(sequence).parallel(), which would protect concurrency from this flux but probably not from other flux instances.</p>
  <p>Also Flux.zip( ) doesn't seem to create an issue with 3 values, but I guess it could have the issue since it's asynchronous.</p>
  <p>Also one thing I noticed is that when you start .publish() and .connect(), then the flux starts publishing a lot of values very fast.
  Doess the notion of unbounded/bounded requests have an influence on this, or do we have no other choices than having a lot of values sent and lost ? This is undesirable in that case. Also using a Duration to slow down would be a solution since it would limit performances.</p>
  <h3>EDIT :</h3>
  <p>With JDBC I can do it with the code below, but with R2DBC I can't do something similar because I can't call .block() from anywhere and keep state, and since spring boot excludes JDBC when you are are setting up R2DBC, I would have though there would be a way to do that exclusively with R2DBC.</p>
  <p>I think I can afford doing that with JDBC because the cost of blocking every 100000 ids, is far inferior to querying the database with R2DBC to get an individual value each time. I haven't tested it though, but I can't imagine how it would be faster, than incrementing ids in memory.</p>
  <p>Also I don't need to do a transaction for a sql nextval query because it's not transactional.</p>
  <p>So I guess my quick win is to create a JDBC datasource in the back of spring boot, and get my sequence that way.</p>
  <pre class="lang-java prettyprint-override"><code>public class SequenceService {
  
      private long sequenceIncrement;
      private long dbNextVal;
      private long limit;
      private AtomicLong virtualSequence;
  
      @PostConstruct
      public void init(){
          sequenceIncrement = jdbc.query(&quot;select sequenceIncrement from sequenceTable where sequenceName='mySequence'&quot;);
          dbNextVal = getDbNextVal();
          limit = dbNextVal + sequenceIncrement
          virtualSequence = new AtomicLong(dbNextVal);
      }
   
      private long resetDbNextVal(){
          return jdbc.query(&quot;select mySequence.nextVal() from dual&quot;);
      }
  
      private Mono&lt;Long&gt; next(){
          Mono.fromSupplier(() -&gt; {
              return virtualSequence.getAndUpdate(seq -&gt; {
                  nextSeq = seq + 1;
                  if(nextSeq == limit) {
                       dbNextVal = getDbNextVal();
                       limit = dbNextVal + sequenceIncrement;
                       nextSeq = dbNextVal;   
                  }
                  return nextSeq;
          });
      }  
  }
  </code></pre>