Paginated scraper flow

gistfile1.scala

def scrape(implicit charset: Charset): Flow[URL, Document, Unit] = Flow[URL].mapAsyncUnordered(10)(httpClient.getDocument(charset))
    .buffer(10, OverflowStrategy.backpressure)
    .map(toDocument)

def scrapePaginated[T](f: Document => (T, Option[URL]))(implicit charset: Charset) = Flow() { implicit b =>
    import akka.stream.scaladsl.FlowGraph.Implicits._

    val merge = b.add(MergePreferred[URL](1))
    val unzip = b.add(Unzip[T, Option[URL]]())

    merge ~> scrape(charset).map(f) ~> unzip.in
    unzip.out1.filter(_.isDefined).map(_.get) ~> Flow[URL].buffer(100, OverflowStrategy.fail) ~> merge.preferred

    (merge.in(0), unzip.out0)
  }