update to version 2.1.0

Actually, packets are not retransmitted for listen operations. This
leads to some problems when they are already several values present
at one hash. Ideally, packets should be transmitted inside a request
to avoid packet loss and get all datas.

This patch breaks the compatibility with previous versions

CMakeLists.txt

@@ -4,7 +4,7 @@ include(CMakeDependentOption)
 
 project (opendht)
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 0.0)
+set (opendht_VERSION_MINOR 1.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

configure.ac

@@ -1,6 +1,6 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
-m4_define([opendht_minor_version], 0)
+m4_define([opendht_minor_version], 1)
 m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

include/opendht/callbacks.h

@@ -82,6 +82,8 @@ struct OPENDHT_PUBLIC NodeInfo {
     NodeStats ipv4 {};
     NodeStats ipv6 {};
     size_t ongoing_ops {0};
+    in_port_t bound4 {0};
+    in_port_t bound6 {0};
 
 #ifdef OPENDHT_JSONCPP
     /**

include/opendht/dht.h

@@ -445,7 +445,7 @@ private:
     void reportedAddr(const SockAddr&);
 
     // Storage
-    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
+    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
     bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
     bool storageErase(const InfoHash& id, Value::Id vid);
     bool storageRefresh(const InfoHash& id, Value::Id vid);
@@ -600,7 +600,8 @@ private:
             const InfoHash& hash,
             const Blob& token,
             size_t socket_id,
-            const Query& query);
+            const Query& query,
+            int version = 0);
     void onListenDone(const Sp<Node>& status,
             net::RequestAnswer& a,
             Sp<Search>& sr);

include/opendht/network_engine.h

@@ -180,7 +180,8 @@ private:
             const InfoHash&,
             const Blob&,
             Tid,
-            const Query&)> onListen {};
+            const Query&,
+            int)> onListen {};
     /**
      * Called on announce request.
      *
@@ -237,9 +238,10 @@ public:
     void clear();
 
     /**
-     * Sends values (with closest nodes) to a listenner.
+     * Sends values (with closest nodes) to a listener.
      *
-     * @param sa          The address of the listenner.
+     * @deprecated
+     * @param sa          The address of the listener.
      * @param sslen       The length of the sockaddr structure.
      * @param socket_id  The tid to use to write to the request socket.
      * @param hash        The hash key of the value.
@@ -248,10 +250,11 @@ public:
      * @param nodes       The ipv4 closest nodes.
      * @param nodes6      The ipv6 closest nodes.
      * @param values      The values to send.
+     * @param version     If version = 1, a request will be used to answer to the listener
      */
     void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
             std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
-            std::vector<Sp<Value>>&& values, const Query& q);
+            std::vector<Sp<Value>>&& values, const Query& q, int version);
 
     void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
     void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
@@ -401,6 +404,24 @@ public:
                                  const Blob& token,
                                  RequestCb&& on_done,
                                  RequestExpiredCb&& on_expired);
