Asynchronous Programming and Monad Transformers in Scala

// https://github.com/twitter/finagle/blob/develop/finagle-memcached/src/main/scala/com/twitter/finagle/Memcached.scala#L89
// finagle.Client trait을 mixin 해서 interface를 구현하고 있습니다.
trait MemcachedRichClient { self: finagle.Client[Command, Response] =>

  def newRichClient(dest: Name, label: String): memcached.Client =
    newTwemcacheClient(dest, label)

  def newRichClient(dest: String): memcached.Client = {
    val (n, l) = Resolver.evalLabeled(dest)
    newTwemcacheClient(n, l)
  }

  def newTwemcacheClient(dest: Name, label: String): TwemcacheClient

  def newTwemcacheClient(dest: String): TwemcacheClient = {
    val (n, l) = Resolver.evalLabeled(dest)
    newTwemcacheClient(n, l)
  }
}

// https://github.com/twitter/finagle/blob/develop/finagle-mysql/src/main/scala/com/twitter/finagle/Mysql.scala#L18
// 위와 동일하게 finagle.Client trait를 mixin하고 이를 통해서 interface를 구현하고 있습니다.
trait MysqlRichClient { self: com.twitter.finagle.Client[Request, Result] =>

  def newRichClient(dest: Name, label: String): mysql.Client with mysql.Transactions =
    mysql.Client(newClient(dest, label))

  def newRichClient(dest: String): mysql.Client with mysql.Transactions =
    mysql.Client(newClient(dest))
}

// https://github.com/twitter/finatra/blob/develop/http/src/main/scala/com/twitter/finatra/http/internal/server/BaseHttpServer.scala

// Finatra BaseHttpserver 코드의 일부
package com.twitter.finatra.http.internal.server

// finagle.Http를 import하고 있습니다.
import com.twitter.finagle.{ListeningServer, Http, Service}

trait BaseHttpServer extends TwitterServer {
...
  // import한 finagle.Http.server를 이용해서 Http서버를 만들고 있습니다.
  private lazy val baseHttpServer: Http.Server = {
    Http.server
      .withMaxRequestSize(maxRequestSizeFlag())
      .withStreaming(streamRequest)
  }
...
}

// file: PingController.scala
package com.github.ikhoon.app.v1.ping

import javax.inject.Inject

import com.google.inject.Singleton
import com.twitter.finagle.http.Request
import com.twitter.finatra.http.Controller
import com.twitter.util.Future

case class Pong(pong: String)

@Singleton
class PingController @Inject() () extends Controller { // finatra.http.Controller를 상속 받습니다.

  // GET /ping URL을 처리하기 위한 router를 정의합니다.
  get("/ping") { request: Request =>
    Future.value(Pong("pong"))  // 비동기 프로그래밍을 위해서 `Future`로 값을 감싸서 return 합니다.
  }
}

package com.github.ikhoon

import com.github.ikhoon.app.v1.ping.PingController
import com.twitter.finagle.http.{ Request, Response }
import com.twitter.finatra.http.HttpServer
import com.twitter.finatra.http.routing.HttpRouter

object FinatraServerMain extends FinatraServer

class FinatraServer extends HttpServer {

  override def defaultFinatraHttpPort = ":9999"

  override def configureHttp(router: HttpRouter) {
    router
      .add[PingController] // Controller를 router에 등록합니다.
  }

}

abstract class BasicHttpClientModule() extends TwitterModule {

  protected def provideHttpClient(mapper: FinatraObjectMapper, host: String, port: Int = 80): HttpClient = {
    // HttpClientmodule을 상속받으면서 값을 override합니다.
    val httpClientModule = new HttpClientModule {
      override def dest: String = s"$host:$port"
      // 기본 Header값을 override 하면서 Host값을 설정합니다.
      override def defaultHeaders: Map[String, String] = Map("Host" -> host)
    }
    httpClientModule.provideHttpClient(mapper, httpClientModule.provideHttpService)
  }
}

import com.twitter.finatra.json.FinatraObjectMapper
import com.typesafe.config.Config
import net.ceedubs.ficus.Ficus._
import com.twitter.finatra.httpclient.modules.HttpClientModule
object FakeHttpClientModule {
  def apply() = new BasicHttpClientModule {
    // `fake`란 이름으로 의존성 주입될수 있도록 이름을 지어줌
    @Named("fake") @Provides @Singleton
    // jackson object mapper와 typesafe config를 의존성을 주입함
    def provideHttpClient(mapper: FinatraObjectMapper, config: Config) = {
      // 위에서 만들었던 `BasicHttpClientModule#provideHttpClient`를 호출하면서
      // typesafe config에 설정되어 있는 fake.host, fake.port값을 가져와 넘겨줌
      super.provideHttpClient(mapper,
        config.as[String]("fake.host"), config.as[Int]("fake.port"))
    }
  }
}

