Update README.md

README.md

@@ -16,7 +16,7 @@ Values are distributed over the network, with redundancy.
  * IPv4 and IPv6 support
  * Clean and powerful C++14 map API
  * Python 3 bindings
- * REST API
+ * REST API with optional HTTP client+server with push notification support
 
 ## Documentation
 See the wiki: <https://github.com/savoirfairelinux/opendht/wiki>

c/opendht.cpp

@@ -124,7 +124,7 @@ void dht_publickey_delete(dht_publickey* pk) {
 
 int dht_publickey_export(const dht_publickey* pk, char* out, size_t* outlen) {
     const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
-    return gnutls_pubkey_export(pkey->pk, GNUTLS_X509_FMT_DER, out, outlen);
+    return pkey->pack((uint8_t*)out, outlen);
 }
 
 dht_infohash dht_publickey_get_id(const dht_publickey* pk) {

docker/DockerfileDeps

@@ -22,8 +22,8 @@ RUN echo "*** Building nodejs/http_parser dependency ***" \
 
 # libasio-dev (1.10.6-3) is too old
 RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
-    && tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
+    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
+    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
     && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
     && make install \
     && cd ../../ && rm -rf asio*

docker/DockerfileDepsBionic

@@ -5,13 +5,13 @@ RUN echo "APT::Acquire::Retries \"3\";" > /etc/apt/apt.conf.d/80-retries
 RUN apt-get update && apt-get install -y \
         apt-transport-https build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
         libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev \
+        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libssl-dev \
     && apt-get clean
 
 # libasio-dev (1.10) is too old
 RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
-    && tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
+    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
+    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
     && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
     && make install \
     && cd ../../ && rm -rf asio*

docker/DockerfileDepsLlvm

@@ -26,8 +26,8 @@ RUN echo "*** Building nodejs/http_parser dependency ***" \
 
 # libasio-dev (1.10.6-3) is too old
 RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
-    && tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
+    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
+    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
     && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
     && make install \
     && cd ../../ && rm -rf asio*

docker/DockerfileTravis

@@ -3,5 +3,5 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 
 COPY . /root/opendht
 RUN cd /root/opendht && mkdir build && cd build \
-	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON ..  \
+	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_C=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON ..  \
 	&& make -j8 && ./opendht_unit_tests && make install

docker/DockerfileTravisLlvm

@@ -3,5 +3,5 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 
 COPY . /root/opendht
 RUN cd /root/opendht && mkdir build && cd build \
-	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_TESTS=ON .. \
+	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_C=On -DOPENDHT_TESTS=ON .. \
 	&& make -j8 && ./opendht_unit_tests && make install

include/opendht/crypto.h

@@ -114,6 +114,7 @@ struct OPENDHT_PUBLIC PublicKey
     }
 
     void pack(Blob& b) const;
+    int pack(uint8_t* out, size_t* out_len) const;
     void unpack(const uint8_t* dat, size_t dat_size);
 
     std::string toString() const;
@@ -127,7 +128,7 @@ struct OPENDHT_PUBLIC PublicKey
         p.pack_bin_body((const char*)b.data(), b.size());
     }
 
-    void msgpack_unpack(msgpack::object o);
+    void msgpack_unpack(const msgpack::object& o);
 
     gnutls_digest_algorithm_t getPreferredDigest() const;
 
@@ -225,7 +226,7 @@ public:
         p.pack_bin_body((const char*)b.data(), b.size());
     }
 
-    void msgpack_unpack(msgpack::object o);
+    void msgpack_unpack(const msgpack::object& o);
 
     void revoke(const Certificate& crt, time_point t = time_point::min());
 
@@ -423,7 +424,7 @@ struct OPENDHT_PUBLIC Certificate {
         p.pack_bin_body((const char*)b.data(), b.size());
     }
 
-    void msgpack_unpack(msgpack::object o);
+    void msgpack_unpack(const msgpack::object& o);
 
     explicit operator bool() const { return cert; }
     PublicKey getPublicKey() const;

include/opendht/default_types.h

@@ -189,7 +189,7 @@ public:
 #endif
     }
 
-    virtual void msgpack_unpack(msgpack::object o)
+    virtual void msgpack_unpack(const msgpack::object& o)
     {
         if (o.type != msgpack::type::ARRAY) throw msgpack::type_error();
         if (o.via.array.size < 2) throw msgpack::type_error();
@@ -229,7 +229,7 @@ public:
         pk.pack_bin_body((const char*)addr.get(), addr.getLength());
     }
 
