scala future

Await.scala

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.concurrent.Await
import scala.concurrent.duration._
import scala.util.{Success, Failure, Try}

val x = Future { 10 }
val hMilliSec = Duration(100, MILLISECONDS)
val y = Await.result(x, hMilliSec)
val z = Await.result(x, 100.milliseconds)

// onComplete을 쓰면 return type이 Unit
scala> val k = x.onComplete { i =>
     |   i match {
     |     case Success(n) => println(s"Successfully returned $n")
     |     case Failure(e) => println(s"Failed because of $e")
     |   }
     | }
val k: Unit = ()
Successfully returned 10

// Try 역시 return type이 Unit
scala> val k = Try(Await.result(x, 100.milliseconds)) match {
     |   case Success(n) => println(s"Successfully returned $n")
     |   case Failure(e) => println(s"Failed because of $e")
     | }
Successfully returned 10

// map을 쓰면 그냥 그대로 Future[T]
scala> val k = x.map { i => i }
val k: scala.concurrent.Future[Int] = Future(<not completed>)

scala> val l = Await.result(k, 100.milliseconds)
val l: Int = 10
val k: Unit = ()

// flatMap을 쓰면 그대로 넘길 수 없고 Future로 감싸줘야 함
scala> val k = x.flatMap { i => i }
                                ^
       error: type mismatch;
        found   : Int
        required: scala.concurrent.Future[?]

scala> val k = x.flatMap { i => Future(i) }
val k: scala.concurrent.Future[Int] = Future(<not completed>)

scala> val l = Await.result(k, 100.milliseconds)
val l: Int = 10

// for를 쓰면 역시 map과 마찬가지로 Future[T]
scala> val k = for ( i <- x ) yield i
val k: scala.concurrent.Future[Int] = Future(<not completed>)

scala> val l = Await.result(k, 100.milliseconds)
val l: Int = 10

// Promise와 Future
scala> :paste
// Entering paste mode (ctrl-D to finish)

val f = Future { 1 }
val p = Promise[Int]()

p completeWith f

p.future foreach { x =>
  println(x)
}

1
val f: scala.concurrent.Future[Int] = Future(Success(1))
val p: scala.concurrent.Promise[Int] = Future(Success(1))

scala> p
val res5: scala.concurrent.Promise[Int] = Future(Success(1))

scala> Await.result(f, 100.milliseconds)
val res6: Int = 1

scala> Await.result(p, 100.milliseconds)
                    ^
       error: type mismatch;
        found   : scala.concurrent.Promise[Int]
        required: scala.concurrent.Awaitable[?]

scala> Await.result(p.future, 100.milliseconds)
val res8: Int = 1

// andThen
scala> implicit val ec: scala.concurrent.ExecutionContext = scala.concurrent.ExecutionContext.global
val ec: scala.concurrent.ExecutionContext = scala.concurrent.impl.ExecutionContextImpl$$anon$3@3466cba1[Running, parallelism = 16, size = 1, active = 0, running = 0, steals = 8, tasks = 0, submissions = 0]

scala> val y = x andThen {
     |   case Success(n) => n
     |   case Failure(e) => None
     | }
val y: scala.concurrent.Future[Int] = Future(<not completed>)

scala> y
val res13: scala.concurrent.Future[Int] = Future(Success(10))

Future[T]-to-T.md

• 상황

  • 이미 Boolean으로 조건을 판단하는데 새로운 조건을 넣어야 하는데 이 새로운 조건을 점검할 때 써야하는 method(아래 코드의 getLanguageId)가 Future[T]
  • Future[T]를 가져와 비교하려고 하는데
    • Await.result는 Duration을 넣어야 하므로 제외
    • Future.result도 Duration이 필요하므로 역시 제외
    • Future.onComplete은 Unit이어서 반환을 안 함
    • Promise를 사용해서 Future.onComplete에서 결과를 저장할 수는 있으나(completeWith 사용) 이건 여전히 Future[T]를 반환

