Fuse multiple requests into a single publisher

gistfile1.scala

object RequestsManager {
  case class Start(requests: () => DatabasePublisher[Request])
  case class OnError(t: Throwable)
  case object OnComplete
}

class RequestsManager() extends Actor with ActorPublisher[Request] {
  import RequestsManager._

  override def receive: Receive = {
    case RequestsManager.Start(requests) =>
      swapInNewRequests(requests)
  }

  def swapInNewRequests(requests: () => DatabasePublisher[Request], outstanding: Long = 0L): Unit = {
    requests().subscribe(new Subscriber[Request] {
      override def onError(t: Throwable) =
        self ! OnError(t)

      override def onSubscribe(s: Subscription) =
        context become forwardWithSubscription(s, requests, outstanding)

      override def onComplete() =
        self ! OnComplete

      override def onNext(r: Request) =
        self ! r
    })
  }
  
  def forwardWithSubscription(subscription: Subscription,
                              requests: () => DatabasePublisher[Request],
                              outstanding: Long): Receive =
  {
    var _outstanding = outstanding

    {
      case OnError(t) =>
        onError(t)
      case OnComplete =>
        swapInNewRequests(requests, _outstanding)
      case r : Request =>
        _outstanding = _outstanding - 1
        onNext(r)
      case ActorPublisherMessage.Request(items) =>
        _outstanding += items
        subscription.request(items)
      case ActorPublisherMessage.Cancel =>
        subscription.cancel()
    }
  }
}