-    virtual void msgpack_unpack(msgpack::object o)
+    virtual void msgpack_unpack(const msgpack::object& o)
     {
         if (o.type == msgpack::type::BIN)
             addr = {(sockaddr*)o.via.bin.ptr, (socklen_t)o.via.bin.size};

include/opendht/dht_proxy_client.h

@@ -303,7 +303,7 @@ private:
     /**
      * Send Listen with httpClient_
      */
-    void sendListen(const restinio::http_request_header_t header, const ValueCallback& cb,
+    void sendListen(const restinio::http_request_header_t& header, const ValueCallback& cb,
                     const Sp<OperationState>& opstate, Listener& listener, ListenMethod method = ListenMethod::LISTEN);
     void handleResubscribe(const asio::error_code& ec, const InfoHash& key,
                            const size_t token, std::shared_ptr<OperationState> opstate);
@@ -327,6 +327,7 @@ private:
     std::shared_ptr<dht::crypto::Certificate> serverCertificate_;
     //std::pair<std::string, std::string> serverHostService_;
     std::string pushClientId_;
+    std::string pushSessionId_;
 
     mutable std::mutex lockCurrentProxyInfos_;
     NodeStatus statusIpv4_ {NodeStatus::Disconnected};

include/opendht/dht_proxy_server.h

@@ -281,7 +281,7 @@ private:
      * @param key of the device
      * @param json, the content to send
      */
-    void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid);
+    void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid, bool highPriority);
 
     /**
      * Send push notification with an expire timeout.
@@ -360,6 +360,7 @@ private:
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
     struct Listener {
         std::string clientId;
+        std::string sessionId;
         std::future<size_t> internalToken;
         std::unique_ptr<asio::steady_timer> expireTimer;
         std::unique_ptr<asio::steady_timer> expireNotifyTimer;

include/opendht/utils.h

@@ -133,7 +133,7 @@ using Blob = std::vector<uint8_t>;
 /**
  * Provides backward compatibility with msgpack 1.0
  */
-OPENDHT_PUBLIC Blob unpackBlob(msgpack::object& o);
+OPENDHT_PUBLIC Blob unpackBlob(const msgpack::object& o);
 
 template <typename Type>
 Blob
@@ -153,6 +153,7 @@ unpackMsg(Blob b) {
 
 msgpack::unpacked unpackMsg(Blob b);
 
-msgpack::object* findMapValue(msgpack::object& map, const std::string& key);
+msgpack::object* findMapValue(const msgpack::object& map, const char* key);
+msgpack::object* findMapValue(const msgpack::object& map, const std::string& key);
 
 } // namespace dht

include/opendht/value.h

@@ -43,6 +43,19 @@
 
 namespace dht {
 
+static const std::string VALUE_KEY_ID("id");
+static const std::string VALUE_KEY_DAT("dat");
+static const std::string VALUE_KEY_PRIO("p");
+static const std::string VALUE_KEY_SIGNATURE("sig");
+
+static const std::string VALUE_KEY_SEQ("seq");
+static const std::string VALUE_KEY_DATA("data");
+static const std::string VALUE_KEY_OWNER("owner");
+static const std::string VALUE_KEY_TYPE("type");
+static const std::string VALUE_KEY_TO("to");
+static const std::string VALUE_KEY_BODY("body");
+static const std::string VALUE_KEY_USERTYPE("utype");
+
 struct Value;
 struct Query;
 
@@ -390,7 +403,7 @@ struct OPENDHT_PUBLIC Value
      * Build a value from a json object
      * @param json
      */
-    Value(Json::Value& json);
+    Value(const Json::Value& json);
 #endif
 
     template <typename Type>
@@ -408,7 +421,9 @@ struct OPENDHT_PUBLIC Value
 
     Value(Value&& o) noexcept
      : id(o.id), owner(std::move(o.owner)), recipient(o.recipient),
-     type(o.type), data(std::move(o.data)), user_type(std::move(o.user_type)), seq(o.seq), signature(std::move(o.signature)), cypher(std::move(o.cypher)) {}
+     type(o.type), data(std::move(o.data)), user_type(std::move(o.user_type)), seq(o.seq)
+     , signature(std::move(o.signature)), cypher(std::move(o.cypher))
+     , priority(o.priority) {}
 
     template <typename Type>
     Value(const Type& vs)
@@ -485,17 +500,17 @@ struct OPENDHT_PUBLIC Value
         bool has_owner = owner && *owner;
         pk.pack_map((user_type.empty()?0:1) + (has_owner?(recipient ? 5 : 4):2));
         if (has_owner) { // isSigned
-            pk.pack(std::string("seq"));   pk.pack(seq);
-            pk.pack(std::string("owner")); owner->msgpack_pack(pk);
+            pk.pack(VALUE_KEY_SEQ);   pk.pack(seq);
+            pk.pack(VALUE_KEY_OWNER); owner->msgpack_pack(pk);
             if (recipient) {
-                pk.pack(std::string("to")); pk.pack(recipient);
+                pk.pack(VALUE_KEY_TO); pk.pack(recipient);
             }
         }
-        pk.pack(std::string("type"));  pk.pack(type);
-        pk.pack(std::string("data"));  pk.pack_bin(data.size());
-                                       pk.pack_bin_body((const char*)data.data(), data.size());
+        pk.pack(VALUE_KEY_TYPE);  pk.pack(type);
+        pk.pack(VALUE_KEY_DATA);  pk.pack_bin(data.size());
+                                  pk.pack_bin_body((const char*)data.data(), data.size());
         if (not user_type.empty()) {
-            pk.pack(std::string("utype")); pk.pack(user_type);
+            pk.pack(VALUE_KEY_USERTYPE); pk.pack(user_type);
         }
     }
 
