diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 49b0a039..b42926ef 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -362,6 +362,37 @@ public: RequestCb&& on_done, RequestExpiredCb&& on_expired, SocketCb&& socket_cb); + /** + * Send a "updateValues" request to a given node. + * + * @param n The node. + * @param hash The storage's hash. + * @param query The query describing filters. + * @param token A security token. + * @param previous The previous request "updateValues" sent to this node. + * @param socket **UNUSED** The socket for further response. + * + * For backward compatibility purpose, sendUpdateValues has to + * handle creation of the socket. Therefor, you cannot + * use openSocket yourself. TODO: Once we don't support + * the old "updateValues" negociation, sendUpdateValues shall not + * create the socket itself. + * + * @param on_done Request callback when the request is completed. + * @param on_expired Request callback when the request expires. + * @param socket_cb Callback to execute each time new updates arrive on + * the socket. + * + * @return the request with information concerning its success. + */ + Sp sendUpdateValues(Sp n, + const InfoHash& hash, + const Query& query, + const Blob& token, + Sp previous, + RequestCb&& on_done, + RequestExpiredCb&& on_expired, + SocketCb&& socket_cb); /** * Send a "announce" request to a given node. * @@ -490,12 +521,13 @@ private: void sendRequest(const Sp& 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 updateValues {0}; + unsigned refresh {0}; }; diff --git a/src/dht.cpp b/src/dht.cpp index ebc8b104..3219b3f0 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -605,7 +605,7 @@ Dht::searchSynchedNodeListen(const Sp& sr, SearchNode& n) }); } auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr; - auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req, + auto new_req = network_engine.sendUpdateValues(n.node, sr->id, *query, n.token, prev_req, [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable { /* on done */ if (auto sr = ws.lock()) { diff --git a/src/net.h b/src/net.h index 15b05b83..17f4805c 100644 --- a/src/net.h +++ b/src/net.h @@ -31,7 +31,8 @@ enum class MessageType { Refresh, Listen, ValueData, - ValueUpdate + ValueUpdate, + UpdateValues }; } /* namespace net */ diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 62de69e3..b0234f18 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -605,6 +605,10 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro sendListenConfirmation(from, msg->tid); break; } + case MessageType::UpdateValues: { + printf("@@@ Receive UpdateValues\n"); + break; + } default: break; } @@ -1075,6 +1079,75 @@ NetworkEngine::sendListen(Sp n, return req; } +Sp +NetworkEngine::sendUpdateValues(Sp n, + const InfoHash& hash, + const Query& query, + const Blob& token, + Sp previous, + RequestCb&& on_done, + RequestExpiredCb&& on_expired, + SocketCb&& socket_cb) +{ + Tid socket; + TransId tid (n->getNewTid()); + if (previous and previous->node == n) { + socket = previous->getSocket(); + } else { + if (previous) + if (logger_) + logger_->e(hash, "[node %s] trying refresh listen contract with wrong node", previous->node->toString().c_str()); + socket = n->openSocket(std::move(socket_cb)); + } + + if (not socket) { + if (logger_) + 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 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_REQ_ID); pk.pack(myid); + 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()); + if (has_query) { + pk.pack(KEY_REQ_QUERY); pk.pack(query); + } + + pk.pack(KEY_Q); pk.pack(QUERY_UPDATE_VALUES); + pk.pack(KEY_TID); pk.pack_bin(tid.size()); + pk.pack_bin_body((const char*)tid.data(), tid.size()); + 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(MessageType::UpdateValues, tid.toInt(), n, + Blob(buffer.data(), buffer.data() + buffer.size()), + [=](const Request& req_status, ParsedMessage&& msg) { /* on done */ + if (on_done) + on_done(req_status, {std::forward(msg)}); + }, + [=](const Request& req_status, bool done) { /* on expired */ + if (on_expired) + on_expired(req_status, done); + }, + socket + ); + sendRequest(req); + ++out_stats.updateValues; + return req; +} + void NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) { msgpack::sbuffer buffer; diff --git a/src/parsed_message.h b/src/parsed_message.h index edf5af2d..f5ed2cb7 100644 --- a/src/parsed_message.h +++ b/src/parsed_message.h @@ -58,6 +58,7 @@ static const std::string KEY_REQ_WANT {"w"}; 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_VALUES {"updateValues"}; static const std::string QUERY_PUT {"put"}; static const std::string QUERY_LISTEN {"listen"}; static const std::string QUERY_REFRESH {"refresh"}; @@ -224,6 +225,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_VALUES) + type = MessageType::UpdateValues; else throw msgpack::type_error(); diff --git a/tests/dhtrunnertester.cpp b/tests/dhtrunnertester.cpp index 06d3a321..4ab1c150 100644 --- a/tests/dhtrunnertester.cpp +++ b/tests/dhtrunnertester.cpp @@ -167,4 +167,54 @@ 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 {}; + for (int i = 0; i < 10000; ++i) { + data += "a"; + } + + auto foo = dht::InfoHash::get("foo"); + constexpr unsigned N = 50; + + for (unsigned i=0; i lock(mutex); + putCount++; + if (ok) putOkCount++; + cv.notify_all(); + }); + } + { + std::unique_lock 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&) { + valueCount++; + cv.notify_all(); + return true; + }); + + { + std::unique_lock lk(mutex); + CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return valueCount == N; })); + } + + node3.cancelListen(foo, ftokenfoo.get()); +} + } // namespace test diff --git a/tests/dhtrunnertester.h b/tests/dhtrunnertester.h index b358d396..0069a5ed 100644 --- a/tests/dhtrunnertester.h +++ b/tests/dhtrunnertester.h @@ -57,6 +57,10 @@ class DhtRunnerTester : public CppUnit::TestFixture { * Test listen method */ void testListen(); + /** + * Test listen method with lot of datas + */ + void testListenLotOfBytes(); }; } // namespace test