Alpakka MongoDB - specify type in MongoSource

Question

I'm currently playing around with Akka Streams and the Alpakka MongoDB connector.

Is it possible to specify the type for MongoSource?

val codecRegistry = fromRegistries(fromProviders(classOf[TodoMongo]), DEFAULT_CODEC_REGISTRY)
  private val todoCollection: MongoCollection[TodoMongo] = mongoDb
    .withCodecRegistry(codecRegistry)
    .getCollection("todo")

I would like to do something like this:

val t: FindObservable[Seq[TodoMongo]] = todoCollection.find()
MongoSource(t) // Stuck here

But I get the following error:

Expected Observable[scala.Document], Actual FindObservable[Seq[TodoMongo]].

I can't find the correct documentation about this part.

Answer 1

This is not published yet, but in Alpakka's master branch, MongoSource.apply takes a type parameter:

object MongoSource {
  def apply[T](query: Observable[T]): Source[T, NotUsed] =
    Source.fromPublisher(ObservableToPublisher(query))
}

Therefore, with the upcoming 0.18 release of Alpakka, you'll be able to do the following:

val source: Source[TodoMongo, NotUsed] = MongoSource[TodoMongo](todoCollection.find())

Note that source here assumes that todoCollection.find() returns an Observable[TodoMongo]; adjust the types as needed.

In the meantime, you could simply add the above code manually. For example:

package akka.stream.alpakka.mongodb.scaladsl

import akka.NotUsed
import akka.stream.alpakka.mongodb.ObservableToPublisher
import akka.stream.scaladsl.Source
import org.mongodb.scala.Observable

object MyMongoSource {
  def apply[T](query: Observable[T]): Source[T, NotUsed] =
    Source.fromPublisher(ObservableToPublisher(query))
}

Note that MyMongoSource is defined to reside in the akka.stream.alpakka.mongodb.scaladsl package (like MongoSource), because ObservableToPublisher is a package-private class. You would use MyMongoSource in the same way that you would use MongoSource:

val source: Source[TodoMongo, NotUsed] = MyMongoSource[TodoMongo](todoCollection.find())