@@ -507,10 +522,10 @@ struct OPENDHT_PUBLIC Value
             pk.pack_bin_body((const char*)cypher.data(), cypher.size());
         } else {
             pk.pack_map(isSigned() ? 2 : 1);
-            pk.pack(std::string("body")); msgpack_pack_to_sign(pk);
+            pk.pack(VALUE_KEY_BODY); msgpack_pack_to_sign(pk);
             if (isSigned()) {
-                pk.pack(std::string("sig")); pk.pack_bin(signature.size());
-                                             pk.pack_bin_body((const char*)signature.data(), signature.size());
+                pk.pack(VALUE_KEY_SIGNATURE); pk.pack_bin(signature.size());
+                                              pk.pack_bin_body((const char*)signature.data(), signature.size());
             }
         }
     }
@@ -518,9 +533,12 @@ struct OPENDHT_PUBLIC Value
     template <typename Packer>
     void msgpack_pack(Packer& pk) const
     {
-        pk.pack_map(2);
-        pk.pack(std::string("id"));  pk.pack(id);
-        pk.pack(std::string("dat")); msgpack_pack_to_encrypt(pk);
+        pk.pack_map(2 + (priority?1:0));
+        pk.pack(VALUE_KEY_ID);  pk.pack(id);
+        pk.pack(VALUE_KEY_DAT); msgpack_pack_to_encrypt(pk);
+        if (priority) {
+            pk.pack(VALUE_KEY_PRIO);  pk.pack(priority);
+        }
     }
 
     template <typename Packer>
@@ -551,7 +569,7 @@ struct OPENDHT_PUBLIC Value
             }
     }
 
-    void msgpack_unpack(msgpack::object o);
+    void msgpack_unpack(const msgpack::object& o);
     void msgpack_unpack_body(const msgpack::object& o);
     Blob getPacked() const {
         msgpack::sbuffer buffer;
@@ -602,6 +620,13 @@ struct OPENDHT_PUBLIC Value
      */
     Blob cypher {};
 
+    /**
+     * Priority of this value (used for push notifications)
+     * 0 = high priority
+     * 1 = normal priority
+     */
+    unsigned priority {0};
+
 private:
     friend class SecureDht;
     /* Cache for crypto ops */
@@ -658,7 +683,7 @@ struct OPENDHT_PUBLIC FieldValue
         }
     }
 
