Mhm weird. Sink seems to work fine but the MongoCursor seem to receive a reply with a cursorId 0 (as of documentation that states that indicates a dead cursor, whatever that means) and thus eagerly closes the stream. Would you be fine if I'd add ResponseFlags to the reply class?
What do you mean by ResponseFlags? Could you point me to the documentation?
The response flags are documented in http://docs.mongodb.org/meta-driver/latest/legacy/mongodb-wire-protocol/#op-reply . They already a field
flags in https://github.com/fehmicansaglam/tepkin/blob/master/tepkin/src/main/scala/net/fehmicansaglam/tepkin/protocol/message/Reply.scala but are not parsed (but maybe I am in the wrong version of the wire protocol).
I think I have misunderstood the tailable cursors then. But why would the mongo give me a dead cursor for a tailable cursor query on an empty collection if I want to query all elements that will arrive there in the near future. I thought that is what tailable cursors / capped collections are made for.
Maybe there is a way to handle empty collections
The dead cursor on empty collection seem to be an off by one corner case of the dead cursor "the cyclic log buffer has overwritten your current read position" - like in "the producer is faster than the consumer". The tailable cursor documentation state that the user should explicitply deal with such errors (in their example, using a for loop). We maybe could fail the source if we receive a dead cursor on tailable
MongoCursors (it should never complete normally). if (cursorID == 0 && tailable) onError(new DeadCursorException()). The user could then decide to use a proper failure policy (restart or fail the whole stream, or switch to another source or whatever)
I think we can rely on stream supervision such that the user can decide if she wants to restart/stop/fail downstream (see http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-M5/scala/stream-error.html)
The restart should just restart the stage (create an new
MongoCursor actor)
BTW I have added . notation support for keys as can be seen here: https://github.com/fehmicansaglam/tepkin/blob/master/bson%2Fsrc%2Ftest%2Fscala%2Fnet%2Ffehmicansaglam%2Fbson%2FBsonSpec.scala#L40
document.getAs[Int]("age").value shouldBe 18
document.getAs[Double]("details.salary").value shouldBe 455.5
document.getAs[String]("details.personal.foo").value shouldBe "bar"
document.getAsList[List[_]]("details.inventory").value shouldBe List("a", 3.5, 1L, true)
val details = document.getAs[BsonDocument]("details").get
details.getAs[Double]("salary").value shouldBe 455.5
val personal = document.getAs[BsonDocument]("details.personal").get
personal.getAs[String]("foo").value shouldBe "bar"
}
@fehmicansaglam this is problem detail. https://groups.google.com/forum/#!topic/reactivemongo/4ZFU4YFJB-4
We agreed I would be maintaining this project from now on