+    /**
+     * Send a "update" request to a given node. Used for Listen operations
+     *
+     * @param n           The node.
+     * @param hash        The target hash.
+     * @param values      The values.
+     * @param created     Time id.
+     * @param token       A security token.
+     * @param sid         The socket id.
+     *
+     * @return the request with information concerning its success.
+     */
+    Sp<Request> sendUpdateValues(Sp<Node> n,
+                                 const InfoHash& infohash,
+                                 const std::vector<Sp<Value>>& values,
+                                 time_point created,
+                                 const Blob& token,
+                                 const size_t& sid);
 
     /**
      * Parses a message and calls appropriate callbacks.
@@ -490,19 +511,20 @@ private:
     void sendRequest(const Sp<Request>& request);
 
     struct MessageStats {
-        unsigned ping    {0};
-        unsigned find    {0};
-        unsigned get     {0};
-        unsigned put     {0};
-        unsigned listen  {0};
-        unsigned refresh {0};
+        unsigned ping         {0};
+        unsigned find         {0};
+        unsigned get          {0};
+        unsigned put          {0};
+        unsigned listen       {0};
+        unsigned updateValue  {0};
+        unsigned refresh      {0};
     };
 
 
     // basic wrapper for socket sendto function
     int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);
 
-    void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
+    void sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr);
     std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
     void maintainRxBuffer(Tid tid);
 

src/dht.cpp

@@ -1201,7 +1201,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
                 vals.push_back(v.data);
                 Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
                 network_engine.tellListener(node_listeners.first, l.first, id, 0, ntoken, {}, {},
-                        std::move(vals), l.second.query);
+                        std::move(vals), l.second.query, l.second.version);
             }
         }
     }
@@ -1259,7 +1259,7 @@ Dht::storageErase(const InfoHash& id, Value::Id vid)
 }
 
 void
-Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
+Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
 {
     const auto& now = scheduler.time();
     auto st = store.find(id);
@@ -1275,9 +1275,9 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
         if (not vals.empty()) {
             network_engine.tellListener(node, socket_id, id, WANT4 | WANT6, makeToken(node->getAddr(), false),
                     dht4.buckets.findClosestNodes(id, now, TARGET_NODES), dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
-                    std::move(vals), query);
+                    std::move(vals), query, version);
         }
-        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query)});
+        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
     }
     else
         l->second.refresh(now, std::forward<Query>(query));
@@ -1755,7 +1755,7 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
             std::bind(&Dht::onPing, this, _1),
             std::bind(&Dht::onFindNode, this, _1, _2, _3),
             std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
-            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
+            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5, _6),
             std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
             std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)),
     persistPath(config.persist_path),
@@ -1982,14 +1982,19 @@ Dht::onDisconnected()
         return;
     if (logger_)
         logger_->d(myid, "Bootstraping");
-    for (const auto& boootstrap : bootstrap_nodes)
+    for (const auto& boootstrap : bootstrap_nodes) {
         try {
-            for (const auto& ip : network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second))
+            auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
+            for (auto& ip : ips) {
+                if (ip.getPort() == 0)
+                    ip.setPort(net::DHT_DEFAULT_PORT);
                 pingNode(ip);
+            }
         } catch (const std::exception& e) {
             if (logger_)
                 logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
         }
+    }
     if (bootstrapJob)
         bootstrapJob->cancel();
     bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
@@ -2320,7 +2325,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
 }
 
 net::RequestAnswer
-Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query)
+Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query, int version)
 {
     if (not hash) {
         if (logger_)
@@ -2336,7 +2341,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
         throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
     }
     Query q = query;
-    storageAddListener(hash, node, socket_id, std::move(q));
+    storageAddListener(hash, node, socket_id, std::move(q), version);
     return {};
 }
 

src/dhtrunner.cpp

@@ -475,6 +475,10 @@ DhtRunner::getNodeInfo() const {
         info.node_id = dht->getNodeId();
         info.ipv4 = dht->getNodesStats(AF_INET);
         info.ipv6 = dht->getNodesStats(AF_INET6);
+        if (auto sock = dht->getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
     }
     info.ongoing_ops = ongoing_ops;
     return info;
@@ -492,6 +496,10 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
         info.node_id = dht.getNodeId();
         info.ipv4 = dht.getNodesStats(AF_INET);
         info.ipv6 = dht.getNodesStats(AF_INET6);
+        if (auto sock = dht.getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
         info.ongoing_ops = ongoing_ops;
         cb(std::move(sinfo));
         opEnded();

src/listener.h

@@ -30,8 +30,9 @@ namespace dht {
 struct Listener {
     time_point time;
     Query query;
+    int version;
 
-    Listener(time_point t, Query&& q) : time(t), query(std::move(q)) {}
+    Listener(time_point t, Query&& q, int version = 0) : time(t), query(std::move(q)), version(version) {}
 
     void refresh(time_point t, Query&& q) {
         time = t;

src/net.h

@@ -31,7 +31,8 @@ enum class MessageType {
     Refresh,
     Listen,
     ValueData,
-    ValueUpdate
+    ValueUpdate,
+    UpdateValue
 };
 
 } /* namespace net */

src/network_engine.cpp

@@ -48,23 +48,6 @@ static constexpr uint8_t v4prefix[16] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xF
 constexpr unsigned SEND_NODES {8};
 
 
