handling notify listeners when storage changed

include/opendht/network_engine.h

@@ -213,6 +213,22 @@ public:
     }
     virtual ~NetworkEngine() {};
 
+    /**
+     * Sends values (with closest nodes) to a listenner.
+     *
+     * @param sa  The address of the listenner.
+     * @param sslen  The length of the sockaddr structure.
+     * @param rid  The request id of the initial listen request.
+     * @param hash  The hash key of the value.
+     * @param want  Wether to send ipv4 and/or ipv6 nodes.
+     * @param ntoken  Listen security token.
+     * @param nodes  The ipv4 closest nodes.
+     * @param nodes6  The ipv6 closest nodes.
+     * @param values  The values to send.
+     */
+    void tellListener(const sockaddr* sa, socklen_t salen, size_t rid, InfoHash hash, want_t want, Blob ntoken,
+            std::vector<std::shared_ptr<Node>> nodes, std::vector<std::shared_ptr<Node>> nodes6,
+            std::vector<std::shared_ptr<Value>> values);
 
     /**
      * Cancel an ongoing request before it's processed.

src/dht.cpp

@@ -1660,12 +1660,14 @@ Dht::storageChanged(Storage& st, ValueStorage& v)
     }
 
     for (const auto& l : st.listeners) {
-        DHT_LOG.WARN("Storage changed. Sending update to %s %s.", l.id.toString().c_str(), print_addr((sockaddr*)&l.ss, l.sslen).c_str());
-        std::vector<ValueStorage> vals;
-        vals.push_back(v);
+        DHT_LOG.WARN("Storage changed. Sending update to %s %s.",
+                l.id.toString().c_str(), print_addr((sockaddr*)&l.ss, l.sslen).c_str());
+        std::vector<std::shared_ptr<Value>> vals;
+        vals.push_back(v.data);
         Blob ntoken = makeToken((const sockaddr*)&l.ss, false);
-        //TODO
-        //sendClosestNodes((const sockaddr*)&l.ss, l.sslen, TransId {TransPrefix::GET_VALUES, l.tid}, st.id, WANT4 | WANT6, ntoken, vals);
+        network_engine->tellListener((const sockaddr*)&l.ss, l.sslen, l.rid, st.id, WANT4 | WANT6, ntoken,
+                buckets.findClosestNodes(st.id, now, TARGET_NODES), buckets6.findClosestNodes(st.id, now, TARGET_NODES),
+                vals);
     }
 }
 

src/network_engine.cpp

@@ -48,6 +48,20 @@ const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::ANNOUNCE_VALUES  =
 const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::LISTEN  = {"lt"};
 constexpr long unsigned NetworkEngine::MAX_REQUESTS_PER_SEC;
 
+void
+NetworkEngine::tellListener(const sockaddr *sa, socklen_t salen, size_t rid, InfoHash hash, want_t want,
+        Blob ntoken, std::vector<std::shared_ptr<Node>> nodes, std::vector<std::shared_ptr<Node>> nodes6,
+        std::vector<std::shared_ptr<Value>> values)
+{
+    auto nnodes = bufferNodes(sa->sa_family, hash, want, nodes, nodes6);
+    try {
+        sendNodesValues(sa, salen, TransId {TransPrefix::GET_VALUES, (uint16_t)rid}, nnodes.first, nnodes.second,
+                values, ntoken);
+    } catch (const std::overflow_error& e) {
+        DHT_LOG.ERROR("Can't send value: buffer not large enough !");
+    }
+}
+
 /* Rate control for requests we receive. */
 bool
 NetworkEngine::rateLimit()