Разбавляем асинхронное программирование функциональным на Scala

// Про 'FState' - далее, пока же просто примем, что это - такая необычная Future
def getNextConfig: FState[Config]
def getTemperature(from: String): FState[Int]

case class State(temperature: Int, sumTemp: Long, count: Int) {
  def isGood = ...
}

// Как видим, получается единый асинхронный алгоритм с состоянием, 
// которое извне этого алгоритма не видно
val handle = 
  while_ ( _.isGood)
  {  for (
        config <- getNextConfig();
        if (config.isDefined);  // пустой конфиг - прекращаем выполнение
        nextValue <- getTemperature(config().source);  // грузим значение температуры
        state <- gets[State];  // тут мы берем текущее состояние
        newState = State(nextValue, state.sumTemp + nextValue, state.count + 1);
        _ <- puts(newState);  // .. и меняем его
        _ <- runInUiThread { drawOnScreen(newState) }
  ) yield() }

val configs: AsyncStream[Config] = ... // получаем откуда-то stream конфигов

def getTemperature(from: String): FState[Int]

case class State(temperature: Int, sumTemp: Long, count: Int)

// Получается то же самое, только вместо зависимости 'getNextConfig'
// мы, по сути, передаем сами данные - stream из конфигов
val handle = 
  foreach(configs) {
    config => for (
        nextValue <- getTemperature(config().source);  // грузим значение температуры
        state <- gets[State];  // тут мы берем текущее состояние
        newState = State(nextValue, state.sumTemp + nextValue, state.count + 1);
        _ <- puts(newState);  // .. и меняем его
        _ <- runInUiThread { drawOnScreen(newState) }
    ) yield()  
  }

class FStateMonad[S] extends Monad[({ type f[X] = FState[S, X]})#f] {
  type F[X] = FState[S, X]

  override def point[A](a: => A): F[A] = FState((s: S) => Future((a, s)))
  override def bind[A, B](m: F[A])(f: A => F[B]): F[B] =
    FState((s: S) => m(s) flatMap { pair => f(pair._1)(pair._2) })
}

// Как видим, чтобы нам получить текущее состояние, его достаточно 
// просто "поднять" в значение.
def gets[S](): FState[S, S] = FState((s: S) => Future((s, s)))

// А изменить состояние еще проще
def puts[S](news: S): FState[S, S] = FState((_: S) => Future((news, news)))

def foldLeft[B](start: B)(f: (B, A) => B): Future[B] = {
  // Стандартно, определим функцию с аккумулятором, которой
  // потом передадим data
  def impl(d: Future[Pair[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
    d flatMap (pair => {
      if (pair eq null) acc
      else impl(pair.second.data, acc map (b => f(b, pair.first)))
    })

   impl(data, Future(start))
}

def flatten : Future[List[A]] =
    foldLeft[List[A]](Nil)((list, el) => el :: list) map (_.reverse)

def takeWhile(p: A => Boolean): AsyncStream[A] =
  new AsyncStream[A](data map (pair => {
     if (pair eq null) null
     else if (!p(pair.first)) null
     else Pair(pair.first, pair.second.takeWhile(p))
   }))

def take(n: Int): AsyncStream[A] =
    if (n <= 0) nil
    else new AsyncStream[A](data map (pair => {
      if (pair eq null) null
      else Pair(pair.first, pair.second.take(n - 1))
    }))

def genAsyncStream[S,A](start: S)(gen: S => Future[(A, S)]): AsyncStream[A] =
  new AsyncStream[A](
    gen(start) match {
      case _: NoFuture => Future(null)
      case future => future map (pair => { // Future[Pair[A, AsyncStream]]
        if (pair eq null) null
        else Pair(pair._1, genAsyncStream(pair._2)(gen))
      })})

def concat[A](s1: AsyncStream[A], s2: AsyncStream[A]): AsyncStream[A] =
  new AsyncStream[A](s1.data flatMap (pair => {
    if (pair eq null) s2.data
    else Future(Pair(pair.first, concat(pair.second, s2)))
  }))

class AsyncStreamMonad extends Monad[AsyncStream] {
  override def point[A](a: => A): AsyncStream[A] = unit(a)
  override def bind[A, B](
    ma: AsyncStream[A])(f: A => AsyncStream[B]): AsyncStream[B] =
    new AsyncStream[B](ma.data flatMap (pair => {
      if (pair eq null) Future(null)
      else f(pair.first).data map (
        pair2 => Pair(pair2.first, concat(pair2.second, bind(pair.second)(f))))
    }))
}

// Сделаем это foldLeft-ом
def foreach[A, S]
  (stream: AsyncStream[A])(f: A => FState[S, Any]): FState[S, Unit] =
  FState(s => {
    stream.foldLeft(Future(s))(
    (futureS, a) => futureS.flatMap(s2 => f(a)(s2).map(_._2))).flatten.map( ((), _) )
  })