-    void msgpack_unpack(msgpack::object msg) {
+    void msgpack_unpack(const msgpack::object& msg) {
         hashValue = {};
         blobValue.clear();
 

src/base64.cpp

@@ -131,7 +131,7 @@ base64_encode(const std::vector<unsigned char>& str)
     return base64_encode(str.cbegin(), str.cend());
 }
 
-std::string
+std::vector<unsigned char>
 base64_decode(const std::string& str)
 {
     size_t output_length = str.length() / 4 * 3 + 2;
@@ -139,5 +139,5 @@ base64_decode(const std::string& str)
     output.resize(output_length);
     base64_decode(str.data(), str.size(), output.data(), &output_length);
     output.resize(output_length);
-    return std::string(output.begin(), output.end());
+    return output;
 }

src/base64.h

@@ -33,4 +33,4 @@ std::string base64_encode(const std::vector<unsigned char>& str);
  * @param str the input buffer
  * @return a base64-decoded buffer
  */
-std::string base64_decode(const std::string& str);
+std::vector<unsigned char> base64_decode(const std::string& str);

src/crypto.cpp

@@ -412,12 +412,18 @@ PublicKey::pack(Blob& b) const
         throw CryptoException(std::string("Could not export public key: null key"));
     std::vector<uint8_t> tmp(2048);
     size_t sz = tmp.size();
-    if (int err = gnutls_pubkey_export(pk, GNUTLS_X509_FMT_DER, tmp.data(), &sz))
+    if (int err = pack(tmp.data(), &sz))
         throw CryptoException(std::string("Could not export public key: ") + gnutls_strerror(err));
     tmp.resize(sz);
     b.insert(b.end(), tmp.begin(), tmp.end());
 }
 
+int
+PublicKey::pack(uint8_t* out, size_t* out_len) const
+{
+    return gnutls_pubkey_export(pk, GNUTLS_X509_FMT_DER, out, out_len);
+}
+
 void
 PublicKey::unpack(const uint8_t* data, size_t data_size)
 {
@@ -447,7 +453,7 @@ PublicKey::toString() const
 }
 
 void
-PublicKey::msgpack_unpack(msgpack::object o)
+PublicKey::msgpack_unpack(const msgpack::object& o)
 {
     if (o.type == msgpack::type::BIN)
         unpack((const uint8_t*)o.via.bin.ptr, o.via.bin.size);
@@ -792,7 +798,7 @@ Certificate::unpack(const uint8_t* dat, size_t dat_size)
 }
 
 void
-Certificate::msgpack_unpack(msgpack::object o)
+Certificate::msgpack_unpack(const msgpack::object& o)
 {
     if (o.type == msgpack::type::BIN)
         unpack((const uint8_t*)o.via.bin.ptr, o.via.bin.size);
@@ -1252,7 +1258,7 @@ RevocationList::unpack(const uint8_t* dat, size_t dat_size)
 }
 
 void
