From a1333fcaa519fde021dc8d28b8b4b871dd2a3155 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20D=C3=A9saulniers?= Date: Tue, 23 Feb 2016 15:49:44 -0500 Subject: [PATCH] handling notify listeners when storage changed --- include/opendht/network_engine.h | 16 ++++++++++++++++ src/dht.cpp | 12 +++++++----- src/network_engine.cpp | 14 ++++++++++++++ 3 files changed, 37 insertions(+), 5 deletions(-) diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 99d01efc..2ada7363 100644 --- a/include/opendht/network_engine.h +++ b/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> nodes, std::vector> nodes6, + std::vector> values); /** * Cancel an ongoing request before it's processed. diff --git a/src/dht.cpp b/src/dht.cpp index 7d1f4332..36c82160 100644 --- a/src/dht.cpp +++ b/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 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> 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); } } diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 41b21603..66327c91 100644 --- a/src/network_engine.cpp +++ b/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> nodes, std::vector> nodes6, + std::vector> 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()