-/* Transaction-ids are 4-bytes long, with the first two bytes identifying
- * the kind of request, and the remaining two a sequence number in
- * host order.
- */
-struct TransId final : public std::array<uint8_t, 4> {
-    TransId() { std::fill(begin(), end(), 0); }
-    TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
-    TransId(uint32_t id) {
-        *reinterpret_cast<uint32_t*>(data()) = htonl(id);
-    }
-
-    uint32_t toInt() const {
-        return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
-    }
-};
-
-
 struct NetworkEngine::PartialMessage {
     SockAddr from;
     time_point start;
@@ -140,11 +123,15 @@ void
 NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash, want_t want,
         const Blob& ntoken, std::vector<Sp<Node>>&& nodes,
         std::vector<Sp<Node>>&& nodes6, std::vector<Sp<Value>>&& values,
-        const Query& query)
+        const Query& query, int version)
 {
     auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
     try {
-        sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        if (version == 1) {
+            sendUpdateValues(node, hash, values, scheduler.time(), ntoken, socket_id);
+        } else {
+            sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        }
     } catch (const std::overflow_error& e) {
         if (logger_)
             logger_->e("Can't send value: buffer not large enough !");
@@ -617,18 +604,28 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 if (logIncoming_ and logger_)
                     logger_->d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                 ++in_stats.listen;
-                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
+                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query), msg->version);
                 auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                 sendListenConfirmation(from, msg->tid);
                 break;
             }
+            case MessageType::UpdateValue: {
+                if (logIncoming_ and logger_)
+                    logger_->d(msg->info_hash, node->id, "[node %s] got 'update' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                ++in_stats.updateValue;
+                if (auto rsocket = node->getSocket(msg->socket_id))
+                    rsocket->on_receive(node, std::move(*msg));
+                else if (logger_)
+                    logger_->e(msg->info_hash, node->id, "[node %s] 'update' request without socket for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                break;
+            }
             default:
                 break;
             }
         } catch (const std::overflow_error& e) {
             if (logger_)
                 logger_->e("Can't send value: buffer not large enough !");
-        } catch (DhtProtocolException& e) {
+        } catch (const DhtProtocolException& e) {
             sendError(from, msg->tid, e.getCode(), e.getMsg().c_str(), true);
         }
     }
@@ -654,7 +651,7 @@ NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool conf
 
 Sp<Request>
 NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (node->getNewTid());
+    Tid tid (node->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -663,15 +660,14 @@ NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& o
      pk.pack(KEY_REQ_ID); pk.pack(myid);
 
     pk.pack(KEY_Q); pk.pack(QUERY_PING);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                              pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Ping, tid.toInt(), node,
+    auto req = std::make_shared<Request>(MessageType::Ping, tid, node,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&&) {
             if (logger_)
@@ -701,9 +697,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
       pk.pack(KEY_REQ_ID); pk.pack(myid);
       insertAddr(pk, addr);
 
-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -716,7 +710,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
 Sp<Request>
 NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -732,15 +726,14 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
     }
 
     pk.pack(KEY_Q); pk.pack(QUERY_FIND);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::FindNode, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::FindNode, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done) {
@@ -762,7 +755,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
 Sp<Request>
 NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -786,15 +779,14 @@ NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query&
       }
 
     pk.pack(KEY_Q); pk.pack(QUERY_GET);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::GetValues, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::GetValues, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done) {
@@ -896,7 +888,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector<Sp<Va
 }
 
 void
-NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr)
+NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr)
 {
     msgpack::sbuffer buffer;
     unsigned i=0;
@@ -911,8 +903,7 @@ NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals
                 pk.pack(KEY_NETID); pk.pack(config.network);
             }
             pk.pack(KEY_Y); pk.pack(KEY_V);
-            pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                                       pk.pack_bin_body((const char*)tid.data(), tid.size());
+            pk.pack(KEY_TID); pk.pack(tid);
             pk.pack(KEY_V); pk.pack_map(1);
                 pk.pack(i); pk.pack_map(2);
                     pk.pack(std::string("o")); pk.pack(start);
@@ -967,9 +958,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
         }
     }
 