-RevocationList::msgpack_unpack(msgpack::object o)
+RevocationList::msgpack_unpack(const msgpack::object& o)
 {
     try {
         if (o.type == msgpack::type::BIN)

src/dht_proxy_client.cpp

@@ -93,6 +93,16 @@ private:
     std::string line_ {};
 };
 
+std::string
+getRandomSessionId(size_t length = 8) {
+    static constexpr const char chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~";
+    std::string str(length, 0);
+    crypto::random_device rdev;
+    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 1);
+    std::generate_n( str.begin(), length, [&]{ return chars[dist(rdev)]; } );
+    return str;
+}
+
 DhtProxyClient::DhtProxyClient() {}
 
 DhtProxyClient::DhtProxyClient(
@@ -101,7 +111,8 @@ DhtProxyClient::DhtProxyClient(
         const std::string& pushClientId, std::shared_ptr<dht::Logger> logger)
     : proxyUrl_(serverHost)
     , clientIdentity_(clientIdentity), serverCertificate_(serverCA)
-    , pushClientId_(pushClientId), loopSignal_(signal)
+    , pushClientId_(pushClientId), pushSessionId_(getRandomSessionId())
+    , loopSignal_(signal)
     , jsonReader_(Json::CharReaderBuilder{}.newCharReader())
     , logger_(logger)
 {
@@ -929,7 +940,7 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash&
 }
 
 void
-DhtProxyClient::sendListen(const restinio::http_request_header_t header,
+DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
                            const ValueCallback& cb,
                            const Sp<OperationState>& opstate,
                            Listener& listener, ListenMethod method)
@@ -1118,6 +1129,12 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
         statusIpv6_ = NodeStatus::Connected;
     }
     try {
+        auto sessionId = notification.find("s");
+        if (sessionId != notification.end() and sessionId->second != pushSessionId_) {
+            if (logger_)
+                logger_->d("[proxy:client] [push] ignoring push for other session");
+            return;
+        }
         std::lock_guard<std::mutex> lock(searchLock_);
         auto timeout = notification.find("timeout");
         if (timeout != notification.cend()) {
@@ -1162,7 +1179,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                 } else {
                     std::stringstream ss(expired->second);
                     std::vector<Value::Id> ids;
-                    while(ss.good()){
+                    while(ss.good()) {
                         std::string substr;
                         getline(ss, substr, ',');
                         ids.emplace_back(std::stoull(substr));
@@ -1237,6 +1254,7 @@ DhtProxyClient::getPushRequest(Json::Value& body) const
 {
     body["key"] = deviceKey_;
     body["client_id"] = pushClientId_;
+    body["session_id"] = pushSessionId_;
 #ifdef __ANDROID__
     body["platform"] = "android";
 #endif

src/dht_proxy_server.cpp

@@ -622,15 +622,16 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
     }
     try {
         std::string err;
-        Json::Value root;
+        Json::Value r;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
         auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
-        if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)){
+        if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)){
             auto response = initHttpResponse(
                 request->create_response(restinio::status_bad_request()));
             response.set_body(RESP_MSG_JSON_INCORRECT);
             return response.done();
         }
+        const Json::Value& root(r); // parse using const Json so [] never creates element
         auto pushToken = root["key"].asString();
         if (pushToken.empty()){
             auto response = initHttpResponse(
@@ -638,12 +639,12 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
             response.set_body(RESP_MSG_NO_TOKEN);
             return response.done();
         }
-        auto platform = root["platform"].asString();
-        auto isAndroid = platform == "android";
-        auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string();
+        auto isAndroid = root["platform"].asString() == "android";
+        auto clientId = root["client_id"].asString();
+        auto sessionId = root["session_id"].asString();
 
         if (logger_)
-            logger_->d("[proxy:server] [subscribe %s] [client %s]", infoHash.toString().c_str(), clientId.c_str());
+            logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());
         // ================ Search for existing listener ===================
         // start the timer
         auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
@@ -653,11 +654,11 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
         auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first;
         auto pushListeners = pushListener->second.listeners.emplace(infoHash, std::vector<Listener>{}).first;
 
-        for (auto &listener: pushListeners->second){
-            if (logger_)
-                logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
+        for (auto &listener: pushListeners->second) {
             // Found -> Resubscribe
-            if (listener.clientId == clientId){
+            if (listener.clientId == clientId) {
+                if (logger_)
+                    logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
                 // Reset timers
                 listener.expireTimer->expires_at(timeout);
                 listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
@@ -692,12 +693,13 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
 
         // Add listen on dht
         listener.internalToken = dht_->listen(infoHash,
-            [this, infoHash, pushToken, isAndroid, clientId]
+            [this, infoHash, pushToken, isAndroid, clientId, sessionId]
             (const std::vector<std::shared_ptr<Value>>& values, bool expired){
                 // Build message content
                 Json::Value json;
                 json["key"] = infoHash.toString();
                 json["to"] = clientId;
+                json["s"] = sessionId;
                 if (expired and values.size() < 2){
                     std::stringstream ss;
                     for(size_t i = 0; i < values.size(); ++i){
@@ -706,12 +708,15 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
                     }
                     json["exp"] = ss.str();
                 }
-                sendPushNotification(pushToken, std::move(json), isAndroid);
+                auto maxPrio = 1000u;
+                for (const auto& v : values)
+                    maxPrio = std::min(maxPrio, v->priority);
+                sendPushNotification(pushToken, std::move(json), isAndroid, !expired and maxPrio == 0);
                 return true;
             }
         );
         // Launch timers
-        auto &ctx = io_context();
+        auto& ctx = io_context();
         // expire notify
         if (!listener.expireNotifyTimer)
             listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(ctx, timeout - proxy::OP_MARGIN);
@@ -720,8 +725,9 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
         Json::Value json;
         json["timeout"] = infoHash.toString();
         json["to"] = clientId;
+        json["s"] = sessionId;
         listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this,
-                                               std::placeholders::_1, pushToken, json, isAndroid));
+                                               std::placeholders::_1, pushToken, std::move(json), isAndroid));
         // cancel push listen
         if (!listener.expireTimer)
             listener.expireTimer = std::make_unique<asio::steady_timer>(ctx, timeout);
@@ -802,7 +808,7 @@ DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const s
     }
     if (logger_)
         logger_->d("[proxy:server] [subscribe] sending put refresh to %s token", pushToken.c_str());
-    sendPushNotification(pushToken, std::move(json), isAndroid);
+    sendPushNotification(pushToken, std::move(json), isAndroid, false);
 }
 
 void
