Observable からデータにアクセスする

観察可能な値を処理する方法

MongoDB操作を実行してそのデータを処理するには、Observable から操作結果をリクエスト必要があります。ドライバーは、任意の数の結果を返す操作用の Observable インターフェースを提供します。結果を生成しないか、結果を 1 つ生成する操作（findOne() メソッドなど）では、SingleObservable[T] が返されます。[T]化パラメータは、SingleObservable が取り扱うデータ型に対応します。

無制限の数の結果を生成できる操作では、Observable[T] 型のインスタンスが返されます。 一部の操作では特定の Observable タイプが返され、サブスクライブ前に結果をフィルタリングして処理するための追加メソッドが提供されます。 次のリストでは、操作固有のメソッドを Observable に連鎖させることができるいくつかのタイプについて説明します。

• FindObservable[T]: find() メソッドによって返される

• DistinctObservable[T]: distinct() メソッドによって返される

• AggregateObservable[T]: aggregate() メソッドによって返される

操作の Observable で subscribe() メソッドを呼び出すことで、操作結果をリクエストできます。Observerクラスのインスタンスを subscribe() メソッドのパラメーターとして渡します。この Observer は、Observable から操作結果を受け取ります。その後、Observerクラスが提供するメソッドを使用して、結果を出力したり、エラーを処理したり、追加のデータ処理を実行したりできます。

結果の処理の詳細については、次のAPIドキュメントを参照してください。

• Observable

• サブスクリプション

• 観察者

サンプル データ

restaurantssample_restaurantsこのガイドの例では、Atlasサンプルデータセット の データベースの コレクションを使用します。Scalaアプリケーションからこのコレクションにアクセスするには、Atlas クラスターに接続する MongoClient を作成し、database 変数と collection 変数に次の値を割り当てます。

val database: MongoDatabase = mongoClient.getDatabase("sample_restaurants")
val collection: MongoCollection[Document] = database.getCollection("restaurants")

MongoDB Atlasクラスターを無料で作成して、サンプルデータセットをロードする方法については、 「Atlas を使い始める」ガイドを参照してください。

読み取り操作の結果にアクセスする

読み取り操作によって取得されたデータにアクセスするには、操作結果を保存するための Observable[T] を作成します。 次に、 Observable をサブスクライブし、Observerクラスのコールバックメソッドをオーバーライドして結果を処理します。

この例では、cuisine 値が "Czech" であるドキュメントを restaurantsコレクションでクエリします。結果を検索して処理するために、この例では操作に Observable[Document] を割り当て、次のアクションを実行します。

• subscribe() メソッドを呼び出して Observable にサブスクライブし、Observer をパラメータとして渡します

• onNext() メソッドをオーバーライドして、検索された各ドキュメント（Document インスタンスである）を出力します

• エラーを出力するために onError() メソッドをオーバーライドします

• Observable からのすべての結果が処理された後にメッセージを出力するように onComplete() メソッドをオーバーライドします

val filter = equal("cuisine", "Czech")
val findObservable: Observable[Document] = collection.find(filter)
findObservable.subscribe(new Observer[Document] {
  override def onNext(result: Document): Unit = println(result)
  override def onError(e: Throwable): Unit = println("Failed: " + e.getMessage)
  override def onComplete(): Unit = println("Processed all results")
})

Iterable((_id, ...), ..., (name,BsonString{value='Koliba Restaurant'}),
(restaurant_id,BsonString{value='40812870'}))
Iterable((_id, ...), ..., (name,BsonString{value='Bohemian Beer Garden'}),
(restaurant_id,BsonString{value='41485121'}))
Iterable((_id,...), ..., (name,BsonString{value='Hospoda'}),
(restaurant_id,BsonString{value='41569184'}))
Iterable((_id,...), ..., (name,BsonString{value='Olde Prague Tavern'}),
(restaurant_id,BsonString{value='41711983'}))
Processed all results

書込み (write) 操作の結果にアクセスする

書込み操作によって検索されたデータにアクセスするには、操作結果を保存するための Observable[T] を作成します。 次に、 Observable をサブスクライブし、Observerクラスのコールバックメソッドをオーバーライドして結果を処理します。

この例では、 cuisine 値が "Nepalese" であるドキュメントを restaurantsコレクションに挿入します。 結果を検索して処理するために、この例では操作に Observable[InsertManyResult] を割り当て、次のアクションを実行します。

• subscribe() メソッドを呼び出して Observable にサブスクライブし、Observer をパラメータとして渡します

• onNext() メソッドをオーバーライドして、挿入操作の結果を InsertManyResult として返します

• エラーを出力するために onError() メソッドをオーバーライドします

• Observable からのすべての結果が処理された後にメッセージを出力するように onComplete() メソッドをオーバーライドします

val docs: Seq[Document] = Seq(
  Document("name" -> "Cafe Himalaya", "cuisine" -> "Nepalese"),
  Document("name" -> "Taste From Everest", "cuisine" -> "Nepalese")
)
val insertObservable: Observable[InsertManyResult] = collection.insertMany(docs)
insertObservable.subscribe(new Observer[InsertManyResult] {
  override def onNext(result: InsertManyResult): Unit = println(result)
  override def onError(e: Throwable): Unit = println("Failed: " + e.getMessage)
  override def onComplete(): Unit = println("Processed all results")
})

AcknowledgedInsertManyResult{insertedIds={0=BsonObjectId{value=...},
1=BsonObjectId{value=...}}}
Processed all results

例

この例では、 restaurantsコレクションを boroughフィールドの個別の値ごとにクエリします。 このコードは distinct() メソッドによって返された Observable をサブスクライブし、Lambda 関数を使用して結果を出力し、エラーを処理します。

collection.distinct("borough")
          .subscribe((value: String) => println(value),
                     (e: Throwable) => println(s"Failed: $e"),
                     () => println("Processed all results"))

Bronx
Brooklyn
Manhattan
Missing
Queens
Staten Island
Processed all results

例

この例では、 nameフィールドの値が "The Halal Guys" であるドキュメントを restaurantsコレクションでクエリします。 操作結果にアクセスするために、コードは Observable を Future に変換し、Future が各結果を収集するまで待機して、結果を反復処理します。

val observable = collection.find(equal("name", "The Halal Guys"))
val results = Await.result(observable.toFuture(), Duration(10, TimeUnit.SECONDS))
results.foreach(println)

Iterable((_id,...), ..., (name,BsonString{value='The Halal Guys'}),
(restaurant_id,BsonString{value='50012258'}))
Iterable((_id,...), ..., (name,BsonString{value='The Halal Guys'}),
(restaurant_id,BsonString{value='50017823'}))