Merge pull request #74 from hundio/master

Add proper support for tailable cursors and awaitData

Author: zookzook

lib/mongo.ex

@@ -657,7 +657,6 @@ defmodule Mongo do
            showRecordId: opts[:show_record_id],
            tailable: opts[:tailable],
            oplogReplay: opts[:oplog_replay],
-           tailable: opts[:tailable],
            noCursorTimeout: opts[:no_cursor_timeout],
            awaitData: opts[:await_data],
            batchSize: opts[:batch_size],

lib/mongo/stream.ex

@@ -5,7 +5,7 @@ defmodule Mongo.Stream do
 
   import Record, only: [defrecordp: 2]
 
-  defstruct [:topology_pid, :session, :cursor, :coll, :docs, :opts]
+  defstruct [:topology_pid, :session, :cursor, :coll, :docs, :cmd, :opts]
 
   alias Mongo.Session
 
@@ -22,7 +22,7 @@ defmodule Mongo.Stream do
                "ns" => coll,
                "firstBatch" => docs}}} when ok == 1 <- Mongo.exec_command_session(session, cmd, opts) do
 
-      %Mongo.Stream{topology_pid: topology_pid, session: session, cursor: cursor_id, coll: coll, docs: docs, opts: opts}
+      %Mongo.Stream{topology_pid: topology_pid, session: session, cursor: cursor_id, coll: coll, docs: docs, cmd: cmd, opts: opts}
     else
       {:error, error} ->
         case Error.should_retry_read(error, cmd, opts) do
@@ -35,11 +35,11 @@ defmodule Mongo.Stream do
 
   defimpl Enumerable do
 
-    defrecordp :state, [:topology_pid, :session, :cursor, :coll,  :docs]
+    defrecordp :state, [:topology_pid, :session, :cursor, :coll, :cmd, :docs]
 
-    def reduce(%Mongo.Stream{topology_pid: topology_pid, session: session, cursor: cursor_id, coll: coll, docs: docs, opts: opts}, acc, reduce_fun) do
+    def reduce(%Mongo.Stream{topology_pid: topology_pid, session: session, cursor: cursor_id, coll: coll, docs: docs, cmd: cmd, opts: opts}, acc, reduce_fun) do
 
-      start_fun = fn -> state(topology_pid: topology_pid, session: session, cursor: cursor_id, coll: coll, docs: docs) end
+      start_fun = fn -> state(topology_pid: topology_pid, session: session, cursor: cursor_id, coll: coll, cmd: cmd, docs: docs) end
       next_fun  = next_fun(opts)
       after_fun = after_fun(opts)
 
@@ -51,9 +51,9 @@ defmodule Mongo.Stream do
         state(docs: [], cursor: 0) = state ->  {:halt, state}
 
         # this is a regular cursor
-        state(docs: [], topology_pid: topology_pid, session: session, cursor: cursor, coll: coll) = state ->
+        state(docs: [], topology_pid: topology_pid, session: session, cursor: cursor, coll: coll, cmd: cmd) = state ->
           case get_more(topology_pid, session, only_coll(coll), cursor, nil, opts) do
-            {:ok, %{cursor_id: cursor_id, docs: []}}   -> {:halt, state(state, cursor: cursor_id)}
+            {:ok, %{cursor_id: cursor_id, docs: []}}   -> {if(cmd[:tailable], do: [], else: :halt), state(state, cursor: cursor_id)}
             {:ok, %{cursor_id: cursor_id, docs: docs}} -> {docs, state(state, cursor: cursor_id)}
             {:error, error}                            -> raise error
           end

test/mongo/cursor_test.exs

@@ -1,6 +1,30 @@
 defmodule Mongo.CursorTest do
   use CollectionCase, async: false
 
+  test "tailable cursors with awaitData", c do
+
+    coll      = "tailable_cursors"
+    init_docs = Stream.cycle([%{"foo" => 42}]) |> Enum.take(5)
+    tail_docs = Stream.cycle([%{"foo" => 10}]) |> Enum.take(10)
+
+    assert :ok      = Mongo.create(c.pid, coll, capped: true, size: 1_000_000)
+    assert {:ok, _} = Mongo.insert_many(c.pid, coll, init_docs)
+
+    tailing_task = Task.async fn ->
+      Mongo.find(c.pid, coll, %{}, tailable: true, await_data: true)
+      |> Enum.take(15)
+    end
+
+    Enum.each tail_docs, fn doc ->
+      Process.sleep 100
+      Mongo.insert_one(c.pid, coll, doc)
+    end
+
+    expected_docs = init_docs ++ tail_docs
+    assert ^expected_docs = Task.await(tailing_task) |> Enum.map(fn m ->  Map.pop(m, "_id") |> elem(1) end)
+
+  end
+
   test "checking if killCursor is called properly", c do
 
     coll    = "kill_cursors"