@@ -844,7 +850,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st
 }
 
 void
-DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid)
+DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid, bool highPriority)
 {
     if (pushServer_.empty())
         return;
@@ -868,7 +874,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
         notification["tokens"] = std::move(tokens);
         notification["platform"] = isAndroid ? 2 : 1;
         notification["data"] = std::move(json);
-        notification["priority"] = "high";
+        notification["priority"] = highPriority ? "high" : "normal";
         notification["time_to_live"] = 600;
 
         Json::Value notifications(Json::arrayValue);

src/utils.cpp

@@ -246,7 +246,7 @@ std::time_t to_time_t(time_point t) {
 }
 
 Blob
-unpackBlob(msgpack::object& o) {
+unpackBlob(const msgpack::object& o) {
     switch (o.type) {
     case msgpack::type::BIN:
         return {o.via.bin.ptr, o.via.bin.ptr+o.via.bin.size};
@@ -270,11 +270,25 @@ unpackMsg(Blob b) {
 }
 
 msgpack::object*
-findMapValue(msgpack::object& map, const std::string& key) {
+findMapValue(const msgpack::object& map, const char* key) {
     if (map.type != msgpack::type::MAP) throw msgpack::type_error();
     for (unsigned i = 0; i < map.via.map.size; i++) {
         auto& o = map.via.map.ptr[i];
-        if (o.key.type == msgpack::type::STR && o.key.as<std::string>() == key)
+        if (o.key.type == msgpack::type::STR 
+            && std::strncmp(o.key.via.str.ptr, key, o.key.via.str.size) == 0)
+            return &o.val;
+    }
+    return nullptr;
+}
+
+msgpack::object*
+findMapValue(const msgpack::object& map, const std::string& key) {
+    if (map.type != msgpack::type::MAP) throw msgpack::type_error();
+    for (unsigned i = 0; i < map.via.map.size; i++) {
+        auto& o = map.via.map.ptr[i];
+        if (o.key.type == msgpack::type::STR 
+            && key.size() == o.key.via.str.size
+            && std::strncmp(o.key.via.str.ptr, key.data(), o.key.via.str.size) == 0)
             return &o.val;
     }
     return nullptr;

src/value.cpp

@@ -87,20 +87,6 @@ ValueType::DEFAULT_STORE_POLICY(InfoHash, const std::shared_ptr<Value>& v, const
     return v->size() <= MAX_VALUE_SIZE;
 }
 
-msgpack::object*
-findMapValue(const msgpack::object& map, const std::string& key) {
-    if (map.type != msgpack::type::MAP) throw msgpack::type_error();
-    for (unsigned i = 0; i < map.via.map.size; i++) {
-        auto& o = map.via.map.ptr[i];
-        if(o.key.type != msgpack::type::STR)
-            continue;
-        if (o.key.as<std::string>() == key) {
-            return &o.val;
-        }
-    }
-    return nullptr;
-}
-
 size_t
 Value::size() const
 {
@@ -108,20 +94,24 @@ Value::size() const
 }
 
 void
-Value::msgpack_unpack(msgpack::object o)
+Value::msgpack_unpack(const msgpack::object& o)
 {
-    if (o.type != msgpack::type::MAP) throw msgpack::type_error();
-    if (o.via.map.size < 2) throw msgpack::type_error();
+    if (o.type != msgpack::type::MAP or o.via.map.size < 2)
+        throw msgpack::type_error();
 
-    if (auto rid = findMapValue(o, "id")) {
+    if (auto rid = findMapValue(o, VALUE_KEY_ID)) {
         id = rid->as<Id>();
     } else
         throw msgpack::type_error();
 
-    if (auto rdat = findMapValue(o, "dat")) {
+    if (auto rdat = findMapValue(o, VALUE_KEY_DAT)) {
         msgpack_unpack_body(*rdat);
     } else
         throw msgpack::type_error();
+
+    if (auto rprio = findMapValue(o, VALUE_KEY_PRIO)) {
+        priority = rprio->as<unsigned>();
+    }
 }
 
 void
@@ -140,37 +130,37 @@ Value::msgpack_unpack_body(const msgpack::object& o)
     } else {
         if (o.type != msgpack::type::MAP)
             throw msgpack::type_error();
-        auto rbody = findMapValue(o, "body");
+        auto rbody = findMapValue(o, VALUE_KEY_BODY);
         if (not rbody)
             throw msgpack::type_error();
 
-        if (auto rdata = findMapValue(*rbody, "data")) {
+        if (auto rdata = findMapValue(*rbody, VALUE_KEY_DATA)) {
             data = unpackBlob(*rdata);
         } else
             throw msgpack::type_error();
 
-        if (auto rtype = findMapValue(*rbody, "type")) {
+        if (auto rtype = findMapValue(*rbody, VALUE_KEY_TYPE)) {
             type = rtype->as<ValueType::Id>();
         } else
             throw msgpack::type_error();
 
-        if (auto rutype = findMapValue(*rbody, "utype")) {
+        if (auto rutype = findMapValue(*rbody, VALUE_KEY_USERTYPE)) {
             user_type = rutype->as<std::string>();
         }
 
-        if (auto rowner = findMapValue(*rbody, "owner")) {
-            if (auto rseq = findMapValue(*rbody, "seq"))
+        if (auto rowner = findMapValue(*rbody, VALUE_KEY_OWNER)) {
+            if (auto rseq = findMapValue(*rbody, VALUE_KEY_SEQ))
                 seq = rseq->as<decltype(seq)>();
             else
                 throw msgpack::type_error();
             crypto::PublicKey new_owner;
             new_owner.msgpack_unpack(*rowner);
             owner = std::make_shared<const crypto::PublicKey>(std::move(new_owner));
-            if (auto rrecipient = findMapValue(*rbody, "to")) {
+            if (auto rrecipient = findMapValue(*rbody, VALUE_KEY_TO)) {
                 recipient = rrecipient->as<InfoHash>();
             }
 
-            if (auto rsig = findMapValue(o, "sig")) {
+            if (auto rsig = findMapValue(o, VALUE_KEY_SIGNATURE)) {
                 signature = unpackBlob(*rsig);
             } else
                 throw msgpack::type_error();
@@ -179,63 +169,65 @@ Value::msgpack_unpack_body(const msgpack::object& o)
 }
 
 #ifdef OPENDHT_JSONCPP
-Value::Value(Json::Value& json)
+Value::Value(const Json::Value& json)
 {
-    id = Value::Id(unpackId(json, "id"));
-    if (json.isMember("cypher")) {
-        auto cypherStr = json["cypher"].asString();
-        cypherStr = base64_decode(cypherStr);
-        cypher = std::vector<unsigned char>(cypherStr.begin(), cypherStr.end());
-    }
-    if (json.isMember("sig")) {
-        auto sigStr = json["sig"].asString();
-        sigStr = base64_decode(sigStr);
-        signature = std::vector<unsigned char>(sigStr.begin(), sigStr.end());
-    }
-    if (json.isMember("seq"))
-        seq = json["seq"].asInt();
-    if (json.isMember("owner")) {
-        auto ownerStr = json["owner"].asString();
+    id = Value::Id(unpackId(json, VALUE_KEY_ID));
+    const auto& jcypher = json["cypher"];
+    if (jcypher.isString())
+        cypher = base64_decode(jcypher.asString());
+    const auto& jsig = json[VALUE_KEY_SIGNATURE];
+    if (jsig.isString())
+        signature = base64_decode(jsig.asString());
+    const auto& jseq = json[VALUE_KEY_SEQ];
+    if (!jseq.isNull())
+        seq = jseq.asInt();
+    const auto& jowner = json[VALUE_KEY_OWNER];
+    if (jowner.isString()) {
+        auto ownerStr = jowner.asString();
         auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
         owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
     }
-    if (json.isMember("to")) {
-        auto toStr = json["to"].asString();
-        recipient = InfoHash(toStr);
-    }
-    if (json.isMember("type"))
-        type = json["type"].asInt();
-    if (json.isMember("data")){
-        auto dataStr = json["data"].asString();
-        dataStr = base64_decode(dataStr);
-        data = std::vector<unsigned char>(dataStr.begin(), dataStr.end());
-    }
-    if (json.isMember("utype"))
-        user_type = json["utype"].asString();
+    const auto& jto = json[VALUE_KEY_TO];
+    if (jto.isString())
+        recipient = InfoHash(jto.asString());
+    const auto& jtype = json[VALUE_KEY_TYPE];
+    if (!jtype.isNull())
+        type = jtype.asInt();
+    const auto& jdata = json[VALUE_KEY_DATA];
+    if (jdata.isString())
+        data = base64_decode(jdata.asString());
+    const auto& jutype = json[VALUE_KEY_USERTYPE];
+    if (jutype.isString())
+        user_type = jutype.asString();
+    const auto& jprio = json["prio"];
+    if (jprio.isIntegral())
+        priority = jprio.asUInt();
 }
 
 Json::Value
 Value::toJson() const
 {
     Json::Value val;
-    val["id"] = std::to_string(id);
+    val[VALUE_KEY_ID] = std::to_string(id);
     if (isEncrypted()) {
         val["cypher"] = base64_encode(cypher);
     } else {
         if (isSigned())
-            val["sig"] = base64_encode(signature);
+            val[VALUE_KEY_SIGNATURE] = base64_encode(signature);
         bool has_owner = owner && *owner;
         if (has_owner) { // isSigned
-            val["seq"] = seq;
-            val["owner"] = owner->toString();
+            val[VALUE_KEY_SEQ] = seq;
+            val[VALUE_KEY_OWNER] = owner->toString();
             if (recipient)
-                val["to"] = recipient.toString();
+                val[VALUE_KEY_TO] = recipient.toString();
         }
-        val["type"] = type;
-        val["data"] = base64_encode(data);
+        val[VALUE_KEY_TYPE] = type;
+        val[VALUE_KEY_DATA] = base64_encode(data);
         if (not user_type.empty())
-            val["utype"] = user_type;
+            val[VALUE_KEY_USERTYPE] = user_type;
     }
+    if (priority)
+        val["prio"] = priority;
     return val;
 }
 
@@ -243,13 +235,11 @@ uint64_t
 unpackId(const Json::Value& json, const std::string& key) {
     uint64_t ret = 0;
     try {
-        if (json.isMember(key)) {
-            const auto& t = json[key];
-            if (t.isString()) {
-                ret = std::stoull(t.asString());
-            } else {
-                ret = t.asLargestUInt();
-            }
+        const auto& t = json[key];
+        if (t.isString()) {
+            ret = std::stoull(t.asString());
+        } else {
+            ret = t.asLargestUInt();
         }
     } catch (...) {}
     return ret;
@@ -419,13 +409,13 @@ Select::Select(const std::string& q_str) {
 
         while (std::getline(fields, token, ',')) {
             trim_str(token);
-            if (token == "id")
+            if (token == VALUE_KEY_ID)
                 field(Value::Field::Id);
             else if (token == "value_type")
                 field(Value::Field::ValueType);
             else if (token == "owner_pk")
                 field(Value::Field::OwnerPk);
-            if (token == "seq")
+            if (token == VALUE_KEY_SEQ)
                 field(Value::Field::SeqNum);
             else if (token == "user_type")
                 field(Value::Field::UserType);
@@ -461,13 +451,13 @@ Where::Where(const std::string& q_str) {
                     s = value_str.substr(1, value_str.size()-2);
                 else
                     s = value_str;
-                if (field_str == "id")
+                if (field_str == VALUE_KEY_ID)
                     id(v);
                 else if (field_str == "value_type")
                     valueType(v);
                 else if (field_str == "owner_pk")
                     owner(InfoHash(s));
-                else if (field_str == "seq")
+                else if (field_str == VALUE_KEY_SEQ)
                     seq(v);
                 else if (field_str == "user_type")
                     userType(s);
@@ -527,7 +517,7 @@ std::ostream& operator<<(std::ostream& s, const dht::Select& select) {
     for (auto fs = select.fieldSelection_.begin() ; fs != select.fieldSelection_.end() ; ++fs) {
         switch (*fs) {
             case Value::Field::Id:
-                s << "id";
+                s << VALUE_KEY_ID;
                 break;
             case Value::Field::ValueType:
                 s << "value_type";
@@ -539,7 +529,7 @@ std::ostream& operator<<(std::ostream& s, const dht::Select& select) {
                 s << "owner_public_key";
                 break;
             case Value::Field::SeqNum:
-                s << "seq";
+                s << VALUE_KEY_SEQ;
                 break;
             default:
                 break;