update to version 1.6.1

dht_proxy_client: fix getCallback for push notifications

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 6.0)
+set (opendht_VERSION_MINOR 6.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -20,7 +20,7 @@ option (OPENDHT_PROXY_SERVER "Enable DHT proxy server, use Restbed and jsoncpp"
 option (OPENDHT_PUSH_NOTIFICATIONS "Enable push notifications support" OFF)
 option (OPENDHT_PROXY_SERVER_IDENTITY "Allow clients to use the node identity" OFF)
 option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restbed and jsoncpp" OFF)
-option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
+option (OPENDHT_INDEX "Build DHT indexation feature" ON)
 
 find_package(Doxygen)
 option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentation (requires Doxygen)" ${DOXYGEN_FOUND})

README.md

@@ -4,14 +4,14 @@
     <a id="user-content-opendht-" class="anchor" href="/savoirfairelinux/opendht/blob/master/README.md#opendht-" aria-hidden="true"></a>OpenDHT
 </h1>
 
-A lightweight C++11 Distributed Hash Table implementation originally based on https://github.com/jech/dht by Juliusz Chroboczek.
+A lightweight C++11 Distributed Hash Table implementation.
 
- * Light and fast C++11 Kademlia DHT library.
+ * Light and fast C++11 Kademlia DHT library
  * Distributed shared key->value data-store
- * Clean and powerful distributed map API with storage of arbitrary binary values of up to 56 KB.
- * Optional public key cryptography layer providing data signature and encryption (using GnuTLS).
- * IPv4 and IPv6 support.
- * Python binding.
+ * Clean and powerful distributed map API with storage of arbitrary binary values of up to 56 KB
+ * Public key cryptography layer providing optional data signature and encryption (using GnuTLS)
+ * IPv4 and IPv6 support
+ * Python binding
 
 ## Documentation
 See the wiki: <https://github.com/savoirfairelinux/opendht/wiki>
@@ -93,23 +93,25 @@ for r in results:
 - Nettle 2.4+, a GnuTLS dependency for crypto.
 - (optional) restbed used for the REST API. commit fb84213e170bc171fecd825a8e47ed9f881a12cd (https://github.com/AmarOk1412/restbed/tree/async_read_until)
 - (optional) jsoncpp 1.7.4-3+, used for the REST API.
-- Build tested with GCC 5.2+ (GNU/Linux, Android, Windows with MinGW), Clang/LLVM (Linux, macOS).
+- Build tested with GCC 5.2+ (GNU/Linux, Windows with MinGW), Clang/LLVM (GNU/Linux, Android, macOS, iOS).
 - Build tested with Microsoft Visual Studio 2015
 
 ## Contact
 
 IRC: join us on Freenode at [`#opendht`](https://webchat.freenode.net/?channels=%23opendht).
 
-## Licence
-Copyright (C) 2014-2017 Savoir-faire Linux Inc.
+## License
+Copyright (C) 2014-2018 Savoir-faire Linux Inc.
 
 OpenDHT is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version.
 
 This program 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.
 
-See COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html for the full GPLv3 licence.
+See COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html for the full GPLv3 license.
 
-This project is independent from another project called OpenDHT (Sean Rhea. Ph.D. Thesis, 2005), now extinct.
+## Acknowledgements
+This project was originally based on https://github.com/jech/dht by Juliusz Chroboczek.
+It is independent from another project called OpenDHT (Sean Rhea. Ph.D. Thesis, 2005), now extinct.
 
 ## Donations
 We gratefully accept Bitcoin donations to support OpenDHT development at: `bitcoin:3EykSd1An888efq4Bq3KaV3hJ3JQ4FPnwm`.

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 6)
-m4_define([opendht_patch_version], 0)
+m4_define([opendht_patch_version], 1)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 
@@ -25,7 +25,7 @@ AC_PROG_CXX
 AM_PROG_AR
 
 dnl Check for logs
-AC_ARG_ENABLE([logs], [AS_HELP_STRING([--disable-logs], [Disble DHT logs])])
+AC_ARG_ENABLE([logs], [AS_HELP_STRING([--disable-logs], [Disable DHT logs])])
 AS_IF([test "x$enable_logs" != "xno"], [
 	AC_DEFINE([OPENDHT_LOG], [true], [Define if DHT logs are enabled])
 ], [
@@ -33,9 +33,9 @@ AS_IF([test "x$enable_logs" != "xno"], [
 ])
 
 dnl Check for indexation
-AC_ARG_ENABLE([indexation], [AS_HELP_STRING([--enable-indexation], [Enable DHT indexation])])
-AM_CONDITIONAL(ENABLE_INDEXATION, test x$enable_indexation == xyes)
-AS_IF([test "x$enable_indexation" = "xyes"], [
+AC_ARG_ENABLE([indexation], [AS_HELP_STRING([--disable-indexation], [Disable DHT indexation])])
+AM_CONDITIONAL(ENABLE_INDEXATION, test x$enable_indexation != "xno")
+AS_IF([test "x$enable_indexation" != "xno"], [
 	AC_DEFINE([OPENDHT_INDEXATION], [1], [Define if DHT indexation is enabled])
 ])
 

include/opendht/dht.h

@@ -99,11 +99,10 @@ public:
     bool isRunning(sa_family_t af = 0) const;
 
     virtual void registerType(const ValueType& type) {
-        types[type.id] = type;
+        types.registerType(type);
     }
     const ValueType& getType(ValueType::Id type_id) const {
-        const auto& t_it = types.find(type_id);
-        return (t_it == types.end()) ? ValueType::USER_DATA : t_it->second;
+        return types.getType(type_id);
     }
 
     /**
@@ -350,7 +349,7 @@ private:
     uint64_t oldsecret {};
 
     // registred types
-    std::map<ValueType::Id, ValueType> types;
+    TypeStore types;
 
     // are we a bootstrap node ?
     // note: Any running node can be used as a bootstrap node.

include/opendht/dht_proxy_client.h

@@ -297,6 +297,7 @@ private:
         Value::Filter filterChain;
         std::thread thread;
         unsigned callbackId;
+        std::shared_ptr<bool> isCanceledViaClose; // NOTE: unused if using push notifications
         std::shared_ptr<unsigned> pushNotifToken; // NOTE: unused if not using push notifications
     };
     std::vector<Listener> listeners_;

include/opendht/value.h

@@ -109,6 +109,19 @@ struct OPENDHT_PUBLIC ValueType {
     EditPolicy editPolicy {DEFAULT_EDIT_POLICY};
 };
 
+class TypeStore {
+public:
+    void registerType(const ValueType& type) {
+        types[type.id] = type;
+    }
+    const ValueType& getType(ValueType::Id type_id) const {
+        const auto& t_it = types.find(type_id);
+        return (t_it == types.end()) ? ValueType::USER_DATA : t_it->second;
+    }
+private:
+    std::map<ValueType::Id, ValueType> types {};
+};
+
 /**
  * A "value" is data potentially stored on the Dht, with some metadata.
  *
@@ -174,6 +187,15 @@ struct OPENDHT_PUBLIC Value
                 return f1(v) or f2(v);
             };
         }
+        std::vector<Sp<Value>> filter(const std::vector<Sp<Value>>& values) {
+            if (not (*this))
+                return values;
+            std::vector<Sp<Value>> ret;
+            for (const auto& v : values)
+                if ((*this)(v))
+                    ret.emplace_back(v);
+            return ret;
+        }
     };
 
     /* Sneaky functions disguised in classes */

src/dht.cpp

@@ -811,6 +811,7 @@ Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter f, Where where)
         }
         if (!newvals.empty()) {
             if (!cb(newvals)) {
+                // cancelListen is useful here, because we need to cancel on IPv4 and 6
                 cancelListen(id, token);
                 return false;
             }

src/dht_proxy_client.cpp

@@ -200,7 +200,7 @@ DhtProxyClient::get(const InfoHash& key, const GetCallback& cb, DoneCallback don
                             if ((not filterChain or filterChain(*value)) && cb) {
                                 std::lock_guard<std::mutex> lock(lockCallbacks);
                                 callbacks_.emplace_back([cb, value, finished]() {
-                                    if (not cb({value}))
+                                    if (not *finished and not cb({value}))
                                         *finished = true;
                                 });
                                 loopSignal_();
@@ -445,13 +445,41 @@ DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter filter
 
     Listener l;
     ++listener_token_;
+    auto isCanceled = std::make_shared<bool>(false);
+
+    // Wrap cb in another callback to avoid value duplication.
+    auto vals = std::make_shared<std::map<Value::Id, Sp<Value>>>();
+    auto gcb = [=](const std::vector<Sp<Value>>& values) {
+        std::vector<Sp<Value>> newvals;
+        for (const auto& v : values) {
+            auto it = vals->find(v->id);
+            if (it == vals->cend() || !(*it->second == *v))
+                newvals.push_back(v);
+        }
+        if (!newvals.empty()) {
+            if (not *isCanceled and !cb(newvals)) {
+                *isCanceled = true;
+                cancelListen(key, listener_token_);
+                return false;
+            }
+            for (const auto& v : newvals) {
+                auto it = vals->emplace(v->id, v);
+                if (not it.second)
+                    it.first->second = v;
+            }
+        }
+        return true;
+    };
+
     l.key = key.toString();
     l.token = listener_token_;
     l.req = req;
-    l.cb = cb;
+    l.cb = gcb;
     l.callbackId = callbackId;
     l.pushNotifToken = pushNotifToken;
     l.filterChain = std::move(filterChain);
+    l.isCanceledViaClose = isCanceled;
+    std::weak_ptr<bool> isCanceledViaClose(l.isCanceledViaClose);
     l.thread = std::thread([=]()
         {
             auto settings = std::make_shared<restbed::Settings>();
@@ -470,7 +498,7 @@ DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter filter
             };
             auto state = std::make_shared<State>();
             restbed::Http::async(req,
-                [this, filterChain, cb, pushNotifToken, state](const std::shared_ptr<restbed::Request>& req,
+                [this, filterChain, gcb, pushNotifToken, state](const std::shared_ptr<restbed::Request>& req,
                                                                const std::shared_ptr<restbed::Response>& reply) {
                 auto code = reply->get_status_code();
                 if (code == 200) {
@@ -505,12 +533,11 @@ DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter filter
                                 auto* char_data = reinterpret_cast<const char*>(&body[0]);
                                 if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
                                     auto value = std::make_shared<Value>(json);
-                                    if ((not filterChain or filterChain(*value)) && cb)  {
+                                    if (not filterChain or filterChain(*value))  {
                                         std::lock_guard<std::mutex> lock(lockCallbacks);
-                                        callbacks_.emplace_back([cb, value, state]() {
-                                            if (not state->cancel and not cb({value})) {
+                                        callbacks_.emplace_back([gcb, value, state]() {
+                                            if (not state->cancel and not gcb({value}))
                                                 state->cancel = true;
-                                            }
                                         });
                                         loopSignal_();
                                     }
@@ -526,7 +553,8 @@ DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter filter
                     state->ok = false;
                 }
             }, settings).get();
-            if (not state->ok) {
+            auto isCanceledNormally = isCanceledViaClose.lock();
+            if (not state->ok and isCanceledNormally and not *isCanceledNormally) {
                 opFailed();
             }
         }
@@ -579,6 +607,7 @@ DhtProxyClient::cancelListen(const InfoHash&, size_t token)
                 // Just stop the request
                 if (listener.thread.joinable()) {
                     // Close connection to stop listener?
+                    *(listener.isCanceledViaClose) = true;
                     if (listener.req)
                         restbed::Http::close(listener.req);
                     listener.thread.join();

src/value.cpp

@@ -176,8 +176,13 @@ Value::msgpack_unpack_body(const msgpack::object& o)
 Value::Value(Json::Value& json)
 {
    try {
-       if (json.isMember("id"))
-           id = ValueType::Id(json["id"].asInt());
+       if (json.isMember("id")) {
+           if (json["id"].isString()) {
+               id = Value::Id(std::stoull(json["id"].asString()));
+           } else {
+               id = Value::Id(json["id"].asLargestUInt());
+           }
+       }
    } catch (...) { }
    if (json.isMember("cypher")) {
        auto cypherStr = json["cypher"].asString();