removed debug code and added a test case for change streams

Author: zookzook

lib/mongo/cursor.ex

@@ -155,20 +155,18 @@ defmodule Mongo.Cursor do
                    maxTimeMS: opts[:max_time]
                  ] |> filter_nils()
 
-      kill_cursors(conn, coll, [cursor_id], opts)
-
       with {:ok, %{"operationTime" => op_time,
                    "cursor" => %{"id" => new_cursor_id,
-                     "nextBatch" => docs} = cursor,
-                   "ok" => ok}} when ok == 1 <- Mongo.direct_command(conn, get_more, opts) do
+                                 "nextBatch" => docs} = cursor,
+                                 "ok" => ok}} when ok == 1 <- Mongo.direct_command(conn, get_more, opts) do
 
         old_token = change_stream(change_stream, :resume_token)
         change_stream = update_change_stream(change_stream, cursor["postBatchResumeToken"], op_time, List.last(docs))
         new_token = change_stream(change_stream, :resume_token)
 
-        case Map.equal?(old_token, new_token) do
-          false -> fun.(new_token)
-          true  -> nil
+        case token_changes(old_token, new_token) do
+          true  -> fun.(new_token)
+          false -> nil
         end
 
         {:ok, %{cursor_id: new_cursor_id, docs: docs, change_stream: change_stream}}
@@ -195,6 +193,11 @@ defmodule Mongo.Cursor do
       end
     end
 
+    defp token_changes(nil, nil), do: false
+    defp token_changes(nil, _new_token), do: true
+    defp token_changes(_old_token, nil), do: true
+    defp token_changes(old_token, new_token), do: not(Map.equal?(old_token, new_token))
+
     ##
     # we are updating the resume token by matching different cases
     #

test/mongo/change_stream_test.exs

@@ -0,0 +1,65 @@
+defmodule Mongo.ChangeStreamTest do
+  use ExUnit.Case # DO NOT MAKE ASYNCHRONOUS
+
+  setup_all do
+    assert {:ok, top} = Mongo.TestConnection.connect
+    assert {:ok, %Mongo.InsertOneResult{}} = Mongo.insert_one(top, "users", %{name: "Waldo"})
+    %{pid: top}
+  end
+
+  def consumer_1(top, monitor) do
+    cursor = Mongo.watch_collection(top, "users", [], fn doc -> send(monitor, {:token, doc}) end, max_time: 1_000 )
+    result = cursor |> Enum.take(2) |> Enum.at(0)
+    send(monitor, {:insert, result})
+  end
+
+  def consumer_2(top, monitor, token) do
+    cursor = Mongo.watch_collection(top, "users", [], fn doc -> send(monitor, {:token, doc}) end, resume_after: token, max_time: 1_000 )
+    result = cursor |> Enum.take(1) |> Enum.at(0)
+    send(monitor, {:insert, result})
+  end
+
+  def consumer_3(top, monitor, token) do
+    cursor = Mongo.watch_collection(top, "users", [], fn doc -> send(monitor, {:token, doc}) end, resume_after: token, max_time: 1_000 )
+    result = cursor |> Enum.take(4) |> Enum.map(fn %{"fullDocument" => %{"name" => name}} -> name end)
+    send(monitor, {:insert, result})
+
+  end
+
+  def producer(top) do
+    Process.sleep(100)
+    assert {:ok, %Mongo.InsertOneResult{}} = Mongo.insert_one(top, "users", %{name: "Greta"})
+    assert {:ok, %Mongo.InsertOneResult{}} = Mongo.insert_one(top, "users", %{name: "Gustav"})
+    assert {:ok, %Mongo.InsertOneResult{}} = Mongo.insert_one(top, "users", %{name: "Tom"})
+  end
+
+  test "change stream: watch and resume_after", %{pid: top} do
+
+    me = self()
+    spawn(fn -> consumer_1(top, me) end)
+    spawn(fn -> producer(top) end)
+
+    assert_receive {:token, nil}, 2_000
+    assert_receive {:token, token}, 2_000
+    assert_receive {:insert, %{"fullDocument" => %{"name" => "Greta"}}}, 2_000
+
+    Process.sleep(500)
+
+    assert {:ok, %Mongo.InsertOneResult{}} = Mongo.insert_one(top, "users", %{name: "Liese"})
+
+    spawn(fn -> consumer_2(top, me, token) end)
+    spawn(fn -> producer(top) end)
+
+    assert_receive {:token, _}, 2_000
+    assert_receive {:insert, %{"fullDocument" => %{"name" => "Gustav"}}}, 2_000
+
+    Process.sleep(500)
+
+    spawn(fn -> consumer_3(top, me, token) end)
+    spawn(fn -> producer(top) end)
+
+    assert_receive {:token, _}, 2_000
+    assert_receive {:insert, ["Gustav", "Tom", "Liese", "Greta"]}, 2_000
+
+  end
+end