http4s, Service Executor and Mongodb: Как подождать окончания insertOne

Question

Извините заранее за основной вопрос. Я начинаю изучать Scala с http4s и в обработчике маршрутизатора, я пытаюсь ввести запись в MongoDB. Насколько я могу судить insertOne возвращает Observable[Completed].

Любая идея, как я могу дождаться завершения наблюдения, прежде чем возвращать ответ?

Мой код:

class Routes { 
    val service: HttpService = HttpService { 
     case r @ GET -> Root/"hello" => { 
      val mongoClient: MongoClient = MongoClient() 
      val database: MongoDatabase = mongoClient.getDatabase("scala") 
      val collection: MongoCollection[Document] = database.getCollection("tests") 
      val doc: Document = Document("_id" -> 0, "name" -> "MongoDB", "type" -> "database", 
           "count" -> 1, "info" -> Document("x" -> 203, "y" -> 102)) 
      collection.insertOne(doc) 
      mongoClient.close() 
      Ok("Hello.") 
     } 
    } 
} 

class GomadApp(host: String, port: Int) { 
    private val pool = Executors.newCachedThreadPool() 

    println(s"Starting server on '$host:$port'") 

    val routes = new Routes().service 

    // Add some logging to the service 
    val service: HttpService = routes.local { req => 
    val path = req.uri 
    val start = System.nanoTime() 
    val result = req 
    val time = ((System.nanoTime() - start)/1000)/1000.0 
    println(s"${req.remoteAddr.getOrElse("null")} -> ${req.method}: $path in $time ms") 
    result 
    } 

    // Construct the blaze pipeline. 
    def build(): ServerBuilder = 
    BlazeBuilder 
     .bindHttp(port, host) 
     .mountService(service) 
     .withServiceExecutor(pool) 
} 

object GomadApp extends ServerApp { 
    val ip = "127.0.0.1" 
    val port = envOrNone("HTTP_PORT") map (_.toInt) getOrElse (8787) 

    override def server(args: List[String]): Task[Server] = 
    new GomadApp(ip, port) 
     .build() 
     .start 

} 

Answer 1

Я бы рекомендовал https://github.com/haghard/mongo-query-streams - хотя вам придется раскошелиться его и до зависимостей немного, scalaz 7,1 и 7,2 не бинарно совместимы.

Менее бегущий (и менее референциально правильный) путь: https://github.com/Verizon/delorean

collection.insertOne(doc).toFuture().toTask.flatMap({res => Ok("Hello")}) 

Последнее решение выглядит проще, но у него есть какие-то скрытые подводные камни. См https://www.reddit.com/r/scala/comments/3zofjl/why_is_future_totally_unusable/

Этот твит заставил меня задаться вопросом: https://twitter.com/timperrett/status/684584581048233984 Считаете ли вы Фьючерс «полностью непригодных» или это просто гипербола? У меня никогда не было серьезной проблемы, но я хочу быть просветленным. Не делает ли следующий код фьючерсами эффективно «ленивыми»? def myFuture = Future {42} И, наконец, я также слышал громыхания, что задачи Задачи Скаласа также имеют некоторые недостатки, но я не нашел много на этом. У кого-нибудь есть подробности?

Ответ:

Основная проблема заключается в том, что построение будущего с выражением бокового осуществления сама по себе является побочным эффектом. Вы можете только рассуждать о будущем для чистых вычислений, которые, к сожалению, не так, как они обычно используются. Вот демонстрация этой операции нарушения ссылочной прозрачности:

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

val f1 = { 
    val r = new Random(0L) 
    val x = Future(r.nextInt) 
    for { 
    a <- x 
    b <- x 
    } yield (a, b) 
} 

// Same as f1, but I inlined `x` 
val f2 = { 
    val r = new Random(0L) 
    for { 
    a <- Future(r.nextInt) 
    b <- Future(r.nextInt) 
    } yield (a, b) 
} 

f1.onComplete(println) // Success((-1155484576,-1155484576)) 
f2.onComplete(println) // Success((-1155484576,-723955400)) <-- not the same 

Однако это прекрасно работает с Task. Обратите внимание, что интересным является не-встроенная версия, которая позволяет создавать два разных значения Int. Это важный бит: у задачи есть конструктор, который фиксирует побочные эффекты как ценности, а Future - нет.

import scalaz.concurrent.Task 

val task1 = { 
    val r = new Random(0L) 
    val x = Task.delay(r.nextInt) 
    for { 
    a <- x 
    b <- x 
    } yield (a, b) 
} 

// Same as task1, but I inlined `x` 
val task2 = { 
    val r = new Random(0L) 
    for { 
    a <- Task.delay(r.nextInt) 
    b <- Task.delay(r.nextInt) 
    } yield (a, b) 
} 

println(task1.run) // (-1155484576,-723955400) 
println(task2.run) // (-1155484576,-723955400) 

Большинство обычно цитируемое различие, как «задача не работать, пока вы не попросите его» и «вы можете составить ту же задачу снова и снова» восходит к этому фундаментальному отличию , Итак, причина, по которой это «совершенно непригодна», заключается в том, что, как только вы привыкли к программированию с чистыми ценностями и полагаетесь на эквациональные рассуждения, чтобы понимать и манипулировать программами, трудно вернуться в побочный мир, где вещи гораздо труднее понять.