CXX-1252 Tailable cursors only work once

The way cursors were marked dead to support multiple iterators
on the same cursor had a side effect of preventing tailable cursors
from working.

This commit makes cursors know if they are tailable and rather than
marking tailable cursors dead, they are marked "pending" so they
can restart if re-iterated.

(cherry picked from commit 32b1b26)

Author: xdg

src/mongocxx/collection.cpp

@@ -316,7 +316,7 @@ cursor collection::find(view_or_value filter, const options::find& options) {
         libmongoc::cursor_set_max_await_time_ms(mongoc_cursor, static_cast<std::uint32_t>(count));
     }
 
-    return cursor{mongoc_cursor};
+    return cursor{mongoc_cursor, options_converted.cursor_type()};
 }
 
 stdx::optional<bsoncxx::document::value> collection::find_one(view_or_value filter,
@@ -921,7 +921,7 @@ cursor collection::distinct(bsoncxx::string::view_or_value field_name, view_or_v
         throw_exception<operation_exception>(error);
     }
 
-    return {fake_cursor};
+    return cursor{fake_cursor};
 }
 
 cursor collection::list_indexes() const {

src/mongocxx/cursor.cpp

@@ -33,8 +33,8 @@
 namespace mongocxx {
 MONGOCXX_INLINE_NAMESPACE_BEGIN
 
-cursor::cursor(void* cursor_ptr)
-    : _impl(stdx::make_unique<impl>(static_cast<mongoc_cursor_t*>(cursor_ptr))) {
+cursor::cursor(void* cursor_ptr, bsoncxx::stdx::optional<cursor::type> cursor_type)
+    : _impl(stdx::make_unique<impl>(static_cast<mongoc_cursor_t*>(cursor_ptr), cursor_type)) {
 }
 
 cursor::cursor(cursor&&) noexcept = default;
@@ -56,7 +56,7 @@ cursor::iterator& cursor::iterator::operator++() {
         _cursor->_impl->mark_dead();
         throw_exception<query_exception>(error);
     } else {
-        _cursor->_impl->mark_dead();
+        _cursor->_impl->mark_nothing_left();
         _cursor = nullptr;  // Set iterator equal to end().
     }
     return *this;

src/mongocxx/cursor.hpp

@@ -17,6 +17,7 @@
 #include <memory>
 
 #include <bsoncxx/document/view.hpp>
+#include <bsoncxx/stdx/optional.hpp>
 
 #include <mongocxx/config/prelude.hpp>
 
@@ -53,6 +54,7 @@ class MONGOCXX_API cursor {
     ///
     ~cursor();
 
+    ///
     /// A cursor::iterator that points to the beginning of any available
     /// results.  If begin() is called more than once, the cursor::iterator
     /// returned points to the next remaining result, not the result of
@@ -64,7 +66,11 @@ class MONGOCXX_API cursor {
     ///
     iterator begin();
 
-    /// A cursor::iterator that points to the end of the results.
+    ///
+    /// A cursor::iterator that points to the end of the results.  In the
+    /// case of a tailable cursor, this iterator will compare equal to an
+    /// exhausted tailable cursor iterator, even if more results are available
+    /// the next time the cursor is iterated.
     ///
     /// @return the cursor::iterator
     ///
@@ -75,7 +81,8 @@ class MONGOCXX_API cursor {
     friend class client;
     friend class database;
 
-    MONGOCXX_PRIVATE cursor(void* cursor_ptr);
+    MONGOCXX_PRIVATE cursor(void* cursor_ptr,
+                            bsoncxx::stdx::optional<type> cursor_type = bsoncxx::stdx::nullopt);
 
     class MONGOCXX_PRIVATE impl;
     std::unique_ptr<impl> _impl;

src/mongocxx/private/cursor.hh

@@ -15,6 +15,7 @@
 #pragma once
 
 #include <bsoncxx/document/view.hpp>
+#include <bsoncxx/stdx/optional.hpp>
 #include <mongocxx/cursor.hpp>
 #include <mongocxx/private/libmongoc.hh>
 
@@ -25,12 +26,16 @@ MONGOCXX_INLINE_NAMESPACE_BEGIN
 
 class cursor::impl {
    public:
-    // States represent a one-way, ordered lifecycle of a cursor. k_started
-    // means that libmongoc::cursor_next has been called at least once.
+    // States represent a one-way, ordered lifecycle of a cursor. k_started means that
+    // libmongoc::cursor_next has been called at least once.  However, for a tailable
+    // cursor, the cursor resets to k_pending on exhaustion so that it can resume later.
     enum class state { k_pending = 0, k_started = 1, k_dead = 2 };
 
-    impl(mongoc_cursor_t* cursor)
-        : cursor_t(cursor), status{cursor ? state::k_pending : state::k_dead} {
+    impl(mongoc_cursor_t* cursor, bsoncxx::stdx::optional<cursor::type> cursor_type)
+        : cursor_t(cursor),
+          status{cursor ? state::k_pending : state::k_dead},
+          tailable{cursor && cursor_type && (*cursor_type == cursor::type::k_tailable ||
+                                             *cursor_type == cursor::type::k_tailable_await)} {
     }
 
     ~impl() {
@@ -45,18 +50,28 @@ class cursor::impl {
         return status == state::k_dead;
     }
 
-    void mark_started() {
-        status = state::k_started;
+    bool is_tailable() const {
+        return tailable;
     }
 
     void mark_dead() {
-        doc = bsoncxx::document::view{};
+        mark_nothing_left();
         status = state::k_dead;
     }
 
+    void mark_nothing_left() {
+        doc = bsoncxx::document::view{};
+        status = tailable ? state::k_pending : state::k_dead;
+    }
+
+    void mark_started() {
+        status = state::k_started;
+    }
+
     mongoc_cursor_t* cursor_t;
     bsoncxx::document::view doc;
     state status;
+    bool tailable;
 };
 
 MONGOCXX_INLINE_NAMESPACE_END

src/mongocxx/test/collection.cpp

@@ -17,6 +17,7 @@
 
 #include <vector>
 
+#include <bsoncxx/builder/basic/document.hpp>
 #include <bsoncxx/builder/stream/document.hpp>
 #include <bsoncxx/builder/stream/helpers.hpp>
 #include <bsoncxx/json.hpp>
@@ -35,6 +36,7 @@
 #include <mongocxx/read_concern.hpp>
 #include <mongocxx/write_concern.hpp>
 
+using bsoncxx::builder::basic::kvp;
 using namespace bsoncxx::builder::stream;
 using namespace mongocxx;
 
@@ -1692,6 +1694,59 @@ TEST_CASE("create_index tests", "[collection]") {
     }
 }
 
+TEST_CASE("Tailable cursors", "[collection][tailable]") {
+    instance::current();
+    client mongodb_client{uri{}};
+    database db = mongodb_client["test"];
+
+    // Drop and (re)create a capped collection.
+    db["tailcheck"].drop();
+    auto create_opts = options::create_collection{}.capped(true).size(1024 * 1024);
+    db.create_collection("tailcheck", create_opts);
+
+    // Create tailable await cursor.
+    options::find opts;
+
+    SECTION("k_tailable") {
+        opts.cursor_type(cursor::type::k_tailable);
+    }
+
+    SECTION("k_tailable_await") {
+        opts.cursor_type(cursor::type::k_tailable_await);
+    }
+
+    INFO(std::string{opts.cursor_type() == cursor::type::k_tailable ? "k_tailable"
+                                                                    : "k_tailable_await"})
+
+    auto coll = db["tailcheck"];
+    auto cursor = coll.find({}, opts);
+
+    // Insert 3 documents.
+    for (int32_t n : {1, 2, 3}) {
+        coll.insert_one(document{} << "x" << n << finalize);
+    }
+
+    // Check that cursor finds three documents.
+    auto expected = 0;
+
+    for (auto&& doc : cursor) {
+        REQUIRE(doc["x"].get_int32() == ++expected);
+    }
+
+    // Insert 3 more documents.
+    for (int32_t n : {4, 5, 6}) {
+        coll.insert_one(document{} << "x" << n << finalize);
+    }
+
+    // Check that cursor finds next three documents.
+    for (auto&& doc : cursor) {
+        REQUIRE(doc["x"].get_int32() == ++expected);
+    }
+
+    // Check that cursor found all documents
+    REQUIRE(expected == 6);
+}
+
 TEST_CASE("regressions", "CXX-986") {
     instance::current();
     mongocxx::uri mongo_uri{"mongodb://non-existent-host.invalid/"};