• 해결

  • Future.value를 사용해 Option[Try[T]]를 반환(https://www.scala-lang.org/api/2.9.3/scala/concurrent/Future.html)
  • isDefined로 Option을 제거하고 case Success로 Try를 제거해서 값을 가져옴

• 하지만 .value를 쓰면 Future가 시작되기도 전에 값을 읽어오려고 시도해서 langId에 None이 할당되고, .isDefined가 항상 false로만 return되어 원하는 대로 코드가 진행되지 않을 가능성이 높다는 이야기를 들음

  • https://lascala.slack.com/archives/C11CB042G/p1598424799008900
  • return type을 Future[T]를 사용하거나 T를 그냥 사용하려면 blocking이 발생할 thread pool을 잘 관리해야 함
  • 애플리케이션 전체가 하나의 IO값이 되지 않으면 결국 어느 시점에서는 Future[T]를 T로 바꾸거나, 전체를 수정해서 하나의 IO값이 되도록 해야 함(예를 들어 하스켈의 main 함수는 IO() 타입)

• example

  def isCNRougeEmailAndMemberIdInExpB(email: IrisEmail): Boolean = {
    val langId: Option[Try[Int]] = languageDetectorClient.getLanguageID(email).map {
      langInfo => if ( langInfo.languageId.nonEmpty ) langInfo.languageId.get else NoReplyDefaultLangId
    }.value
    if ( langId.isDefined ) {
      langId.get match {
          case Success(id) =>
            if (id == NoReplyCNLangId) {
              email.memberId match {
                case Some(memberId) =>
                  experimentService.determineVariant(Experiment.UNBLOCK_CN_ROGUE_EMAILS, memberId.toString) == Variant.B
                case _ =>
                  false
              }
            } else
              false
          case Failure(_) => false
        }
    } else
      false
  }
  

Future_flatten.scala

@ import scala.concurrent.Future

// String을 받아 Future[Option[Long]]을 반환하는 test function
@ def foo(s: String): Future[Option[Long]] = {
    if ( s == "b" )
      Future.successful(None)
    else
      Future.successful(Some(1L))
  }
defined function foo

@ val s1 = Seq("a", "b", "c")
s1: Seq[String] = List("a", "b", "c")

@ val s2 = Seq("b", "b", "b")
s2: Seq[String] = List("b", "b", "b")

// s1, s2에 foo를 호출하면 List of Future가 만들어지는 데 이 List of Future를 Future of List로 바꾸고 싶음
@ s1.map(foo(_))
res39: Seq[Future[Option[Long]]] = List(Future(Success(Some(1))), Future(Success(None)), Future(Success(Some(1))))

@ s2.map(foo(_))
res40: Seq[Future[Option[Long]]] = List(Future(Success(None)), Future(Success(None)), Future(Success(None)))

// Future.sequence를 이용해 List of Future에 map을 사용하고, flatten으로 List에서 None을 제거하면 List가 empty이거나 아니거나를 알 수 있음
@ Future.sequence(s1.map(foo(_))).map { m =>
    m match {
      case l if l.flatten.length > 0 => true
      case _ => false
    }
  }
res41: Future[Boolean] = Future(Success(true))

@ Future.sequence(s2.map(foo(_))).map { m =>
    m match {
      case l if l.flatten.length > 0 => true
      case _ => false
    }
  }
res42: Future[Boolean] = Future(Success(false))

future_fold.scala

// Vector[Boolean]
@ Vector(false, true, false).fold(false)((acc, e) => acc | e)
res400: Boolean = true

@ Vector(false, false, false).fold(false)((acc, e) => acc | e)
res401: Boolean = false

// Future[Vector[Boolean]]
@ Future.successful(Vector(false, true, false)).map(_.fold(false)((acc, e) => acc | e))
res402: Future[Boolean] = Future(Success(true))

@ Future.successful(Vector(false, false, false)).map(_.fold(false)((acc, e) => acc | e))
res403: Future[Boolean] = Future(Success(false))

// Vector[(Int, Boolean)]
@ Vector((2, false), (4, true), (7, false)).foldLeft(0)((acc, e) => acc + e._1)
res415: Int = 13

@ Vector((2, false), (4, true), (7, false)).fold(false)((acc, e) => acc | e._2)
cmd416.sc:1: value | is not a member of Any
val res416 = Vector((2, false), (4, true), (7, false)).fold(false)((acc, e) => acc | e._2)
                                                                                   ^
cmd416.sc:1: value _2 is not a member of Any
val res416 = Vector((2, false), (4, true), (7, false)).fold(false)((acc, e) => acc | e._2)
                                                                                       ^
Compilation Failed

@ Vector((2, false), (4, true), (7, false)).foldLeft(false)((acc, e) => acc | e._2)
res416: Boolean = true

// Future[Vector[(Int, Boolean)]]
@ Future.successful(Vector((2, false), (4, true), (7, false))).map(_.foldLeft(false)((acc, e) => acc | e._2))
res417: Future[Boolean] = Future(Success(true))

// Vector[Future[Boolean]]
@ val l = Vector(Future.successful(false), Future.successful(true), Future.successful(false))
l: Vector[Future[Boolean]] = Vector(Future(Success(false)), Future(Success(true)), Future(Success(false)))

@ Future.fold(l)(false)((acc, e) => acc | e)
res394: Future[Boolean] = Future(Success(true))

multiple_future.scala

import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

implicit val ec = scala.concurrent.ExecutionContext.global

val a = Future { 10 }
val b = Future { 5 }
val c = Future { 2 }

val e = for {
  i <- a
  j <- b
  k <- c
} yield (i * j * k)
println(Await.result(e, 100.milliseconds))

val f = for { ((i, j), k) <- a zip b zip c } yield (i * j * k)
println(Await.result(f, 100.milliseconds))

val g = a.map { i =>
  b.map { j =>
    c.map { k =>
      i * j * k
    }
  }
}
println(Await.result(g, 100.milliseconds))

val h = a.flatMap { i =>
  b.flatMap { j =>
    //c.flatMap { k =>
    // Future(i * j * k)
    c.map { k =>
     i * j * k
    }
  }
}
println(Await.result(h, 100.milliseconds))

// two flatMaps combination
def foo(i: Int): Future[Option[String]] =
  if (i % 2 == 0)
    Future(Some(s"Even $i"))
  else
    Future(None)

def bar(a: Int, b: Int) =
  foo(a).flatMap { a =>
    foo(b).flatMap { b =>
      if ( a.isDefined && b.isDefined )
        Future(Some(s"$a $b"))
      else {
        if ( a.isDefined )
          Future(Some(s"$a defined only"))
        else if (b.isDefined)
          Future(Some(s"$b defined only"))
        else
          Future(Some("Nothing defined"))
      }
    }
  }

Await.result(bar(2, 4), 100.milliseconds)
Await.result(bar(2, 3), 100.milliseconds)
Await.result(bar(3, 2), 100.milliseconds)
Await.result(bar(3, 3), 100.milliseconds)

// 위랑 같은 예제를 Option[String]에 대해서
def foo(s: Option[String]): Future[Option[String]] =
  if (s.isDefined) {
    if (s.get.isEmpty)
      Future(Some(""))
    else
      Future(Some(s"Normal $s"))
  } else
    Future(None)

def bar(a: Option[String], b: Option[String]) =
  foo(a).flatMap { a =>
    foo(b).flatMap { b =>
      if ( a.isDefined && b.isEmpty )
        Future(Some(s"${a.get}"))
      else {
        Future(Some("Nothing defined"))
      }
    }
  }

Await.result(bar(Some("string"), None), 100.milliseconds)
Await.result(bar(Some("string"), Some("x")), 100.milliseconds)
Await.result(bar(None, None), 100.milliseconds)
Await.result(bar(None, Some("x")), 100.milliseconds)