ScalaでMongoDBを何とかしたい

# はじめに
MongoDBって今まであまり興味を持ってなかったけど、ちょこっとScalaで触ってみたところ、`Observable`がいまいちScalaらしくなかったりしたので`Future`で扱った方が楽だよねって感じな話。

## 環境

- Scala: 2.13
- sbt: 1.3.4
- mongo-scala-driver: 2.7.0
- cats: 2.0.0
- 他

## 成果物
https://github.com/lightstaff/scala-mongo-example

# 標準的な使用方法
標準的な使用方法から始めます。

まずはスキーマを`case class`で定義できるようなので利用します。単純に子として`B`を複数持つ`A`という何の役にも立ちそうにないスキーマを定義します。

```scala:models.scala
package com.example

import org.mongodb.scala.bson.ObjectId

final case class A(_id: ObjectId, name: String, children: Seq[B])

final case class B(name: String)
```

上記を`Main`等のデータベース初期化時かコレクション初期化時に`withCodecRegistry...`として渡してやると自動でBSONにパースしてくれるようです。

では早速、上記スキーマをCRUDしてみます。

公式ライブラリである[mongo-scala-driver](https://github.com/mongodb/mongo-scala-driver/tree/master)は[mongo-java-driver](https://github.com/mongodb/mongo-java-driver/tree/master)をラップしたものらしく、諸々の処理結果を`Observable`というRx系等に慣れた人なら動作が想像できそうな非同期っぽいJavaの型に包んで返してきます。これを`onNext`と`onError`と`onComplete`を持つ`Observer`型のコールバックを引数に持つ`subscribe`メソッドで処理する形式になっています（RxJSとは違い`Observable`は`subscribe`メソッドが呼ばれるまで動かないらしい・・・非同期って難しい）。

```scala:main.scala
package com.example

import scala.concurrent.ExecutionContext

import org.bson.codecs.configuration.CodecRegistries
import org.mongodb.scala._
import org.mongodb.scala.bson.ObjectId
import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
import org.mongodb.scala.bson.codecs.Macros._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.model.Updates._

object Main extends App with LazyLogging {

  implicit val executionContext: ExecutionContext = ExecutionContext.global

  val mongoClient = MongoClient(
    "mongodb://root:1234@localhost:27017/?authSource=admin"
  )

  val mongoDb = mongoClient
    .getDatabase("test_db")
    .withCodecRegistry(
      CodecRegistries.fromRegistries(
        CodecRegistries.fromProviders(classOf[A], classOf[B]),
        DEFAULT_CODEC_REGISTRY
      )
    )

  val mongoCollection = mongoDb.getCollection[A]("test")

  val id1 = new ObjectId()

  val id2 = new ObjectId()

  val id3 = new ObjectId()

  logger.info("=========== start version of normal ==========")

  val f1 = for {
    _ <- mongoCollection.drop()
    _ <- mongoCollection.insertOne(
          A(id1, "a_1", Seq(B("b_1"), B("b_2")))
        )
    _ <- mongoCollection.insertOne(
          A(id2, "a_2", Seq.empty[B])
        )
    _ <- mongoCollection.insertOne(
          A(id3, "a_3", Seq.empty[B])
        )
    _ <- mongoCollection.updateOne(
          equal("_id", id2),
          combine(set("name", "a_4"), set("children", Seq(B("b_3"))))
        )
    _ <- mongoCollection.deleteOne(equal("_id", id3))
    found <- mongoCollection.find()
  } yield found

  f1.subscribe(new Observer[A] {
    override def onNext(result: A): Unit = logger.info(result.toString)

    override def onError(e: Throwable): Unit = logger.error("failed!!")

    override def onComplete(): Unit ={
      logger.info("=========== finish version of normal ==========")
      
      mongoClient.close()
    }
  })
}
```

う〜ん・・・。一応上記のようにfor構文を使えなくもない感じですが、それぞれのコールバック（`onNext`,`onError`,`onComplete`）がどう処理されてんのかよく分からりません。特に途中の`onError`が処理されない気配がプンプンする。

for構文が使えなかったらコールバック地獄行きだし、for構文使うならやっぱりMonadとして組みたいのでどうにかしたいなぁって感じです。

# どうにかする
実は`Observable`には`toFuture`という`Observable -> Future`な関数が最初から用意されているので割と簡単にどうにでもなったりする。

という訳で毎度のパターンですがRepositoryに切り出してみましょう。今回はシンプルにMonadを期待する形式にしました。

```scala:repository.scala
package com.example

import org.mongodb.scala.bson.ObjectId

trait Repository[F[_]] {

  def drop: F[Unit]

  def findAll: F[Seq[A]]

  def insert(entity: A): F[Unit]

  def update(entity: A): F[Unit]

  def delete(_id: ObjectId): F[Unit]
}
```

上記をCatsの`Kleisli`と`IO`辺りを利用して実装してみます。

```scala:repositoryImpl.scala
package com.example

import scala.concurrent.ExecutionContext

import cats.data.Kleisli
import cats.effect.{ContextShift, IO}
import org.mongodb.scala.MongoCollection
import org.mongodb.scala.bson.ObjectId
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.model.Updates._

object RepositoryImpl extends Repository[Kleisli[IO, MongoCollection[A], *]] {

  implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global)

  override def drop: Kleisli[IO, MongoCollection[A], Unit] =
    Kleisli(c => IO.fromFuture(IO(c.drop().toFuture())).map(_ => ()))

  override def findAll: Kleisli[IO, MongoCollection[A], Seq[A]] =
    Kleisli(c => IO.fromFuture(IO(c.find().toFuture())))

  override def insert(entity: A): Kleisli[IO, MongoCollection[A], Unit] =
    Kleisli(c => IO.fromFuture(IO(c.insertOne(entity).toFuture())).map(_ => ()))

  override def update(entity: A): Kleisli[IO, MongoCollection[A], Unit] =
    Kleisli(
      c =>
        IO.fromFuture(
            IO(
              c.updateOne(equal("_id", entity._id),
                           combine(set("name", entity.name),
                                   set("children", entity.children)))
                .toFuture()
            )
          )
          .map(_ => ())
    )

  override def delete(
      _id: ObjectId
  ): Kleisli[IO, MongoCollection[A], Unit] =
    Kleisli(
      c =>
        IO.fromFuture(IO(c.deleteOne(equal("_id", _id)).toFuture()))
          .map(_ => ())
    )
}
```

走り書き感が溢れてますが、とりあえず利用する側は`Observable`を意識せず、関数合成できるようになるはずです。

という訳で`Main`に戻り、リポジトリを利用したバージョンを追記+一部修正します。

```scala:main.scala
package com.example

import java.util.concurrent.TimeUnit

import scala.concurrent.duration.Duration
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.util.{Failure, Success, Try}

import cats.data.Kleisli
import cats.implicits._
import com.typesafe.scalalogging.LazyLogging
import org.bson.codecs.configuration.CodecRegistries
import org.mongodb.scala._
import org.mongodb.scala.bson.ObjectId
import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
import org.mongodb.scala.bson.codecs.Macros._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.model.Updates._

object Main extends App with LazyLogging {

  implicit val executionContext: ExecutionContext = ExecutionContext.global

  val mongoClient = MongoClient(
    "mongodb://root:1234@localhost:27017/?authSource=admin"
  )

  val mongoDb = mongoClient
    .getDatabase("test_db")
    .withCodecRegistry(
      CodecRegistries.fromRegistries(
        CodecRegistries.fromProviders(classOf[A], classOf[B]),
        DEFAULT_CODEC_REGISTRY
      )
    )

  val mongoCollection = mongoDb.getCollection[A]("test")

  val id1 = new ObjectId()

  val id2 = new ObjectId()

  val id3 = new ObjectId()

  logger.info("=========== start version of normal ==========")

  // f1 = Observable[A]
  val f1 = for {
    _ <- mongoCollection.drop()
    _ <- mongoCollection.insertOne(
          A(id1, "a_1", Seq(B("b_1"), B("b_2")))
        )
    _ <- mongoCollection.insertOne(
          A(id2, "a_2", Seq.empty[B])
        )
    _ <- mongoCollection.insertOne(
          A(id3, "a_3", Seq.empty[B])
        )
    _ <- mongoCollection.updateOne(
          equal("_id", id2),
          combine(set("name", "a_4"), set("children", Seq(B("b_3"))))
        )
    _ <- mongoCollection.deleteOne(equal("_id", id3))
    found <- mongoCollection.find()
  } yield found

  f1.subscribe(new Observer[A] {
    override def onNext(result: A): Unit = logger.info(result.toString)

    override def onError(e: Throwable): Unit = logger.error("failed!!")

    override def onComplete(): Unit =
      logger.info("=========== finish version of normal ==========")
  })

  logger.info("=========== start version of repository ==========")

  // f2 = Klesli[IO, MongoCollection[A], Seq[A]]
  val f2 = for {
    _ <- RepositoryImpl.drop
    _ <- RepositoryImpl.insert(
          A(id1,
            "a_1",
            Seq(
              B("b_1"),
              B("b_2")
            ))
        )
    _ <- RepositoryImpl.insert(
          A(id2, "a_2", Seq.empty[B])
        )
    _ <- RepositoryImpl.insert(
          A(id3, "a_3", Seq.empty[B])
        )
    _ <- RepositoryImpl
          .update(A(id2, "a_4", Seq(B("b_3"))))
    _ <- RepositoryImpl.delete(id3)
    found <- RepositoryImpl.findAll
  } yield found

  f2.run(mongoCollection).unsafeRunAsync {
    case Right(v) =>
      v.foreach(a => logger.info(a.toString))
      logger.info("=========== finish version of repository ==========")

    case Left(_) =>
      logger.error("failed!!")
  }

  // 30秒もあれば終わるはず
  Await.result(Future(Thread.sleep(30000)), Duration(30, TimeUnit.SECONDS))

  mongoClient.close()
}
```

上記を実行するとコンソールには下記のように表示されるはず（ドライバが出すログは省略）。

```console
=========== start version of normal ==========
=========== start version of repository ==========
A(5ddcb68d27918964b23fddb4,a_1,List(B(b_1), B(b_2)))
A(5ddcb68d27918964b23fddb5,a_4,List(B(b_3)))
=========== finish version of normal ==========
A(5ddcb68d27918964b23fddb4,a_1,List(B(b_1), B(b_2)))
A(5ddcb68d27918964b23fddb5,a_4,List(B(b_3)))
=========== finish version of repository ==========
```

# おわりに
ちなみに[公式のチュートリアル](http://mongodb.github.io/mongo-scala-driver/2.7/getting-started/quick-tour/)はチュートリアル用のヘルパーが用意してあってあっさり`Future`->`Await`にしてたりする。

[ReactiveStream](https://github.com/mongodb/mongo-scala-driver/tree/master/examples/src/test/scala/reactivestreams)とかも使えるらしいけどその辺はまた今度・・・があれば・・・。