-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                      pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1045,7 +1034,7 @@ NetworkEngine::sendListen(Sp<Node> n,
         SocketCb&& socket_cb)
 {
     Tid socket;
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     if (previous and previous->node == n) {
         socket = previous->getSocket();
     } else {
@@ -1060,33 +1049,30 @@ NetworkEngine::sendListen(Sp<Node> n,
             logger_->e(hash, "[node %s] unable to get a valid socket for listen. Aborting listen", n->toString().c_str());
         return {};
     }
-    TransId sid(socket);
-
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
 
     auto has_query = not query.where.empty() or not query.select.empty();
-    pk.pack(KEY_A); pk.pack_map(4 + has_query);
+    pk.pack(KEY_A); pk.pack_map(5 + has_query);
       pk.pack(KEY_REQ_ID);    pk.pack(myid);
+      pk.pack(KEY_VERSION);   pk.pack(1);
       pk.pack(KEY_REQ_H);     pk.pack(hash);
       pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-      pk.pack(KEY_REQ_SID);   pk.pack_bin(sid.size());
-                              pk.pack_bin_body((const char*)sid.data(), sid.size());
+      pk.pack(KEY_REQ_SID);   pk.pack(socket);
       if (has_query) {
           pk.pack(KEY_REQ_QUERY); pk.pack(query);
       }
 
     pk.pack(KEY_Q); pk.pack(QUERY_LISTEN);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Listen, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Listen, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done)
@@ -1113,9 +1099,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
       pk.pack(KEY_REQ_ID); pk.pack(myid);
       insertAddr(pk, addr);
 
-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1134,7 +1118,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
         RequestCb&& on_done,
         RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1150,15 +1134,14 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
       pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
 
     pk.pack(KEY_Q);   pk.pack(QUERY_PUT);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y);   pk.pack(KEY_Q);
     pk.pack(KEY_UA);  pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (msg.value_id == Value::INVALID_ID) {
@@ -1184,6 +1167,52 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
     return req;
 }
 