class FinatraServer extends HttpServer {
  override def modules = Seq(FakeHttpClientModule())
  ...
}

import javax.inject.{ Inject, Named }

import com.fasterxml.jackson.databind.JsonNode
import com.twitter.finatra.httpclient.{ HttpClient, RequestBuilder }
import com.twitter.util.Future
import com.typesafe.config.Config
import net.ceedubs.ficus.Ficus._

// @Named("fake") 어노테이션을 이용하여 위에서 정의한 HttpClient의 의존성을 주입합니다.
class FakeService @Inject() (@Named("fake") httpClient: HttpClient, config: Config) {
  def withSleepAsync(sec: Int): Future[JsonNode] = {
    val url = config.as[String]("fake.host") + s"?sleep=$sec"
    httpClient.executeJson[JsonNode](RequestBuilder.get(url))
  }

  def withSleepSync(sec: Int): JsonNode = {
    val url = config.as[String]("fake.host") + s"?sleep=$sec"
    val jsonNode = httpClient.executeJson[JsonNode](RequestBuilder.get(url))
    // 의도적으로 Blocking IO를 유발함
    Await.result(jsonNode, Duration.fromSeconds(100))
  }
}

import javax.inject.Inject

import com.github.ikhoon.swagger.SimpleSwaggerSupport
import com.twitter.finagle.http.Request
import com.twitter.finatra.http.Controller


class FakeController @Inject() (fakeService: FakeService) extends Controller {
  get("/sleep/:id/async") { request: Request =>
    fakeService.withSleepAsync(request.getIntParam("id"))
  }

  get("/sleep/:id/sync") { request: Request =>
    fakeService.withSleepSync(request.getIntParam("id"))
  }
}

class FinatraServer extends HttpServer {
  override def modules = Seq(FakeHttpClientModule())
  ...
  override def configureHttp(router: HttpRouter) {
    router
      .add[FakeController]  // FakeController를 라우터에 등록

}

// 주문내역을 가져오는 API
case class Order(id: Int, userId: Long, addr: String)
case class Item(id: Int, name: String, price: Long)

val orders = Map(
  1 -> List(Order(1, 1, "Seoul"), Order(2, 1, "Seoul")),
  2 -> List(Order(3, 2, "Tokyo"), Order(4, 2, "Tokyo")),
  3 -> List(Order(5, 3, "NewYork"),Order(6, 3, "NewYork")),
  4 -> List(Order(7, 4, "Daegu"), Order(8, 4, "Daegu"))
)

val orderItems = Map(
  1 -> List(Item(1, "I1", 2000), Item(2, "I2", 4000)),
  2 -> List(Item(2, "I2", 4000), Item(3, "I3", 6000), Item(4, "I4", 8000)),
  3 -> List(Item(3, "I3", 6000), Item(4, "I4", 8000), Item(5, "I5", 10000)),
  4 -> List(Item(4, "I4", 4000), Item(5, "I5", 10000), Item(6, "I6", 12000)),
  5 -> List(Item(1, "I1", 2000), Item(2, "I2", 4000), Item(3, "I3", 6000)),
  6 -> List(Item(2, "I2", 4000), Item(4, "I4", 8000)),
  7 -> List(Item(3, "I3", 6000), Item(4, "I4", 8000), Item(5, "I5", 10000), Item(6, "I6", 12000)),
  8 -> List(Item(4, "I4", 4000), Item(5, "I5", 10000), Item(7, "I7", 14000))
)

// 주문내역을 가져오는 API
def getOrders(userId: Int): Future[List[Order]] = Future.successful(orders.get(userId).get)
// 주문의 상품 내역을 가져오는 API
def getOrderItems(order: Order): Future[List[Item]] = Future.successful(orderItems.get(order.id).get)

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

val userId = 1
val orderItems = for {
  // 비동기로 유저의 상품내역을 가져옴
  orderList: List[Order] <- getOrders(userId)
  // 주문 상품내역을 추출하기 위해서는 List에서 Order값을 가져와야함
  itemList: List[Item] <- {
    val itemListFutureList: List[Future[List[Item]]] = for {
        order <- orderList
      } yield getOrderItems(order)
    // itemListFutureList가 List안에 Future가 들어있기 때문에 Future의 sequence 함수를 통해서 순서를 변경합니다.
    val itemFutureListList: Future[List[List[Item]]] = Future.sequence(itemListFutureList)
    // 변경된 Future안에 List가 중첩해서 들어있기 때문에 flatten 함수를 이용해서 이를 없애줍니다.
    val itemFutureList: Future[List[Item]] = itemFutureListList.map(_.flatten)
    // 이제 itemFutureList는 상위 for comprehension을 돌수있는 monad의 형태 Future[M[T]]의 형태가 되었습니다.
    itemFutureList
  }
} yield (itemList)
orderItems.foreach(println)
// List(Item(1,I1,2000), Item(2,I2,4000), Item(2,I2,4000), Item(3,I3,6000), Item(4,I4,8000))

case class FutureSeq[+A](future: Future[Seq[A]]) {
  def flatMap[B](f: A => FutureSeq[B])(implicit ec: ExecutionContext): FutureSeq[B] =
    FutureSeq(future.flatMap {
      case a => Future.sequence(a.map(f andThen (_.future))).map(_.flatten)
    })

  def map[B](f: A => B)(implicit ec: ExecutionContext): FutureSeq[B] = FutureSeq(future.map(_.map(f)))

  def filter(p: A ⇒ Boolean)(implicit ec: ExecutionContext): FutureSeq[A] = withFilter(p)(ec)

  def withFilter(p: A ⇒ Boolean)(implicit ec: ExecutionContext): FutureSeq[A] =
    FutureSeq(future.map(_.filter(p)))
}

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
val orderItems = for {
  order: Order <- FutureSeq(getOrders(userId))
  item: Item <- FutureSeq(getOrderItems(order))
} yield item
orderItems.future.foreach(println)
// List(Item(1,I1,2000), Item(2,I2,4000), Item(2,I2,4000), Item(3,I3,6000), Item(4,I4,8000))

import scalaz._
import Scalaz._
import ListT._
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
val orderItems = for {
  order: Order <- listT(getOrders(userId))
  item: Item <- listT(getOrderItems(order))
} yield item
orderItems.run.foreach(println)
// List(Item(1,I1,2000), Item(2,I2,4000), Item(2,I2,4000), Item(3,I3,6000), Item(4,I4,8000))

import javax.inject.{ Inject, Singleton }
import java.util.Date
import com.github.ikhoon.modules.QuillDatabaseModule.QuillDatabaseSource
import com.twitter.util.Future
import io.getquill._

// Users 테이블의 모델 정의
case class Users(id: Int, name: String, email: String, createdAt: Date)

@Singleton
class QuillUserRepository @Inject() (db: QuillDatabaseSource) {

