title: How to introduce delay within a RxJava group processed parallely?
body: <p>My use case is to group a stream, start processing some of the groups parallely, and within each group, delay processing of each item by a constant interval. I can't seem to get the delay within the group right because instead of being emitted periodically, they are emitted almost instantaneously. Following is my test code using RxJava 2.0.5:</p>
  
  <pre><code>@Slf4j
  public class GroupsAndDelays {
      Function&lt;Integer, Flowable&lt;Integer&gt;&gt; remoteClient;
      int groupMemberDelaySeconds;
      int remoteCallTimeoutSeconds;
      int maxRetryCount;
      int retryDelaySeconds;
      Map&lt;Long, List&lt;Integer&gt;&gt; threadMap;
      Map&lt;Long, List&lt;Integer&gt;&gt; resultThreadMap;
  
      public ParallelFlowable&lt;Integer&gt; doStuff(Flowable&lt;Integer&gt; src,
                                               Function&lt;Integer, Integer&gt; groupByFn,
                                               Function&lt;Integer, Flowable&lt;Integer&gt;&gt; responseMapper) {
          return src
                  .groupBy(groupByFn)
                  .parallel(5).runOn(Schedulers.newThread())
                  .map(g -&gt; g.distinct().toList())
                  .flatMap(i -&gt; i.toFlowable())
                  .flatMap(i -&gt; {
                      log.debug("Processing group: {}.", i);
                      return Flowable.fromIterable(i)
                              .delay(groupMemberDelaySeconds, SECONDS);
                  })
                  .flatMap(i -&gt; {
                      log.debug("Processing: {}.", i);
                      putInThreadMap(threadMap, i);
                      return remoteCall(i * 2, responseMapper);
                  });
      }
  
      private Flowable&lt;Integer&gt; remoteCall(int i, Function&lt;Integer, Flowable&lt;Integer&gt;&gt; responseMapper) throws
              Exception {
          return remoteClient.apply(i)
                  .timeout(remoteCallTimeoutSeconds, SECONDS)
                  .retryWhen(t -&gt; t.zipWith(Flowable.range(1, maxRetryCount), (ex, retryCount) -&gt; retryCount)
                          .flatMap(retryCount -&gt; Flowable.timer(retryCount * retryDelaySeconds, SECONDS)))
                  .flatMap(result -&gt; {
                      log.debug("Processing result: {}.", result);
                      putInThreadMap(resultThreadMap, result);
                      return responseMapper.apply(result);
                  })
                  .onErrorReturnItem(-1);
      }
  
      private void putInThreadMap(Map&lt;Long, List&lt;Integer&gt;&gt; map, int i) {
          map.merge(Thread.currentThread().getId(), singletonList(i), this::merge);
      }
  
      private List&lt;Integer&gt; merge(List&lt;Integer&gt; a, List&lt;Integer&gt; b) {
          return Stream.concat(a.stream(), b.stream()).collect(Collectors.toList());
      }
  }
  </code></pre>
  
  <p>Here's a Spock test:</p>
  
  <pre><code>class GroupsAndDelaysSpec extends Specification {
      final int groupMemberDelaySeconds = 3
      final int remoteCallTimeoutSeconds = 3
      final int maxRetryCount = 2
      final int retryDelaySeconds = 2
      Function&lt;Integer, Flowable&lt;Integer&gt;&gt; remoteClient
      Function&lt;Integer, Integer&gt; groupByFn
      Function&lt;Integer, Flowable&lt;Integer&gt;&gt; responseMapper
  
      GroupsAndDelays groupsAndDelays
  
      final Flowable&lt;Integer&gt; src = Flowable.fromArray(
              1, 2, 3, 4, 5, 1, 2, 3, 4, 5,
              11, 12, 13, 14, 15, 11, 12, 13, 14, 15,
              21, 22, 23, 24, 25, 21, 22, 23, 24, 25,
              31, 32, 33, 34, 35, 31, 32, 33, 34, 35,
              41, 42, 43, 44, 45, 41, 42, 43, 44, 45
      )
  
      def setup() {
          remoteClient = Mock(Function)
  
          groupsAndDelays = new GroupsAndDelays()
          groupsAndDelays.groupMemberDelaySeconds = groupMemberDelaySeconds
          groupsAndDelays.remoteCallTimeoutSeconds = remoteCallTimeoutSeconds
          groupsAndDelays.maxRetryCount = maxRetryCount
          groupsAndDelays.retryDelaySeconds = retryDelaySeconds
          groupsAndDelays.remoteClient = remoteClient
          groupsAndDelays.threadMap = new ConcurrentHashMap&lt;Long, List&lt;Integer&gt;&gt;()
          groupsAndDelays.resultThreadMap = new ConcurrentHashMap&lt;Long, List&lt;Integer&gt;&gt;()
  
          groupByFn = Mock(Function)
          groupByFn.apply(_) &gt;&gt; { args -&gt; args[0] % 10 }
  
          responseMapper = Mock(Function)
          responseMapper.apply(_) &gt;&gt; { args -&gt; args[0] }
      }
  
      def cleanup() {
          println("Thread map: ${groupsAndDelays.threadMap}")
          println("Result thread map: ${groupsAndDelays.resultThreadMap}")
  
          assert groupsAndDelays.threadMap.size() == 5
          assert groupsAndDelays.threadMap.findAll { k, v -&gt; v.size() == 5 }.size() == 5
      }
  
      def "each group executes on a separate thread"() {
          setup:
          remoteClient.apply(_) &gt;&gt; { args -&gt; Flowable.just(args[0]) }
  
          when:
          groupsAndDelays.doStuff(src, groupByFn, responseMapper)
                  .sequential()
                  .toList()
                  .blockingGet()
  
          then:
          true
      }
  }
  </code></pre>
  
  <p>Sample run:</p>
  
  <pre><code>2017-02-04 00:49:19.430 [RxNewThreadScheduler-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$1 - Processing group: [3, 13, 23, 33, 43].
  2017-02-04 00:49:19.430 [RxNewThreadScheduler-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$1 - Processing group: [1, 11, 21, 31, 41].
  2017-02-04 00:49:19.430 [RxNewThreadScheduler-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$1 - Processing group: [5, 15, 25, 35, 45].
  2017-02-04 00:49:19.430 [RxNewThreadScheduler-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$1 - Processing group: [2, 12, 22, 32, 42].
  2017-02-04 00:49:19.430 [RxNewThreadScheduler-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$1 - Processing group: [4, 14, 24, 34, 44].
  2017-02-04 00:49:22.443 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 2.
  2017-02-04 00:49:22.443 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 1.
  2017-02-04 00:49:22.443 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 5.
  2017-02-04 00:49:22.443 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 4.
  2017-02-04 00:49:22.443 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 3.
  2017-02-04 00:49:22.456 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 10.
  2017-02-04 00:49:22.456 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 2.
  2017-02-04 00:49:22.456 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 8.
  2017-02-04 00:49:22.456 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 6.
  2017-02-04 00:49:22.456 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 4.
  2017-02-04 00:49:22.459 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 13.
  2017-02-04 00:49:22.459 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 14.
  2017-02-04 00:49:22.459 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 11.
  2017-02-04 00:49:22.459 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 15.
  2017-02-04 00:49:22.459 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 12.
  2017-02-04 00:49:22.466 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 26.
  2017-02-04 00:49:22.466 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 30.
  2017-02-04 00:49:22.466 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 24.
  2017-02-04 00:49:22.466 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 22.
  2017-02-04 00:49:22.466 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 28.
  2017-02-04 00:49:22.466 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 23.
  2017-02-04 00:49:22.467 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 25.
  2017-02-04 00:49:22.467 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 22.
  2017-02-04 00:49:22.467 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 21.
  2017-02-04 00:49:22.467 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 24.
  2017-02-04 00:49:22.467 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 46.
  2017-02-04 00:49:22.467 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 50.
  2017-02-04 00:49:22.467 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 44.
  2017-02-04 00:49:22.468 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 42.
  2017-02-04 00:49:22.468 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 48.
  2017-02-04 00:49:22.468 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 33.
  2017-02-04 00:49:22.468 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 35.
  2017-02-04 00:49:22.468 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 32.
  2017-02-04 00:49:22.468 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 31.
  2017-02-04 00:49:22.468 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 34.
  2017-02-04 00:49:22.469 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 66.
  2017-02-04 00:49:22.469 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 62.
  2017-02-04 00:49:22.469 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 68.
  2017-02-04 00:49:22.469 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 64.
  2017-02-04 00:49:22.469 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 70.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 43.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 44.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 41.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 42.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$6 - Processing: 45.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-3] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 86.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-4] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 88.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-1] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 82.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-2] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 84.
  2017-02-04 00:49:22.470 [RxComputationThreadPool-5] [DEBUG] n.a.j.r.GroupsAndDelays.lambda$null$5 - Processing result: 90.
  Thread map: [20:[3, 13, 23, 33, 43], 21:[2, 12, 22, 32, 42], 22:[5, 15, 25, 35, 45], 23:[4, 14, 24, 34, 44], 24:[1, 11, 21, 31, 41]]
  Result thread map: [20:[6, 26, 46, 66, 86], 21:[4, 24, 44, 64, 84], 22:[10, 30, 50, 70, 90], 23:[8, 28, 48, 68, 88], 24:[2, 22, 42, 62, 82]]
  
  Process finished with exit code 0
  </code></pre>
  
  <p><strong>Edit</strong>:</p>
  
  <p>Bonus points if you can also show how to do this in <a href="https://projectreactor.io/" rel="nofollow noreferrer">project Reactor</a>.</p>
  
  <p><strong>Edit 2</strong>:
  The solution using project Reactor is <a href="https://stackoverflow.com/questions/42082334/reactive-parallelization-doesnt-work/42094037">here</a>.</p>