+Sp<Request>
+NetworkEngine::sendUpdateValues(Sp<Node> n,
+                                const InfoHash& infohash,
+                                const std::vector<Sp<Value>>& values,
+                                time_point created,
+                                const Blob& token,
+                                const size_t& socket_id)
+{
+    Tid tid (n->getNewTid());
+    Tid sid (socket_id);
+
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+    pk.pack_map(5+(config.network?1:0));
+
+    pk.pack(KEY_A); pk.pack_map((created < scheduler.time() ? 7 : 6));
+      pk.pack(KEY_REQ_ID);     pk.pack(myid);
+      pk.pack(KEY_VERSION);    pk.pack(1);
+      pk.pack(KEY_REQ_H);      pk.pack(infohash);
+      pk.pack(KEY_REQ_SID);   pk.pack(sid);
+      auto v = packValueHeader(buffer, values);
+      if (created < scheduler.time()) {
+          pk.pack(KEY_REQ_CREATION);
+          pk.pack(to_time_t(created));
+      }
+      pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
+
+    pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
+    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_Y);   pk.pack(KEY_Q);
+    pk.pack(KEY_UA);  pk.pack(my_v);
+    if (config.network) {
+        pk.pack(KEY_NETID); pk.pack(config.network);
+    }
+
+    auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+        Blob(buffer.data(), buffer.data() + buffer.size()),
+        [=](const Request&, ParsedMessage&&) { /* on done */ },
+        [=](const Request&, bool) { /* on expired */ }
+    );
+    req->parts = std::move(v);
+    sendRequest(req);
+    ++out_stats.updateValue;
+    return req;
+}
+
 Sp<Request>
 NetworkEngine::sendRefreshValue(Sp<Node> n,
                 const InfoHash& infohash,
@@ -1192,7 +1221,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                 RequestCb&& on_done,
                 RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1204,15 +1233,14 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
       pk.pack(KEY_REQ_TOKEN);    pk.pack(token);
 
     pk.pack(KEY_Q); pk.pack(QUERY_REFRESH);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Refresh, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Refresh, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (msg.value_id == Value::INVALID_ID) {
@@ -1248,9 +1276,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
       pk.pack(KEY_REQ_VALUE_ID); pk.pack(vid);
       insertAddr(pk, addr);
 
-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1280,9 +1306,7 @@ NetworkEngine::sendError(const SockAddr& addr,
           pk.pack(KEY_REQ_ID); pk.pack(myid);
     }
 
-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_E);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {

src/parsed_message.h

@@ -54,10 +54,12 @@ static const std::string KEY_REQ_EXPIRED {"exp"};
 static const std::string KEY_REQ_REFRESHED {"re"};
 static const std::string KEY_REQ_FIELDS {"fileds"};
 static const std::string KEY_REQ_WANT {"w"};
+static const std::string KEY_VERSION {"ve"};
 
 static const std::string QUERY_PING {"ping"};
 static const std::string QUERY_FIND {"find"};
 static const std::string QUERY_GET {"get"};
+static const std::string QUERY_UPDATE {"update"};
 static const std::string QUERY_PUT {"put"};
 static const std::string QUERY_LISTEN {"listen"};
 static const std::string QUERY_REFRESH {"refresh"};
@@ -113,6 +115,7 @@ struct ParsedMessage {
     uint16_t error_code;
     /* reported address by the distant node */
     std::string ua;
+    int version {0};
     SockAddr addr;
     void msgpack_unpack(const msgpack::object& o);
 
@@ -224,6 +227,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
         type = MessageType::AnnounceValue;
     else if (parsed.q == QUERY_REFRESH)
         type = MessageType::Refresh;
+    else if (parsed.q == QUERY_UPDATE)
+        type = MessageType::UpdateValue;
     else
         throw msgpack::type_error();
 
@@ -295,6 +300,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
             parsedReq.fields = &o.val;
         else if (key == KEY_REQ_WANT)
             parsedReq.want = &o.val;
+        else if (key == KEY_VERSION)
+            version = o.val.as<int>();
     }
 
     if (parsedReq.sa) {

tests/dhtrunnertester.cpp

@@ -167,4 +167,51 @@ DhtRunnerTester::testListen() {
     node1.cancelListen(d, tokend);
 }
 
+void
+DhtRunnerTester::testListenLotOfBytes() {
+    std::mutex mutex;
+    std::condition_variable cv;
+    std::atomic_uint valueCount(0);
+    unsigned putCount(0);
+    unsigned putOkCount(0);
+
+    std::string data(10000, 'a');
+
+    auto foo = dht::InfoHash::get("foo");
+    constexpr unsigned N = 50;
+
+    for (unsigned i=0; i<N; i++) {
+        node2.put(foo, data, [&](bool ok) {
+            std::lock_guard<std::mutex> lock(mutex);
+            putCount++;
+            if (ok) putOkCount++;
+            cv.notify_all();
+        });
+    }
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N; }));
+    }
+
+    dht::DhtRunner node3 {};
+    dht::DhtRunner::Config config;
+    config.dht_config.node_config.max_peer_req_per_sec = -1;
+    config.dht_config.node_config.max_req_per_sec = -1;
+    node3.run(42242, config);
+    node3.bootstrap(node1.getBound());
+
+    auto ftokenfoo = node3.listen(foo, [&](const std::shared_ptr<dht::Value>&) {
+        valueCount++;
+        cv.notify_all();
+        return true;
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return valueCount == N; }));
+    }
+
+    node3.cancelListen(foo, ftokenfoo.get());
+}
+
 }  // namespace test

tests/dhtrunnertester.h

@@ -32,6 +32,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testConstructors);
     CPPUNIT_TEST(testGetPut);
     CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -57,6 +58,10 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test listen method
      */
     void testListen();
+    /**
+     * Test listen method with lot of datas
+     */
+    void testListenLotOfBytes();
 };
 
 }  // namespace test

tools/dhtchat.cpp

@@ -64,10 +64,10 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         dht.run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);
 
-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
         std::cout << "  type 'c {hash}' to join a channel" << std::endl << std::endl;
 
         bool connected {false};

tools/dhtnode.cpp