  // Email을 가지고 user 정보를 가져오는 쿼리 수행
  def findByEmail(email: String): Future[Option[Users]] = {
    val q = quote { (email: String) =>
      query[Users].filter(i => i.email == email).take(1)
    }
    db.run(q)(email).map(_.headOption)
  }
}

import javax.inject.{ Inject, Singleton }
import java.util.Date
import com.github.ikhoon.modules.QuillDatabaseModule._
import com.twitter.util.Future
import io.getquill._

// Points 테이블의 모델 정의
case class Points(id: Int, userId: Int, point: Int, createdAt: Date)

@Singleton
class QuillPointRepository @Inject() (db: QuillDatabaseSource) {

  // Uesr ID를 가지고 포인트 정보를 가져오는 쿼리수행
  def findByUserId(userId: Int): Future[Option[Points]] = {
    val q = quote { (userId: Int) =>
      query[Points].filter(p => p.userId == userId).take(1)
    }
    db.run(q)(userId).map(_.headOption)
  }

}

class UserPointService @Inject() (
  quillUserRepository:  QuillUserRepository,
  quillPointRepository: QuillPointRepository
) {
  def getPointByUserEmail(email: String): Future[Option[Points]] =
    for {
      userOption: Option[Users] <- quillUserRepository.findByEmail(email)
      // option 타입이기 때문에 quillPointRepository.findByUserid 를 호출할수 없기 때문에 pattern matching 을 통해서 이부분을 해결합니다.
      pointOption: Option[Points] <- userOption match {
        case Some(user) => quillPointRepository.findByUserId(user.id)
        case _          => Future.value(None)
      }
    } yield pointOption

case class FutureOption[+A](future: Future[Option[A]]) extends AnyVal {
  def flatMap[B](f: A => FutureOption[B])(implicit ec: ExecutionContext): FutureOption[B] =
    FutureOption(future.flatMap {
      case Some(a) => f(a).future
      case None    => Future.value(None)
    })

  def map[B](f: A => B)(implicit ec: ExecutionContext): FutureOption[B] = FutureOption(future.map(_.map(f)))

  def filter(p: A ⇒ Boolean)(implicit ec: ExecutionContext): FutureOption[A] = withFilter(p)(ec)

  def withFilter(p: A ⇒ Boolean)(implicit ec: ExecutionContext): FutureOption[A] =
    FutureOption(future.map {
      case Some(a) => if (p(a)) Some(a) else None
      case _       => None
    })
}

def getPointByUserEmail(email: String): Future[Option[Points]] =
  (for {
    user <- FutureOption(quillUserRepository.findByEmail(email))
    point <- FutureOption(quillPointRepository.findByUserId(user.id))
  } yield point).future

import scalaz._
import Scalaz._
import OptionT._
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
def getPointByUserEmail(email: String): Future[Option[Points]] =
  (for {
    user <- optionT(quillUserRepository.findByEmail(email))
    point <- optionT(quillPointRepository.findByUserId(user.id))
  } yield point).run