CXX-1414: Add ChangeStreams Support

Author: rtimmons

.travis.yml

@@ -61,8 +61,7 @@ install:
     # Install Mongo C Driver
     - pushd mongo-c-driver
 
-    # TODO: Update this to our real minimum for the C++11 driver 3.2 release, once known.
-    - git checkout 1.9.2
+    - git checkout master
 
     - ./autogen.sh --enable-tests=no --enable-examples=no --with-libbson=bundled; make; sudo make install
 

src/mongocxx/CMakeLists.txt

@@ -50,6 +50,7 @@ add_subdirectory(config)
 set(mongocxx_sources
     bulk_write.cpp
     client.cpp
+    change_stream.cpp
     collection.cpp
     cursor.cpp
     database.cpp
@@ -73,6 +74,7 @@ set(mongocxx_sources
     model/write.cpp
     options/aggregate.cpp
     options/bulk_write.cpp
+    options/change_stream.cpp
     options/client.cpp
     options/count.cpp
     options/create_collection.cpp

src/mongocxx/change_stream.cpp

@@ -0,0 +1,117 @@
+// Copyright 2018-present MongoDB Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <mongocxx/change_stream.hpp>
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <tuple>
+
+#include <bsoncxx/private/libbson.hh>
+#include <bsoncxx/stdx/make_unique.hpp>
+#include <mongocxx/exception/private/error_category.hh>
+#include <mongocxx/exception/private/mongoc_error.hh>
+#include <mongocxx/exception/query_exception.hpp>
+#include <mongocxx/private/change_stream.hh>
+#include <mongocxx/private/libmongoc.hh>
+
+#include <mongocxx/config/private/prelude.hh>
+
+namespace mongocxx {
+MONGOCXX_INLINE_NAMESPACE_BEGIN
+
+// Requirements for concept Iterator:
+// http://en.cppreference.com/w/cpp/concept/Iterator
+static_assert(std::is_copy_constructible<change_stream::iterator>::value, "");
+static_assert(std::is_copy_assignable<change_stream::iterator>::value, "");
+static_assert(std::is_destructible<change_stream::iterator>::value, "");
+
+// Below basically assert that we have the traits on change_stream::iterator
+// so they can't be accidentally removed.
+static_assert(std::is_integral<change_stream::iterator::difference_type>::value, "");
+static_assert(std::is_class<change_stream::iterator::value_type>::value, "");
+static_assert(std::is_pointer<change_stream::iterator::pointer>::value, "");
+static_assert(std::is_reference<change_stream::iterator::reference>::value, "");
+
+change_stream::change_stream(change_stream&&) noexcept = default;
+
+change_stream& change_stream::operator=(change_stream&&) noexcept = default;
+
+change_stream::~change_stream() = default;
+
+change_stream::iterator change_stream::begin() {
+    if (_impl->is_dead()) {
+        return end();
+    }
+    return iterator{this};
+}
+
+change_stream::iterator change_stream::end() {
+    return iterator{};
+}
+
+// void* since we don't leak C driver defs into C++ driver
+change_stream::change_stream(void* change_stream_ptr)
+    : _impl(stdx::make_unique<impl>(*static_cast<mongoc_change_stream_t*>(change_stream_ptr))) {}
+
+change_stream::iterator::iterator() : change_stream::iterator::iterator{nullptr} {}
+
+const bsoncxx::document::view& change_stream::iterator::operator*() const noexcept {
+    return _change_stream->_impl->doc();
+}
+
+const bsoncxx::document::view* change_stream::iterator::operator->() const noexcept {
+    return &_change_stream->_impl->doc();
+}
+
+change_stream::iterator& change_stream::iterator::operator++() {
+    _change_stream->_impl->advance_iterator();
+    return *this;
+}
+
+void change_stream::iterator::operator++(int) {
+    operator++();
+}
+
+change_stream::iterator::iterator(change_stream* change_stream) : _change_stream(change_stream) {
+    if (!_change_stream || _change_stream->_impl->has_started()) {
+        return;
+    }
+
+    _change_stream->_impl->mark_started();
+    // Advance to first event on begin() to keep operator*() state-machine-free.
+    operator++();
+}
+
+// Don't worry about the case of two iterators being created from
+// different change_streams
+bool MONGOCXX_CALL operator==(const change_stream::iterator& lhs,
+                              const change_stream::iterator& rhs) noexcept {
+    return rhs.is_exhausted() && lhs.is_exhausted();
+}
+
+bool MONGOCXX_CALL operator!=(const change_stream::iterator& lhs,
+                              const change_stream::iterator& rhs) noexcept {
+    return !(lhs == rhs);
+}
+
+bool change_stream::iterator::is_exhausted() const {
+    // An iterator is exhausted if it is the end-iterator (_change_stream == nullptr)
+    // or if the underlying _change_stream is marked exhausted.
+    return !_change_stream || _change_stream->_impl->is_exhausted();
+}
+
+MONGOCXX_INLINE_NAMESPACE_END
+}  // namespace mongocxx

src/mongocxx/change_stream.hpp

@@ -0,0 +1,168 @@
+// Copyright 2018-present MongoDB Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <memory>
+
+#include <bsoncxx/document/view.hpp>
+#include <bsoncxx/stdx/optional.hpp>
+
+#include <mongocxx/config/prelude.hpp>
+
+namespace mongocxx {
+MONGOCXX_INLINE_NAMESPACE_BEGIN
+
+class collection;
+
+class MONGOCXX_API change_stream {
+   public:
+    class MONGOCXX_API iterator;
+
+    ///
+    /// Move constructs a change_stream.
+    ///
+    change_stream(change_stream&& other) noexcept;
+
+    ///
+    /// Move assigns a change_stream.
+    ///
+    change_stream& operator=(change_stream&& other) noexcept;
+
+    ///
+    /// Destroys a change_stream.
+    ///
+    ~change_stream();
+
+    ///
+    /// A change_stream::iterator points to the beginning of any
+    /// available notifications. Each call to begin() advances to the next
+    /// available notification. The state of all iterators is tracked by the
+    /// change_stream itself, so advancing one iterator advances all iterators.
+    ///
+    /// change_stream::begin() and the increment operators are blocking operations.
+    /// They will not return until a notification is available, the max_await_time (from
+    /// the options::change_stream) milliseconds have elapsed, or a server
+    /// error is encountered.
+    ///
+    /// When change_stream.begin() == change_stream.end(), no notifications
+    /// are available. Each call to change_stream.begin() checks again for
+    /// newly-available notifications.
+    ///
+    /// @return
+    ///   The change_stream::iterator
+    /// @exception
+    ///   Throws mongocxx::query_exception if the query failed.
+    ///
+    iterator begin();
+
+    ///
+    /// A change_stream::iterator indicating stream exhaustion, meaning that
+    /// no notifications are available from the stream.
+    ///
+    /// @return
+    ///   The change_stream::iterator indicating exhaustion
+    ///
+    iterator end();
+
+   private:
+    friend class collection;
+    friend class change_stream::iterator;
+
+    MONGOCXX_PRIVATE change_stream(void* change_stream_ptr);
+
+    class MONGOCXX_PRIVATE impl;
+    std::unique_ptr<impl> _impl;
+};
+
+class MONGOCXX_API change_stream::iterator {
+   public:
+    // Support input-iterator (caveat of post-increment returning void)
+    using difference_type = long;
+    using value_type = const bsoncxx::document::view;
+    using pointer = std::add_pointer<value_type>::type;
+    using reference = std::add_lvalue_reference<value_type>::type;
+    using iterator_category = std::input_iterator_tag;
+
+    ///
+    /// Default-construct an iterator.
+    /// This is equivalent to change_stream::end()
+    ///
+    iterator();
+
+    ///
+    /// Dereferences the view for the document currently being pointed to.
+    ///
+    const bsoncxx::document::view& operator*() const noexcept;
+
+    ///
+    /// Accesses a member of the dereferenced document currently being pointed to.
+    ///
+    const bsoncxx::document::view* operator->() const noexcept;
+
+    ///
+    /// Pre-increments the iterator to move to the next document.
+    ///
+    /// change_stream::begin() and increment operators are blocking operations.
+    /// They will not return until a notification is available, the max_await_time (from
+    /// the options::change_stream) miliseconds have elapsed, or a server
+    /// error is encountered.
+    ///
+    /// @throws mongocxx::query_exception if the query failed
+    ///
+    iterator& operator++();
+
+    ///
+    /// Post-increments the iterator to move to the next document.
+    ///
+    /// change_stream::begin() and increment operators are blocking operations.
+    /// They will not return until a notification is available, the max_await_time (from
+    /// the options::change_stream) miliseconds have elapsed, or a server
+    /// error is encountered.
+    ///
+    /// @throws mongocxx::query_exception if the query failed
+    ///
+    void operator++(int);
+
+   private:
+    friend class change_stream;
+
+    MONGOCXX_PRIVATE explicit iterator(change_stream* change_stream);
+
+    ///
+    /// @{
+    ///
+    /// Compare two iterators for (in)-equality.  Iterators compare equal if
+    /// they point to the same underlying change_stream or if both are exhausted.
+    ///
+    /// @relates iterator
+    ///
+    friend MONGOCXX_API bool MONGOCXX_CALL operator==(const change_stream::iterator&,
+                                                      const change_stream::iterator&) noexcept;
+
+    friend MONGOCXX_API bool MONGOCXX_CALL operator!=(const change_stream::iterator&,
+                                                      const change_stream::iterator&) noexcept;
+    ///
+    /// @}
+    ///
+
+    MONGOCXX_PRIVATE bool is_exhausted() const;
+
+    change_stream* _change_stream;
+};
+
+MONGOCXX_INLINE_NAMESPACE_END
+}  // namespace mongocxx
+
+#include <mongocxx/config/postlude.hpp>

src/mongocxx/collection.cpp

@@ -138,6 +138,35 @@ guard<T> make_guard(T&& t) {
     return guard<T>{std::forward<T>(t)};
 }
 
+// TODO: Consider extending the builders to directly accept optional values.
+template <typename T>
+inline void append_if(bsoncxx::builder::basic::document& doc,
+                      const std::string& key,
+                      const mongocxx::stdx::optional<T>& opt) {
+    if (opt) {
+        doc.append(bsoncxx::builder::basic::kvp(key, opt.value()));
+    }
+}
+
+bsoncxx::document::value as_bson(const mongocxx::options::change_stream& cs) {
+    // Construct new bson rep each time since values may change after this is called.
+    bsoncxx::builder::basic::document out{};
+
+    append_if(out, "fullDocument", cs.full_document());
+    append_if(out, "resumeAfter", cs.resume_after());
+    append_if(out, "batchSize", cs.batch_size());
+
+    if (cs.max_await_time()) {
+        auto count = cs.max_await_time().value().count();
+        if ((count < 0) || (count >= std::numeric_limits<std::uint32_t>::max())) {
+            throw mongocxx::logic_error{mongocxx::error_code::k_invalid_parameter};
+        }
+        out.append(bsoncxx::builder::basic::kvp("maxAwaitTimeMS", count));
+    }
+
+    return out.extract();
+}
+
 }  // namespace
 
 namespace mongocxx {
@@ -963,6 +992,21 @@ class write_concern collection::write_concern() const {
     return wc;
 }
 
+class change_stream collection::watch(const options::change_stream& options) {
+    return watch(pipeline{}, options);
+}
+
+class change_stream collection::watch(const pipeline& pipe, const options::change_stream& options) {
+    scoped_bson_t pipeline_bson{bsoncxx::document::view(pipe._impl->view_array())};
+
+    scoped_bson_t options_bson;
+    options_bson.init_from_static(as_bson(options));
+
+    // NOTE: collection_watch copies what it needs so we're safe to destroy our copies.
+    return change_stream{libmongoc::collection_watch(
+        _get_impl().collection_t, pipeline_bson.bson(), options_bson.bson())};
+}
+
 class index_view collection::indexes() {
     return index_view{_get_impl().collection_t};
 }