withdraw_service_story_13.scala

class WithdrawBatchingExecutor(repository: WithdrawalRepository, 
                                 externalSystem: ExternalWithdrawSystem) {

  def startExecutor: Observable[Unit] =
    Observable
      .intervalWithFixedDelay(config.interval)
      .mapEval( _ => repository.findAllWaitingOrderedBySerialIdAsc())
      .distinctUntilChanged
      .scan(List.empty[Withdrawal]) { (oldSet, newSet) =>
        newSet.takeWhile(w => !oldSet.contains(w))
      }
      .flatMap(Observable.fromIterable)
      .groupBy(_.currency)
      .mergeMap { partition =>
        partition
          .bufferTimedAndCounted(config.timeWindow, config.batchSizeThreshold)
          .mapEval(batch => externalSystem.executeWithdrawals(batch))
      }
}