# Flow using itself recursively and merge all contents

**URL:** <https://discuss.akka.io/t/flow-using-itself-recursively-and-merge-all-contents/813>\
**Category:** Akka Libraries\
**Created:** [April 24, 2018, 3:16pm UTC](https://discuss.akka.io/t/flow-using-itself-recursively-and-merge-all-contents/813 "2018-04-24T15:16:24Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![KannarFr](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/kannarfr/32/1581_2.png) [@KannarFr](https://discuss.akka.io/u/KannarFr)\
**Post date:** [April 24, 2018, 3:16pm UTC](https://discuss.akka.io/t/flow-using-itself-recursively-and-merge-all-contents/813/1 "2018-04-24T15:16:24Z")

</div>

Hi,

I’m trying to fetch data from GTS database. Consider that this database provides fetch only with using now (which is a timestamp) and n which is the records number limit per request. And in this example I don’t use the limit param for the global fetch yet. But I have a 3 gts batch per request which is a test.

So, I want to chain fetch using batch and from the last fetched batch retrieve the oldest date to fetch the next using the now/limit pattern. I’m trying to do all this stuff using akka stream recursively.

So here is the code:

```auto
  def rangedFetch(
    readToken: String,
    query: Query[StartStopRangeMicros],
    limit: Long
  )(
    implicit warpClientContext: WarpClientContext
  ): Flow[Query[StartStopRangeMicros], Seq[GTS], NotUsed] = {
    val uuid = UUID.randomUUID
    Flow[Query[StartStopRangeMicros]]
      .map(query => query.copy(range = RecordsSinceMicros(query.range.newestDate, 3)))
      .map(query => fetchRequest(readToken, query))
      .map(request => (request -> uuid)) // cf. https://doc.akka.io/docs/akka-http/current/client-side/host-level.html
      .via(warpClientContext.poolClientFlow)
      .filter({ case (_, key) => key == uuid })
      .map({ case (responseTry, _) => responseTry })
      .via(processResponseTry)
      .via(byteStringToGTS)
      .fold(Seq[GTS]()) { case (seq, item) => { seq :+ item }}
      .flatMapConcat { gtsSeq: Seq[GTS] =>
        val currentOldestGTS = gtsSeq.last
        val chainedNow = currentOldestGTS.ts.get
        if (chainedNow > query.range.oldestDate) {
          val newQuery = query.copy(range = StartStopRangeMicros(chainedNow, query.range.oldestDate))
          Source.combine(
            Source
              .single(gtsSeq),
            Source
              .single(newQuery)
              .via(rangedFetch(readToken, newQuery, limit))
          )(Concat(_))
        } else {
          Source
            .single(gtsSeq)
        }
      }
  }

```

The problem is about getting all the content and not only the last result. And using this Combine make the method providing only the last fetch content. Can’t understand why.

Maybe the way how I do this job isn’t the best and I’m open to every possibility…

---

<div class="post-metadata">

**Author:** ![KannarFr](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/kannarfr/32/1581_2.png) [@KannarFr](https://discuss.akka.io/u/KannarFr)\
**Post date:** [April 24, 2018, 3:46pm UTC](https://discuss.akka.io/t/flow-using-itself-recursively-and-merge-all-contents/813/2 "2018-04-24T15:46:27Z")

</div>

This is working, but this is so ugly.

```auto
 if (chainedNow > query.range.oldestDate) {
          val newQuery = query.copy(range = StartStopRangeMicros(chainedNow, query.range.oldestDate))
          println(newQuery)
          Source
            .single(newQuery)
            .via(rangedFetch(readToken, newQuery, limit))
            .flatMapConcat(gtsSeq2 =>
              Source
                .single(gtsSeq ++ gtsSeq2)
            )
        } else {
          Source
            .single(gtsSeq)
        }

```

---

<div class="post-metadata">

**Author:** ![KannarFr](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/kannarfr/32/1581_2.png) [@KannarFr](https://discuss.akka.io/u/KannarFr)\
**Post date:** [April 25, 2018, 11:21am UTC](https://discuss.akka.io/t/flow-using-itself-recursively-and-merge-all-contents/813/3 "2018-04-25T11:21:25Z")

</div>

This way is only running the Source concat at the end of all rangedFetch calls so this way is bad because the limit can only be performed after all the fetch and not during the fetches (one by one).