utils: add observer/observable templates

An observable is used to decouple an information producer from a consumer
(in the same way as a callback), while allowing multiple consumers (called
observers) to coexist and to manage their lifetime separately.

Two classes are introduced:

 observable: a producer class; when an observable is invoked all observers
        receive the information
 observer: a consumer class; receives information from a observable

Modelled after boost::signals2, with the following changes
 - all signals return void; information is passed from the producer to
   the consumer but not back
 - thread-unsafe
 - modern C++ without preprocessor hacks
 - connection lifetime is always managed rather than leaked by default
 - renamed to avoid the funky "slot" name
Message-Id: <20180709172726.5079-1-avi@scylladb.com>

configure.py

@@ -304,6 +304,7 @@ def endswith(self, end):
     'tests/partition_data_test',
     'tests/reusable_buffer_test',
     'tests/multishard_writer_test',
+    'tests/observable_test',
 ]
 
 perf_tests = [
@@ -742,6 +743,7 @@ def endswith(self, end):
     'tests/imr_test',
     'tests/partition_data_test',
     'tests/reusable_buffer_test',
+    'tests/observable_test',
 ])
 
 tests_not_using_seastar_test_framework = set([

tests/observable_test.cc

@@ -0,0 +1,71 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define BOOST_TEST_MODULE observable_test
+
+#include <boost/test/unit_test.hpp>
+
+
+#include "../utils/observable.hh"
+
+using namespace utils;
+
+BOOST_AUTO_TEST_CASE(test_basic_functionality) {
+    observable<int> pub;
+    int v1 = 0, v2 = 0;
+    observer<int> sub1 = pub.observe([&] (int x) { v1 = x; });
+    observer<int> sub2 = pub.observe([&] (int x) { v2 = x; });
+    pub(7);
+    BOOST_REQUIRE_EQUAL(v1, 7);
+    BOOST_REQUIRE_EQUAL(v2, 7);
+    sub1.disconnect();
+    pub(3);
+    BOOST_REQUIRE_EQUAL(v1, 7);
+    BOOST_REQUIRE_EQUAL(v2, 3);
+    sub1 = std::move(sub2);
+    pub(4);
+    BOOST_REQUIRE_EQUAL(v1, 7);
+    BOOST_REQUIRE_EQUAL(v2, 4);
+    pub = observable<int>();
+    pub(5);
+    BOOST_REQUIRE_EQUAL(v1, 7);
+    BOOST_REQUIRE_EQUAL(v2, 4);
+}
+
+BOOST_AUTO_TEST_CASE(test_exceptions) {
+    observable<> pub;
+    bool saw1 = false;
+    observer<> sub1 = pub.observe([&] { saw1 = true; });
+    observer<> sub2 = pub.observe([&] { throw 2; });
+    bool saw3 = false;
+    observer<> sub3 = pub.observe([&] { saw3 = true; });
+    observer<> sub4 = pub.observe([&] { throw 4; });
+    bool caught = false;
+    try {
+        pub();
+    } catch (int v) {
+        BOOST_REQUIRE(saw1);
+        BOOST_REQUIRE(saw3);
+        BOOST_REQUIRE(v == 2 || v == 4);
+        caught = true;
+    }
+    BOOST_REQUIRE(caught);
+}

utils/observable.hh

@@ -0,0 +1,132 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <functional>
+#include <vector>
+#include <boost/range/algorithm/replace.hpp>
+#include <boost/range/algorithm/remove.hpp>
+
+namespace utils {
+
+template <typename... Args>
+class observable {
+public:
+    class observer;
+private:
+    std::vector<observer*> _observers;
+public:
+    class observer {
+        friend class observable;
+        observable* _observable;
+        std::function<void (Args...)> _callback;
+    private:
+        void moved(observer* from) {
+            if (_observable) {
+                _observable->moved(from, this);
+            }
+        }
+    public:
+        observer(observable* o, std::function<void (Args...)> callback) noexcept
+                : _observable(o), _callback(std::move(callback)) {
+        }
+        observer(observer&& o) noexcept
+                : _observable(std::exchange(o._observable, nullptr))
+                , _callback(std::move(o._callback)) {
+            moved(&o);
+        }
+        observer& operator=(observer&& o) noexcept {
+            if (this != &o) {
+                disconnect();
+                _observable = std::exchange(o._observable, nullptr);
+                _callback = std::move(o._callback);
+                moved(&o);
+            }
+            return *this;
+        }
+        ~observer() {
+            disconnect();
+        }
+        void disconnect() {
+            if (_observable) {
+                _observable->destroyed(this);
+            }
+        }
+    };
+    friend class observer;
+private:
+    void destroyed(observer* dead) {
+        _observers.erase(boost::remove(_observers, dead), _observers.end());
+    }
+    void moved(observer* from, observer* to) {
+        boost::replace(_observers, from, to);
+    }
+    void update_observers(observable* ob) {
+        for (auto&& c : _observers) {
+            c->_observable = ob;
+        }
+    }
+public:
+    observable() = default;
+    observable(observable&& o) noexcept
+            : _observers(std::move(o._observers)) {
+        update_observers(this);
+    }
+    observable& operator=(observable&& o) noexcept {
+        if (this != &o) {
+            update_observers(nullptr);
+            _observers = std::move(o._observers);
+            update_observers(this);
+        }
+        return *this;
+    }
+    ~observable() {
+        update_observers(nullptr);
+    }
+    // Send args to all connected observers
+    void operator()(Args... args) const {
+        std::exception_ptr e;
+        for (auto&& ob : _observers) {
+            try {
+                ob->_callback(args...);
+            } catch (...) {
+                if (!e) {
+                    e = std::current_exception();
+                }
+            }
+        }
+        if (e) {
+            std::rethrow_exception(std::move(e));
+        }
+    }
+    // Adds an observer to an observable
+    observer observe(std::function<void (Args...)> callback) {
+        observer ob(this, std::move(callback));
+        _observers.push_back(&ob);
+        return ob;
+    }
+};
+
+template <typename... Args>
+using observer = typename observable<Args...>::observer;
+
+}