@@ -114,7 +114,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif
 )
 {
-    print_node_info(*node, params);
+    print_node_info(node->getNodeInfo());
     std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;
 
 #ifndef WIN32_NATIVE
@@ -143,22 +143,23 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
             print_help();
             continue;
         } else if (op == "ll") {
-            auto nodeInfo = node->getNodeInfo();
-            print_node_info(*node, params);
-            std::cout << nodeInfo.ongoing_ops << " ongoing operations" << std::endl;
-            std::cout << "IPv4 stats:" << std::endl;
-            std::cout << nodeInfo.ipv4.toString() << std::endl;
-            std::cout << "IPv6 stats:" << std::endl;
-            std::cout << nodeInfo.ipv6.toString() << std::endl;
+            node->getNodeInfo([&](const std::shared_ptr<dht::NodeInfo>& nodeInfo) {
+                print_node_info(*nodeInfo);
+                std::cout << nodeInfo->ongoing_ops << " ongoing operations" << std::endl;
+                std::cout << "IPv4 stats:" << std::endl;
+                std::cout << nodeInfo->ipv4.toString() << std::endl;
+                std::cout << "IPv6 stats:" << std::endl;
+                std::cout << nodeInfo->ipv6.toString() << std::endl;
 #ifdef OPENDHT_PROXY_SERVER
-            for (const auto& proxy : proxies) {
-                std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
-                if (auto stats = proxy.second->stats())
-                    std::cout << "  " << stats->toString() << std::endl;
-                else
-                    std::cout << "  (stats not available yet)" << std::endl;                
-            }
+                for (const auto& proxy : proxies) {
+                    std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
+                    if (auto stats = proxy.second->stats())
+                        std::cout << "  " << stats->toString() << std::endl;
+                    else
+                        std::cout << "  (stats not available yet)" << std::endl;                
+                }
 #endif
+            });
             continue;
         } else if (op == "lr") {
             std::cout << "IPv4 routing table:" << std::endl;
@@ -190,14 +191,10 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
             continue;
         } else if (op == "b") {
             iss >> idstr;
-            try {
-                auto addr = splitPort(idstr);
-                if (not addr.first.empty() and addr.second.empty())
-                    addr.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
-                node->bootstrap(addr.first.c_str(), addr.second.c_str());
-            } catch (const std::exception& e) {
-                std::cerr << e.what() << std::endl;
-            }
+            if (not idstr.empty())
+                node->bootstrap(idstr);
+            else
+                std::cerr << "No service provided. Syntax: b hostname:port" << std::endl;
             continue;
         } else if (op == "log") {
             iss >> idstr;
@@ -537,9 +534,9 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         node->run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.first.empty()) {
-            std::cout << "Bootstrap: " << params.bootstrap.first << ":" << params.bootstrap.second << std::endl;
-            node->bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty()) {
+            std::cout << "Bootstrap: " << params.bootstrap << std::endl;
+            node->bootstrap(params.bootstrap);
         }
 
 #ifdef OPENDHT_PROXY_SERVER

tools/dhtscanner.cpp

@@ -91,10 +91,10 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         dht.run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);
 
-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
         std::cout << "Scanning network..." << std::endl;
         auto all_nodes = std::make_shared<NodeSet>();
 

tools/tools_common.h

@@ -124,7 +124,7 @@ struct dht_params {
     bool log {false};
     bool syslog {false};
     std::string logfile {};
-    std::pair<std::string, std::string> bootstrap {};
+    std::string bootstrap {};
     dht::NetId network {0};
     in_port_t port {0};
     in_port_t proxyserver {0};
@@ -191,16 +191,14 @@ getDhtConfig(dht_params& params)
     return {std::move(config), std::move(context)};
 }
 
-void print_node_info(const dht::DhtRunner& node, const dht_params& params) {
-    std::cout << "OpenDHT node " << node.getNodeId() << " running on ";
-    auto port4 = node.getBoundPort(AF_INET);
-    auto port6 = node.getBoundPort(AF_INET6);
-    if (port4 == port6)
-        std::cout << "port " << port4 << std::endl;
+void print_node_info(const dht::NodeInfo& info) {
+    std::cout << "OpenDHT node " << info.node_id << " running on ";
+    if (info.bound4 == info.bound6)
+        std::cout << "port " << info.bound4 << std::endl;
     else
-        std::cout << "IPv4 port " << port4 << ", IPv6 port " << port6 << std::endl;
-    if (params.id.first)
-        std::cout << "Public key ID " << node.getId() << std::endl;
+        std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
+    if (info.id)
+        std::cout << "Public key ID " << info.id << std::endl;
 }
 
 static const constexpr struct option long_options[] = {
@@ -291,10 +289,7 @@ parseArgs(int argc, char **argv) {
             params.public_stable = true;
             break;
         case 'b':
-            params.bootstrap = dht::splitPort((optarg[0] == '=') ? optarg+1 : optarg);
-            if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
-                params.bootstrap.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
-            }
+            params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
             break;
         case 'V':
             params.version = true;