How to page REST calls in a future with Dispatch and Scala

Question

I use Scala and Dispatch to get JSON from a paged REST API. The reason I use futures with Dispatch here is because I want to execute the calls to fetchIssuesByFile in parallel, because that function could result in many REST calls for every lookupId (1x findComponentKeyByRest, n x fetchIssuePage, where n is the number of pages yielded by the REST API).

Here's my code so far:

def fetchIssuePage(componentKey: String, pageIndex: Int): Future[json4s.JValue]

def extractPagingInfo(json: org.json4s.JValue): PagingInfo

def extractIssues(json: org.json4s.JValue): Seq[Issue]

def findAllIssuesByRest(componentKey: String): Future[Seq[Issue]] = {
  Future {
    var paging = PagingInfo(pageIndex = 0, pages = 0)
    val allIssues = mutable.ListBuffer[Issue]()

    do {
      fetchIssuePage(componentKey, paging.pageIndex) onComplete {
        case Success(json) =>
          allIssues ++= extractIssues(json)
          paging = extractPagingInfo(json)
        case _ => //TODO error handling
      }
    } while (paging.pageIndex < paging.pages)

    allIssues // (1)
  }
}

def findComponentKeyByRest(lookupId: String): Future[Option[String]]

def fetchIssuesByFile(lookupId: String): Future[(String, Option[Seq[Issue]])] =
  findComponentKeyByRest(lookupId).flatMap {
    case Some(key) => findAllIssuesByRest(key).map(
      issues => // (2) process issues
    )
    case None => //TODO error handling
  }

The actual problem is that I never get the collected issues from findAllIssuesByRest (1) (i.e., the sequence of issues is always empty) when I try to process them at (2). Any ideas? Also, the pagination code isn't very functional, so I'm also open to ideas on how to improve this with Scala.

Any help is much appreciated.

Thanks, Michael

Answer 1

I think you could do something like:

def findAllIssuesByRest(componentKey: String): Future[Seq[Issue]] = 
  // fetch first page
  fetchIssuePage(componentKey, 0).flatMap { json =>
    val nbPages = extractPagingInfo(json).pages // get the total number of pages
    val firstIssues = extractIssues(json)       // the first issues

    // get the other pages
    Future.sequence((1 to nbPages).map(page => fetchIssuePage(componentKey, page)))
      // get the issues from the other pages
      .map(pages => pages.map(extractIssues))
      // combine first issues with other issues
      .flatMap(issues => (firstIssues +: issues).flatten)
  }

Answer 2

That's because fetchIssuePage returns a Future that the code doesn't await the result for.

Solution would be to build up a Seq of Futures from the fetchIssuePage calls. Then Future.sequence the Seq to produce a single future. Return this instead. This future will fire when they're all complete, ready for your flatMap code.

Update: Although Michael understood the above well (see comments), I thought I'd put in a much simplified code for the benefit of other readers, just to illustrate the point:

def fetch(n: Int): Future[Int] = Future { n+1 }

val fetches = Seq(1, 2, 3).map(fetch)
// a simplified version of your while loop, just to illustrate the point

Future.sequence(fetches).flatMap(results => ...)
// where results will be Seq[Int] - i.e., 2, 3, 4