title: Using cyclops-react for batching on a async queue stream
body: <p>I am trying to use cyclops-react to batch the elements from a queue, based on size, but also on time, so it doesn't block when there are no elements</p>
  
  <p>Maybe the functionality is not what I expected or I am doing something wrong</p>
  
  <p>The complete code (Groovy) is like this with the producer in another thread:</p>
  
  <pre><code>            Queue&lt;String&gt; queue = QueueFactories.&lt;String&gt;unboundedQueue().build();
      new Thread({
          while (true) {
              sleep(1000)
              queue.offer("New message " + System.currentTimeMillis());
          }
      }).start();
  
      StreamSource.futureStream(queue, new LazyReact(ThreadPools.queueCopyExecutor))
              .groupedBySizeAndTime(10,500,TimeUnit.MILLISECONDS)
              .forEach({i-&gt;println(i + " Batch Time: ${System.currentTimeMillis()}")})
  </code></pre>
  
  <p>The output is:</p>
  
  <pre><code>    [New message 1487673650332,  Batch Time: 1487673651356]
      [New message 1487673651348, New message 1487673652352,  Batch Time: 1487673653356]
      [New message 1487673653355, New message 1487673654357,  Batch Time: 1487673655362]
      [New message 1487673655362, New message 1487673656364,  Batch Time: 1487673657365]
  </code></pre>
  
  <p>But I was expecting one element in each batch since the delay between elements offered is 10seconds but the batching is every half a second</p>
  
  <p>Also I tried with an asynchronous stream (Groovy code):</p>
  
  <pre><code>    Queue&lt;String&gt; queue = QueueFactories.&lt;String&gt;unboundedQueue().build();
      StreamSource.futureStream(queue, new LazyReact(ThreadPools.queueCopyExecutor))
              .async()
              .groupedBySizeAndTime(10, 500,TimeUnit.MILLISECONDS)
              .peek({i-&gt;println(i + "Batch Time: ${System.currentTimeMillis()}")}).run();
  
      while (true) {
          queue.offer("New message " + System.currentTimeMillis());
          sleep(1000)
      }
  </code></pre>
  
  <p>Again, it only batches every 2 seconds, sometimes waiting for two elements per batch, even if the timeout in the batch is half second:</p>
  
  <pre><code>    [New message 1487673877780, Batch Time: 1487673878819]
      [New message 1487673878811, New message 1487673879812, Batch Time: 1487673880815]
      [New message 1487673880814, New message 1487673881819, Batch Time: 1487673882823]
      [New message 1487673882823, New message 1487673883824, Batch Time: 1487673884828]
      [New message 1487673884828, New message 1487673885831, Batch Time: 1487673886835]
  </code></pre>
  
  <p>I did a third experiment with a non future non lazy stream, and this time it worked.</p>
  
  <pre><code>    Queue&lt;String&gt; queue = QueueFactories.&lt;String&gt;unboundedQueue().build();
      new Thread({
          while (true) {
              sleep(1000)
              queue.offer("New message " + System.currentTimeMillis());
          }
      }).start();
  
      queue.stream()
              .groupedBySizeAndTime(10,500,TimeUnit.MILLISECONDS)
              .forEach({i-&gt;println(i + " Batch Time " + System.currentTimeMillis())})
  </code></pre>
  
  <p>Result:</p>
  
  <pre><code>    [New message 1487673288017, New message 1487673289027,  Batch Time , 1487673289055]
      [New message 1487673290029,  Batch Time , 1487673290029]
      [New message 1487673291033,  Batch Time , 1487673291033]
      [New message 1487673292037,  Batch Time , 1487673292037]
  </code></pre>
  
  <p>Why the behaviour of the batching seems to be wrong when you use a future stream?</p>