akka/akka

In the context of akka/akka-stream-contrib#93

If I do not know the initial page token, how to use PagedSource (or any other source operator)? (I like PagedSource, it is clean approach :) )

For example, our 3rd party REST API provides a token after first invocation, with which we need to use to iterate it until the token is null / None. I could not come up with any nice solution apart from the following (and Option[Option[T]]), do you have any suggestions?

I feel this is common use case, do you have already something in the library for this use case?

case class Page[T, K](items: immutable.Iterable[T], nextKey: Option[K])
final class PagedSourceStageAsync[T, K](initial: Option[K], f: Option[K] => Future[Page[T, K]])
    extends GraphStage[SourceShape[T]] {
  import scala.concurrent.ExecutionContext.Implicits.global
  val outlet: Outlet[T] = Outlet("PagedSourceStageAsync.out")
  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) with OutHandler {
      private[this] var state: Option[K] = initial
      private[this] var asyncHandler: Try[Page[T, K]] => Unit = _

      override def preStart(): Unit = {
        val ac = getAsyncCallback[Try[Page[T, K]]] {
          case Failure(ex) => fail(outlet, ex)
          case Success(Page(items, None)) =>
            push(outlet, items)
            complete(outlet)
          case Success(Page(items, Some(nextKey))) =>
            push(outlet, items)
            state = Some(nextKey)
        }
        asyncHandler = ac.invoke
      }

      def onPull(): Unit = f(state).onComplete(asyncHandler)

      setHandler(outlet, this)
    }

  override def shape: SourceShape[T] = SourceShape(outlet)
}