http4s、Service ExecutorとMongodb：insertOneの終了を待つ方法

Question

基本的な質問は事前にお詫びください。私はhttp4sでScalaを学び始めています。ルータハンドラでは、MongoDBへのエントリを入力しようとしています。私が言うことができる限り、insertOneはObservable[Completed]を返します。

私はobservalbeが応答を返す前に完了するのをどのように待つことができますか？

私のコードは次のとおりです。 - あなたは少しそれをフォークと依存関係までする必要がありますが、scalaz 7.1および7.2は、バイナリ互換性がありません

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をお勧めします。

あまりStreamyに（と少ない参照上正しい）道：https://github.com/Verizon/delorean

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

後者の解決策は簡単に見えますが、それはいくつかの隠された落とし穴があります。 https://twitter.com/timperrett/status/684584581048233984 あなたは先物「完全に使用不可能」またはこの単なる誇張で検討してください：https://www.reddit.com/r/scala/comments/3zofjl/why_is_future_totally_unusable/

このツイートは私が疑問に思う作っ参照してください？私は大きな問題を抱えたことは一度もありませんが、私は啓発されたいです。次のコードによってFuturesが効果的に「怠け者」になることはありませんか？ 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 

しかし、これはタスクと正常に動作します。興味深いのは非インラインバージョンであり、これは2つの異なる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) 

と「あなたは何度も同じタスクを構成することができます」「あなたがして、それを頼むまで、タスクは実行されません」のような一般的に引用された差異のほとんどは、この根本的な違いに遡ります。 「完全に使用できない」理由は、純粋な値でプログラミングすることと、プログラムを理解し操作するための等式的な推論に頼ることができれば、物事がはるかに分かりにくい副作用の世界に戻るのは難しいということです。