diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 2571cba1..b771bcb6 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -28,20 +28,23 @@ namespace dht { -struct DhtProxyClient::InfoState { +struct DhtProxyClient::InfoState +{ std::atomic_uint ipv4 {0}, ipv6 {0}; std::atomic_bool cancel {false}; }; -struct DhtProxyClient::OperationState { +struct DhtProxyClient::OperationState +{ std::atomic_bool ok {true}; std::atomic_bool stop {false}; }; struct DhtProxyClient::Listener { - Listener(OpValueCache&& c): - cache(std::move(c)) {} + Listener(OpValueCache&& c) + : cache(std::move(c)) + {} OpValueCache cache; CacheValueCallback cb; @@ -52,10 +55,12 @@ struct DhtProxyClient::Listener #endif }; -struct PermanentPut { - PermanentPut(const Sp& v, std::unique_ptr&& j, - const Sp& o): - value(v), refreshPutTimer(std::move(j)), ok(o) +struct PermanentPut +{ + PermanentPut(const Sp& v, std::unique_ptr&& j, const Sp& o) + : value(v) + , refreshPutTimer(std::move(j)) + , ok(o) {} Sp value; @@ -63,19 +68,20 @@ struct PermanentPut { Sp ok; }; -struct DhtProxyClient::ProxySearch { +struct DhtProxyClient::ProxySearch +{ SearchCache ops {}; std::unique_ptr opExpirationTimer; std::map listeners {}; std::map puts {}; - std::set> pendingPuts {}; + std::set> pendingPuts {}; }; -struct LineSplit { - void append(const char* d, size_t l) { - buf_.insert(buf_.end(), d, d+l); - } - bool getLine(char c) { +struct LineSplit +{ + void append(const char* d, size_t l) { buf_.insert(buf_.end(), d, d + l); } + bool getLine(char c) + { auto it = buf_.begin(); while (it != buf_.end()) { if (*(it++) == c) { @@ -87,39 +93,41 @@ struct LineSplit { } return false; } - const std::string& line() const { return line_; } + const std::string& line() const { return line_; } + private: std::deque buf_ {}; std::string line_ {}; }; std::string -getRandomSessionId(size_t length = 8) { - static constexpr const char chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~"; +getRandomSessionId(size_t length = 8) +{ + static constexpr const char chars[] + = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~"; std::string str(length, 0); std::random_device rdev; - std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 2); - std::generate_n( str.begin(), length, [&]{ return chars[dist(rdev)]; } ); + std::uniform_int_distribution<> dist(0, (sizeof(chars) / sizeof(char)) - 2); + std::generate_n(str.begin(), length, [&] { return chars[dist(rdev)]; }); return str; } DhtProxyClient::DhtProxyClient() {} -DhtProxyClient::DhtProxyClient( - std::shared_ptr serverCA, - dht::crypto::Identity clientIdentity, - std::function signal, - const std::string& serverHost, - const std::string& userAgent, - const std::string& pushClientId, - const std::string& pushToken, - const std::string& pushTopic, - const std::string& pushPlatform, - std::shared_ptr logger -) +DhtProxyClient::DhtProxyClient(std::shared_ptr serverCA, + dht::crypto::Identity clientIdentity, + std::function signal, + const std::string& serverHost, + const std::string& userAgent, + const std::string& pushClientId, + const std::string& pushToken, + const std::string& pushTopic, + const std::string& pushPlatform, + std::shared_ptr logger) : DhtInterface(logger) , proxyUrl_(serverHost) - , clientIdentity_(clientIdentity), serverCertificate_(serverCA) + , clientIdentity_(clientIdentity) + , serverCertificate_(serverCA) , userAgent_(userAgent) , pushClientId_(pushClientId) , pushSessionId_(getRandomSessionId()) @@ -127,7 +135,7 @@ DhtProxyClient::DhtProxyClient( , notificationTopic_(pushTopic) , platform_(pushPlatform) , loopSignal_(signal) - , jsonReader_(Json::CharReaderBuilder{}.newCharReader()) + , jsonReader_(Json::CharReaderBuilder {}.newCharReader()) { localAddrv4_.setFamily(AF_INET); localAddrv6_.setFamily(AF_INET6); @@ -137,13 +145,13 @@ DhtProxyClient::DhtProxyClient( if (logger_) { if (serverCertificate_) logger_->d("[proxy:client] using ca certificate for ssl:\n%s", - serverCertificate_->toString(false/*chain*/).c_str()); + serverCertificate_->toString(false /*chain*/).c_str()); if (clientIdentity_.first and clientIdentity_.second) logger_->d("[proxy:client] using client certificate for ssl:\n%s", - clientIdentity_.second->toString(false/*chain*/).c_str()); + clientIdentity_.second->toString(false /*chain*/).c_str()); } // run http client - httpClientThread_ = std::thread([this](){ + httpClientThread_ = std::thread([this]() { try { if (logger_) logger_->d("[proxy:client] starting io_context"); @@ -152,8 +160,7 @@ DhtProxyClient::DhtProxyClient( httpContext_.run(); if (logger_) logger_->d("[proxy:client] http client io_context stopped"); - } - catch(const std::exception& ex){ + } catch (const std::exception& ex) { if (logger_) logger_->e("[proxy:client] run error: %s", ex.what()); } @@ -180,11 +187,11 @@ DhtProxyClient::startProxy() } void -DhtProxyClient::handleProxyConfirm(const asio::error_code &ec) +DhtProxyClient::handleProxyConfirm(const asio::error_code& ec) { if (ec == asio::error::operation_aborted) return; - else if (ec){ + else if (ec) { if (logger_) logger_->e("[proxy:client] confirm error: %s", ec.message().c_str()); return; @@ -227,7 +234,8 @@ DhtProxyClient::stop() } std::vector> -DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const { +DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const +{ std::lock_guard lock(searchLock_); auto s = searches_.find(k); if (s == searches_.end()) @@ -236,7 +244,8 @@ DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const { } Sp -DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const { +DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const +{ std::lock_guard lock(searchLock_); auto s = searches_.find(k); if (s == searches_.end()) @@ -250,8 +259,8 @@ DhtProxyClient::cancelAllListeners() std::lock_guard lock(searchLock_); if (logger_) logger_->d("[proxy:client] [listeners] [%zu searches] cancel all", searches_.size()); - for (auto& s: searches_) { - s.second.ops.cancelAll([&](size_t token){ + for (auto& s : searches_) { + s.second.ops.cancelAll([&](size_t token) { auto l = s.second.listeners.find(token); if (l == s.second.listeners.end()) return; @@ -275,8 +284,7 @@ NodeStatus DhtProxyClient::getStatus(sa_family_t af) const { std::lock_guard l(lockCurrentProxyInfos_); - switch (af) - { + switch (af) { case AF_INET: return statusIpv4_; case AF_INET6: @@ -290,8 +298,7 @@ bool DhtProxyClient::isRunning(sa_family_t af) const { std::lock_guard l(lockCurrentProxyInfos_); - switch (af) - { + switch (af) { case AF_INET: return statusIpv4_ != NodeStatus::Disconnected; case AF_INET6: @@ -317,7 +324,8 @@ DhtProxyClient::periodic(const uint8_t*, size_t, SockAddr, const time_point& /*n } void -DhtProxyClient::setHeaderFields(http::Request& request){ +DhtProxyClient::setHeaderFields(http::Request& request) +{ request.set_header_field(restinio::http_field_t::accept, "*/*"); request.set_header_field(restinio::http_field_t::content_type, "application/json"); } @@ -329,26 +337,25 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va logger_->d("[proxy:client] [get] [search %s]", key.to_c_str()); if (isDestroying_) { - if (donecb) donecb(false, {}); + if (donecb) + donecb(false, {}); return; } try { auto request = buildRequest("/key/" + key.toString()); auto reqid = request->id(); - //request->set_connection_type(restinio::http_connection_header_t::keep_alive); + // request->set_connection_type(restinio::http_connection_header_t::keep_alive); request->set_method(restinio::http_method_get()); setHeaderFields(*request); auto opstate = std::make_shared(); - request->add_on_body_callback([ - this, - key, - opstate, - filter = Value::Filter::chain(std::move(f), w.getFilter()), - rxBuf = std::make_shared(), - cb - ](const char* at, size_t length){ + request->add_on_body_callback([this, + key, + opstate, + filter = Value::Filter::chain(std::move(f), w.getFilter()), + rxBuf = std::make_shared(), + cb](const char* at, size_t length) { try { rxBuf->append(at, length); // one value per body line @@ -357,7 +364,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va std::string err; Json::Value json; const auto& line = rxBuf->line(); - if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){ + if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)) { opstate->ok.store(false); return; } @@ -368,21 +375,21 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va if (not values.empty() and cb) { { std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([opstate, cb, values = std::move(values)](){ - if (not opstate->stop.load() and not cb(values)){ + callbacks_.emplace_back([opstate, cb, values = std::move(values)]() { + if (not opstate->stop.load() and not cb(values)) { opstate->stop.store(true); } }); } loopSignal_(); } - } catch(const std::exception& e) { + } catch (const std::exception& e) { if (logger_) logger_->e("[proxy:client] [get %s] body parsing error: %s", key.to_c_str(), e.what()); opstate->ok.store(false); } }); - request->add_on_done_callback([this, reqid, opstate, donecb, key] (const http::Response& response){ + request->add_on_done_callback([this, reqid, opstate, donecb, key](const http::Response& response) { if (response.status_code != 200) { if (logger_) logger_->e("[proxy:client] [get %s] failed with code=%i", key.to_c_str(), response.status_code); @@ -393,7 +400,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va if (donecb) { { std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([donecb, opstate](){ + callbacks_.emplace_back([donecb, opstate]() { donecb(opstate->ok, {}); opstate->stop.store(true); }); @@ -410,8 +417,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va requests_[reqid] = request; } request->send(); - } - catch (const std::exception &e){ + } catch (const std::exception& e) { if (logger_) logger_->e("[proxy:client] [get %s] error: %s", key.to_c_str(), e.what()); } @@ -421,7 +427,8 @@ void DhtProxyClient::put(const InfoHash& key, Sp val, DoneCallback cb, time_point created, bool permanent) { if (not val or isDestroying_) { - if (cb) cb(false, {}); + if (cb) + cb(false, {}); return; } if (logger_) @@ -434,35 +441,40 @@ DhtProxyClient::put(const InfoHash& key, Sp val, DoneCallback cb, time_po auto& search = searches_[key]; if (val->id) { auto id = val->id; - auto refreshPutTimer = std::make_unique(httpContext_, proxy::OP_TIMEOUT - proxy::OP_MARGIN); - refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id)); + auto refreshPutTimer = std::make_unique(httpContext_, + proxy::OP_TIMEOUT - proxy::OP_MARGIN); + refreshPutTimer->async_wait( + std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id)); search.puts.erase(id); search.puts.emplace(std::piecewise_construct, - std::forward_as_tuple(id), - std::forward_as_tuple(val, std::move(refreshPutTimer), ok)); + std::forward_as_tuple(id), + std::forward_as_tuple(val, std::move(refreshPutTimer), ok)); } else { search.pendingPuts.emplace(val); } } - doPut(key, val, [this, cb, ok](bool result){ - if (ok) - *ok = result; - if (cb) { - std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([cb, result](){ - cb(result, {}); - }); - } - loopSignal_(); - }, created, permanent); + doPut( + key, + val, + [this, cb, ok](bool result) { + if (ok) + *ok = result; + if (cb) { + std::lock_guard lock(lockCallbacks_); + callbacks_.emplace_back([cb, result]() { cb(result, {}); }); + } + loopSignal_(); + }, + created, + permanent); } void -DhtProxyClient::handleRefreshPut(const asio::error_code &ec, InfoHash key, Value::Id id) +DhtProxyClient::handleRefreshPut(const asio::error_code& ec, InfoHash key, Value::Id id) { if (ec == asio::error::operation_aborted) return; - else if (ec){ + else if (ec) { if (logger_) logger_->e("[proxy:client] [put] [refresh %s] %s", key.toString().c_str(), ec.message().c_str()); return; @@ -473,12 +485,11 @@ DhtProxyClient::handleRefreshPut(const asio::error_code &ec, InfoHash key, Value auto search = searches_.find(key); if (search != searches_.end()) { auto p = search->second.puts.find(id); - if (p != search->second.puts.end()){ - doPut(key, p->second.value, [ok = p->second.ok](bool result){ - *ok = result; - }, time_point::max(), true); + if (p != search->second.puts.end()) { + doPut(key, p->second.value, [ok = p->second.ok](bool result) { *ok = result; }, time_point::max(), true); p->second.refreshPutTimer->expires_after(proxy::OP_TIMEOUT - proxy::OP_MARGIN); - p->second.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id)); + p->second.refreshPutTimer->async_wait( + std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id)); } } } @@ -491,9 +502,8 @@ DhtProxyClient::buildRequest(const std::string& target) l.unlock(); if (not resolver) resolver = std::make_shared(httpContext_, proxyUrl_, logger_); - auto request = target.empty() - ? std::make_shared(httpContext_, resolver) - : std::make_shared(httpContext_, resolver, target); + auto request = target.empty() ? std::make_shared(httpContext_, resolver) + : std::make_shared(httpContext_, resolver, target); if (serverCertificate_) request->set_certificate_authority(serverCertificate_); if (clientIdentity_.first and clientIdentity_.second) @@ -529,13 +539,16 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallbackSimple cb, } } request->set_body(Json::writeString(jsonBuilder_, json)); - request->add_on_done_callback([this, reqid, cb, val, key, permanent] (const http::Response& response){ + request->add_on_done_callback([this, reqid, cb, val, key, permanent](const http::Response& response) { bool ok = response.status_code == 200; if (ok) { if (val->id == Value::INVALID_ID) { std::string err; Json::Value parsedValue; - if (jsonReader_->parse(response.body.data(), response.body.data() + response.body.size(), &parsedValue, &err)){ + if (jsonReader_->parse(response.body.data(), + response.body.data() + response.body.size(), + &parsedValue, + &err)) { auto id = dht::Value(parsedValue).id; val->id = id; if (permanent) { @@ -544,11 +557,14 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallbackSimple cb, auto it = search.pendingPuts.find(val); if (it != search.pendingPuts.end()) { auto sok = std::make_shared(ok); - auto refreshPutTimer = std::make_unique(httpContext_, proxy::OP_TIMEOUT - proxy::OP_MARGIN); - refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id)); + auto refreshPutTimer = std::make_unique(httpContext_, + proxy::OP_TIMEOUT + - proxy::OP_MARGIN); + refreshPutTimer->async_wait( + std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id)); search.puts.emplace(std::piecewise_construct, - std::forward_as_tuple(id), - std::forward_as_tuple(val, std::move(refreshPutTimer), sok)); + std::forward_as_tuple(id), + std::forward_as_tuple(val, std::move(refreshPutTimer), sok)); search.pendingPuts.erase(it); } } @@ -575,8 +591,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallbackSimple cb, requests_[reqid] = request; } request->send(); - } - catch (const std::exception &e){ + } catch (const std::exception& e) { if (logger_) logger_->error("[proxy:client] [put {}] error: {}", key, e.what()); opFailed(); @@ -587,7 +602,8 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallbackSimple cb, * Get data currently being put at the given hash. */ std::vector> -DhtProxyClient::getPut(const InfoHash& key) const { +DhtProxyClient::getPut(const InfoHash& key) const +{ std::vector> ret; auto search = searches_.find(key); if (search != searches_.end()) { @@ -602,7 +618,8 @@ DhtProxyClient::getPut(const InfoHash& key) const { * Get data currently being put at the given hash with the given id. */ Sp -DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) const { +DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) const +{ auto search = searches_.find(key); if (search == searches_.end()) return {}; @@ -671,21 +688,25 @@ DhtProxyClient::queryProxyInfo(const Sp& infoState, const Spid(); request->set_method(restinio::http_method_get()); setHeaderFields(*request); - request->add_on_done_callback([this, reqid, family, infoState] (const http::Response& response){ + request->add_on_done_callback([this, reqid, family, infoState](const http::Response& response) { if (infoState->cancel.load()) return; if (response.status_code != 200) { if (logger_) logger_->e("[proxy:client] [status] ipv%i failed with code=%i", - family == AF_INET ? 4 : 6, response.status_code); + family == AF_INET ? 4 : 6, + response.status_code); // pass along the failures if ((family == AF_INET and infoState->ipv4 == 0) or (family == AF_INET6 and infoState->ipv6 == 0)) - onProxyInfos(Json::Value{}, family); + onProxyInfos(Json::Value {}, family); } else { std::string err; Json::Value proxyInfos; - if (!jsonReader_->parse(response.body.data(), response.body.data() + response.body.size(), &proxyInfos, &err)){ - onProxyInfos(Json::Value{}, family); + if (!jsonReader_->parse(response.body.data(), + response.body.data() + response.body.size(), + &proxyInfos, + &err)) { + onProxyInfos(Json::Value {}, family); } else if (not infoState->cancel) { if (auto req = response.request.lock()) { if (auto conn = req->get_connection()) { @@ -709,8 +730,7 @@ DhtProxyClient::queryProxyInfo(const Sp& infoState, const Spsend(); - } - catch (const std::exception &e){ + } catch (const std::exception& e) { if (logger_) logger_->e("[proxy:client] [status] error sending request: %s", e.what()); } @@ -725,8 +745,8 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa auto oldStatus = std::max(statusIpv4_, statusIpv6_); auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_; auto ipChanged = false; - auto& pubAddress = family == AF_INET? publicAddressV4_ : publicAddressV6_; - auto& localAddress = family == AF_INET? localAddrv4_ : localAddrv6_; + auto& pubAddress = family == AF_INET ? publicAddressV4_ : publicAddressV6_; + auto& localAddress = family == AF_INET ? localAddrv4_ : localAddrv6_; if (not proxyInfos.isMember("node_id")) { if (logger_) logger_->e("[proxy:client] [info] request failed for %s", family == AF_INET ? "ipv4" : "ipv6"); @@ -737,8 +757,7 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa } } else { if (logger_) - logger_->debug("[proxy:client] [info] got proxy reply for {}", - family == AF_INET ? "ipv4" : "ipv6"); + logger_->debug("[proxy:client] [info] got proxy reply for {}", family == AF_INET ? "ipv4" : "ipv6"); try { myid = InfoHash(proxyInfos["node_id"].asString()); stats4_ = NodeStats(proxyInfos["ipv4"]); @@ -752,7 +771,7 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa std::string localIp = proxyInfos["local_ip"].asString(); if (localAddress.toString() != localIp) { localAddress.setAddress(localIp.c_str()); - ipChanged = (bool)localAddress; + ipChanged = (bool) localAddress; } } @@ -763,9 +782,8 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa if (publicAddressV6_) addresses.emplace_back(publicAddressV6_); std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([cb=publicAddressChangedCb_, addresses = std::move(addresses)](){ - cb(std::move(addresses)); - }); + callbacks_.emplace_back( + [cb = publicAddressChangedCb_, addresses = std::move(addresses)]() { cb(std::move(addresses)); }); } if (!ipChanged && stats4_.good_nodes + stats6_.good_nodes) @@ -797,14 +815,15 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa } } nextProxyConfirmationTimer_->expires_at(std::chrono::steady_clock::now() + std::chrono::minutes(15)); - nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1)); - } - else if (newStatus == NodeStatus::Disconnected) { + nextProxyConfirmationTimer_->async_wait( + std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1)); + } else if (newStatus == NodeStatus::Disconnected) { auto next = std::chrono::steady_clock::now(); if (!ipChanged) next += std::chrono::minutes(1); nextProxyConfirmationTimer_->expires_at(next); - nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1)); + nextProxyConfirmationTimer_->async_wait( + std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1)); } l.unlock(); loopSignal_(); @@ -815,8 +834,9 @@ DhtProxyClient::parsePublicAddress(const Json::Value& val) { auto public_ip = val.asString(); auto [host, service] = splitPort(public_ip); - auto sa = SockAddr::resolve(host); - if (sa.empty()) return {}; + auto sa = SockAddr::resolve(std::string(host)); + if (sa.empty()) + return {}; return sa.front().getMappedIPv4(); } @@ -825,8 +845,10 @@ DhtProxyClient::getPublicAddress(sa_family_t family) { std::lock_guard l(lockCurrentProxyInfos_); std::vector result; - if (publicAddressV6_ && family != AF_INET) result.emplace_back(publicAddressV6_); - if (publicAddressV4_ && family != AF_INET6) result.emplace_back(publicAddressV4_); + if (publicAddressV6_ && family != AF_INET) + result.emplace_back(publicAddressV6_); + if (publicAddressV4_ && family != AF_INET6) + result.emplace_back(publicAddressV4_); return result; } @@ -840,83 +862,90 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt std::lock_guard lock(searchLock_); auto& search = searches_[key]; - auto query = std::make_shared(Select{}, std::move(where)); - return search.ops.listen(cb, query, std::move(filter), [this, key](Sp, ValueCallback cb, SyncCallback) -> size_t { - // Find search - auto search = searches_.find(key); - if (search == searches_.end()) { - if (logger_) - logger_->e("[proxy:client] [listen] [search %s] search not found", key.to_c_str()); - return 0; - } - if (logger_) - logger_->d("[proxy:client] [listen] [search %s] sending %s", key.to_c_str(), - deviceKey_.empty() ? "listen" : "subscribe"); - // Add listener - auto token = ++listenerToken_; - auto l = search->second.listeners.find(token); - if (l == search->second.listeners.end()) { - l = search->second.listeners.emplace(std::piecewise_construct, - std::forward_as_tuple(token), - std::forward_as_tuple(std::move(cb))).first; - } else { - if (l->second.opstate) - l->second.opstate->stop = true; - } - // Add cache callback - auto opstate = std::make_shared(); - l->second.opstate = opstate; - l->second.cb = [this,key,token,opstate](const std::vector>& values, bool expired, system_clock::time_point t){ - if (opstate->stop) - return false; - std::lock_guard lock(searchLock_); - auto s = searches_.find(key); - if (s != searches_.end()) { - auto l = s->second.listeners.find(token); - if (l != s->second.listeners.end()) { - return l->second.cache.onValue(values, expired, t); - } + auto query = std::make_shared(Select {}, std::move(where)); + return search.ops + .listen(cb, query, std::move(filter), [this, key](Sp, ValueCallback cb, SyncCallback) -> size_t { + // Find search + auto search = searches_.find(key); + if (search == searches_.end()) { + if (logger_) + logger_->e("[proxy:client] [listen] [search %s] search not found", key.to_c_str()); + return 0; } - return false; - }; + if (logger_) + logger_->d("[proxy:client] [listen] [search %s] sending %s", + key.to_c_str(), + deviceKey_.empty() ? "listen" : "subscribe"); + // Add listener + auto token = ++listenerToken_; + auto l = search->second.listeners.find(token); + if (l == search->second.listeners.end()) { + l = search->second.listeners + .emplace(std::piecewise_construct, + std::forward_as_tuple(token), + std::forward_as_tuple(std::move(cb))) + .first; + } else { + if (l->second.opstate) + l->second.opstate->stop = true; + } + // Add cache callback + auto opstate = std::make_shared(); + l->second.opstate = opstate; + l->second.cb = [this, key, token, opstate](const std::vector>& values, + bool expired, + system_clock::time_point t) { + if (opstate->stop) + return false; + std::lock_guard lock(searchLock_); + auto s = searches_.find(key); + if (s != searches_.end()) { + auto l = s->second.listeners.find(token); + if (l != s->second.listeners.end()) { + return l->second.cache.onValue(values, expired, t); + } + } + return false; + }; #ifdef OPENDHT_PUSH_NOTIFICATIONS - if (not deviceKey_.empty()) { - /* - * Relaunch push listeners even if a timeout is not received - * (if the proxy crash for any reason) - */ - if (!l->second.refreshSubscriberTimer) - l->second.refreshSubscriberTimer = std::make_unique(httpContext_); - l->second.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + - proxy::OP_TIMEOUT - proxy::OP_MARGIN); - l->second.refreshSubscriberTimer->async_wait(std::bind(&DhtProxyClient::handleResubscribe, this, - std::placeholders::_1, key, token, opstate)); - } + if (not deviceKey_.empty()) { + /* + * Relaunch push listeners even if a timeout is not received + * (if the proxy crash for any reason) + */ + if (!l->second.refreshSubscriberTimer) + l->second.refreshSubscriberTimer = std::make_unique(httpContext_); + l->second.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT + - proxy::OP_MARGIN); + l->second.refreshSubscriberTimer->async_wait( + std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate)); + } #endif - ListenMethod method; - restinio::http_request_header_t header; - if (deviceKey_.empty()){ // listen - method = ListenMethod::LISTEN; - header.method(restinio::http_method_get()); - header.request_target("/key/" + key.toString() + "/listen"); - } - else { - method = ListenMethod::SUBSCRIBE; - header.method(restinio::http_method_subscribe()); - header.request_target("/key/" + key.toString()); - } - sendListen(header, l->second.cb, opstate, l->second, method); - return token; - }); + ListenMethod method; + restinio::http_request_header_t header; + if (deviceKey_.empty()) { // listen + method = ListenMethod::LISTEN; + header.method(restinio::http_method_get()); + header.request_target("/key/" + key.toString() + "/listen"); + } else { + method = ListenMethod::SUBSCRIBE; + header.method(restinio::http_method_subscribe()); + header.request_target("/key/" + key.toString()); + } + sendListen(header, l->second.cb, opstate, l->second, method); + return token; + }); } void -DhtProxyClient::handleResubscribe(const asio::error_code &ec, const InfoHash& key, - const size_t token, std::shared_ptr opstate) +DhtProxyClient::handleResubscribe(const asio::error_code& ec, + const InfoHash& key, + const size_t token, + std::shared_ptr opstate) { if (ec == asio::error::operation_aborted) return; - else if (ec){ + else if (ec) { if (logger_) logger_->e("[proxy:client] [resubscribe %s] %s", key.toString().c_str(), ec.message().c_str()); return; @@ -925,12 +954,11 @@ DhtProxyClient::handleResubscribe(const asio::error_code &ec, const InfoHash& ke return; std::lock_guard lock(searchLock_); auto s = searches_.find(key); - if (s != searches_.end()){ + if (s != searches_.end()) { auto l = s->second.listeners.find(token); if (l != s->second.listeners.end()) { resubscribe(key, token, l->second); - } - else { + } else { if (logger_) logger_->e("[proxy:client] [resubscribe %s] token not found", key.toString().c_str()); } @@ -956,16 +984,17 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) it->second.opExpirationTimer = std::make_unique(httpContext_, ops.getExpiration()); else it->second.opExpirationTimer->expires_at(ops.getExpiration()); - it->second.opExpirationTimer->async_wait(std::bind(&DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key)); + it->second.opExpirationTimer->async_wait( + std::bind(&DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key)); return canceled; } void -DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& key) +DhtProxyClient::handleExpireListener(const asio::error_code& ec, const InfoHash& key) { if (ec == asio::error::operation_aborted) return; - else if (ec){ + else if (ec) { if (logger_) logger_->e("[proxy:client] [listen %s] error in cancel: %s", key.toString().c_str(), ec.message().c_str()); return; @@ -999,11 +1028,12 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& body["key"] = deviceKey_; body["client_id"] = pushClientId_; request->set_body(Json::writeString(jsonBuilder_, body)); - request->add_on_done_callback([this, reqid, key] (const http::Response& response){ + request->add_on_done_callback([this, reqid, key](const http::Response& response) { if (response.status_code != 200) { if (logger_) logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i", - key.to_c_str(), response.status_code); + key.to_c_str(), + response.status_code); if (not response.aborted and response.status_code == 0) opFailed(); } @@ -1017,10 +1047,9 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& requests_[reqid] = request; } request->send(); - } - catch (const std::exception &e){ + } catch (const std::exception& e) { if (logger_) - logger_->e("[proxy:client] [unsubscribe %s] failed: %s", key.to_c_str(), e.what()); + logger_->e("[proxy:client] [unsubscribe %s] failed: %s", key.to_c_str(), e.what()); } } else { // stop the request @@ -1032,14 +1061,15 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& search->second.listeners.erase(it); if (logger_) logger_->d("[proxy:client] [listen:cancel] [search %s] %zu listener remaining", - key.to_c_str(), search->second.listeners.size()); + key.to_c_str(), + search->second.listeners.size()); }); - if (next != time_point::max()){ + if (next != time_point::max()) { search->second.opExpirationTimer->expires_at(next); - search->second.opExpirationTimer->async_wait(std::bind( - &DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key)); + search->second.opExpirationTimer->async_wait( + std::bind(&DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key)); } - if (search->second.listeners.empty()){ + if (search->second.listeners.empty()) { searches_.erase(search); } } @@ -1048,10 +1078,11 @@ void DhtProxyClient::sendListen(const restinio::http_request_header_t& header, const CacheValueCallback& cb, const Sp& opstate, - Listener& listener, ListenMethod method) + Listener& listener, + ListenMethod method) { if (logger_) - logger_->e("[proxy:client] [listen] sendListen: %d", (int)method); + logger_->e("[proxy:client] [listen] sendListen: %d", (int) method); try { auto request = buildRequest(); listener.request = request; @@ -1067,7 +1098,7 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header, request->set_body(body); #endif auto rxBuf = std::make_shared(); - request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length){ + request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length) { try { auto& b = *rxBuf; b.append(at, length); @@ -1077,7 +1108,7 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header, std::string err; Json::Value json; const auto& line = b.line(); - if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){ + if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)) { opstate->ok.store(false); return; } @@ -1086,29 +1117,31 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header, } auto value = std::make_shared(json); - if (cb){ + if (cb) { auto expired = json.get("expired", Json::Value(false)).asBool(); { std::lock_guard lock(lockCallbacks_); callbacks_.emplace_back([cb, value, opstate, expired]() { - if (not opstate->stop.load() and not cb({value}, expired, system_clock::time_point::min())) + if (not opstate->stop.load() + and not cb({value}, expired, system_clock::time_point::min())) opstate->stop.store(true); }); } loopSignal_(); } } - } catch(const std::exception& e) { + } catch (const std::exception& e) { if (logger_) logger_->e("[proxy:client] [listen] request #%i error in parsing: %s", reqid, e.what()); opstate->ok.store(false); } }); - request->add_on_done_callback([this, opstate, reqid] (const http::Response& response) { + request->add_on_done_callback([this, opstate, reqid](const http::Response& response) { if (response.status_code != 200) { if (logger_) logger_->e("[proxy:client] [listen] send request #%i failed with code=%i", - reqid, response.status_code); + reqid, + response.status_code); opstate->ok.store(false); if (not response.aborted and response.status_code == 0) opFailed(); @@ -1122,15 +1155,14 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header, std::lock_guard l(requestLock_); requests_[reqid] = request; } - request->add_on_status_callback([request, seconds = this->listenKeepIdle()] (unsigned status_code) { - if(status_code == 200) { + request->add_on_status_callback([request, seconds = this->listenKeepIdle()](unsigned status_code) { + if (status_code == 200) { // increase TCP_KEEPIDLE to save power request->get_connection()->set_keepalive(seconds); } }); request->send(); - } - catch (const std::exception &e){ + } catch (const std::exception& e) { if (logger_) logger_->e("[proxy:client] [listen] request failed: %s", e.what()); } @@ -1162,11 +1194,11 @@ DhtProxyClient::getConnectivityStatus() } void -DhtProxyClient::restartListeners(const asio::error_code &ec) +DhtProxyClient::restartListeners(const asio::error_code& ec) { if (ec == asio::error::operation_aborted) return; - else if (ec){ + else if (ec) { if (logger_) logger_->e("[proxy:client] restart error: %s", ec.message().c_str()); return; @@ -1181,15 +1213,14 @@ DhtProxyClient::restartListeners(const asio::error_code &ec) for (auto& search : searches_) { const auto& key = search.first; for (auto& put : search.second.puts) { - doPut(key, put.second.value, [ok = put.second.ok](bool result){ - *ok = result; - }, time_point::max(), true); + doPut(key, put.second.value, [ok = put.second.ok](bool result) { *ok = result; }, time_point::max(), true); if (!put.second.refreshPutTimer) { put.second.refreshPutTimer = std::make_unique(httpContext_); } - put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); - put.second.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, - std::placeholders::_1, key, put.first)); + put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT + - proxy::OP_MARGIN); + put.second.refreshPutTimer->async_wait( + std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, put.first)); } // Restart failed pending puts for (auto& pput : search.second.pendingPuts) { @@ -1208,8 +1239,8 @@ DhtProxyClient::restartListeners(const asio::error_code &ec) } if (logger_) logger_->d("[proxy:client] [listeners] restarting listeners"); - for (auto& search: searches_) { - for (auto& l: search.second.listeners) { + for (auto& search : searches_) { + for (auto& l : search.second.listeners) { auto& listener = l.second; if (auto opstate = listener.opstate) opstate->stop = true; @@ -1217,8 +1248,8 @@ DhtProxyClient::restartListeners(const asio::error_code &ec) listener.request.reset(); } } - for (auto& search: searches_) { - for (auto& l: search.second.listeners) { + for (auto& search : searches_) { + for (auto& l : search.second.listeners) { auto& listener = l.second; auto opstate = listener.opstate; // Redo listen @@ -1267,23 +1298,27 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::mapsecond); auto& put = search.puts.at(vid); if (!put.refreshPutTimer) - put.refreshPutTimer = std::make_unique(httpContext_, std::chrono::steady_clock::now()); + put.refreshPutTimer = std::make_unique(httpContext_, + std::chrono::steady_clock::now()); else put.refreshPutTimer->expires_at(std::chrono::steady_clock::now()); - put.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid)); + put.refreshPutTimer->async_wait( + std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid)); ret = PushNotificationResult::PutRefresh; } else { // Refresh listen for (auto& list : search.listeners) resubscribe(key, list.first, list.second); - ret = search.listeners.empty() ? PushNotificationResult::IgnoredNoOp : PushNotificationResult::ListenRefresh; + ret = search.listeners.empty() ? PushNotificationResult::IgnoredNoOp + : PushNotificationResult::ListenRefresh; } } else { auto key = InfoHash(notification.at("key")); system_clock::time_point sendTime = system_clock::time_point::min(); try { sendTime = system_clock::time_point(std::chrono::milliseconds(std::stoull(notification.at("t")))); - } catch (...) {} + } catch (...) { + } auto& search = searches_.at(key); auto expired = notification.find("exp"); for (auto& list : search.listeners) { @@ -1296,18 +1331,19 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map>& vals) { - return cb(vals, false, sendTime); - }, [cb, oldValues, sendTime](bool /*ok*/) { - // Decrement old values refcount to expire values not - // present in the new list - cb(oldValues, true, sendTime); - }); + get( + key, + [cb, sendTime](const std::vector>& vals) { return cb(vals, false, sendTime); }, + [cb, oldValues, sendTime](bool /*ok*/) { + // Decrement old values refcount to expire values not + // present in the new list + cb(oldValues, true, sendTime); + }); ret = PushNotificationResult::Values; } else { std::stringstream ss(expired->second); std::vector ids; - while(ss.good()) { + while (ss.good()) { std::string substr; getline(ss, substr, ','); ids.emplace_back(std::stoull(substr)); @@ -1347,7 +1383,9 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map(httpContext_); } - listener.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + - proxy::OP_TIMEOUT - proxy::OP_MARGIN); - listener.refreshSubscriberTimer->async_wait(std::bind(&DhtProxyClient::handleResubscribe, this, - std::placeholders::_1, key, token, opstate)); + listener.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); + listener.refreshSubscriberTimer->async_wait( + std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate)); auto vcb = listener.cb; sendListen(header, vcb, opstate, listener, ListenMethod::RESUBSCRIBE); #endif @@ -1405,7 +1442,8 @@ DhtProxyClient::fillBody(bool resubscribe) #endif // OPENDHT_PUSH_NOTIFICATIONS void -DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token) { +DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token) +{ #ifdef OPENDHT_PUSH_NOTIFICATIONS std::unique_lock l(lockCurrentProxyInfos_); if (deviceKey_ != token) { @@ -1423,5 +1461,4 @@ DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& tok #endif } - } // namespace dht diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index a6c29fab..6d2f027d 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -40,18 +40,20 @@ using namespace std::literals; constexpr auto PEER_DISCOVERY_DHT_SERVICE = "dht"sv; -struct NodeInsertionPack { +struct NodeInsertionPack +{ dht::InfoHash nodeId; in_port_t port; dht::NetId net; MSGPACK_DEFINE(nodeId, port, net) }; -DhtRunner::DhtRunner() : dht_() +DhtRunner::DhtRunner() + : dht_() { #ifdef _WIN32 WSADATA wsd; - if (WSAStartup(MAKEWORD(2,2), &wsd) != 0) + if (WSAStartup(MAKEWORD(2, 2), &wsd) != 0) throw DhtException("Can't initialize Winsock2"); #endif } @@ -95,7 +97,8 @@ DhtRunner::run(const Config& config, Context&& context) auto expected = State::Idle; if (not running.compare_exchange_strong(expected, State::Running)) { if (context.logger) - context.logger->w("[runner %p] Node is already running. Call join() first before calling run() again.", fmt::ptr(this)); + context.logger->w("[runner %p] Node is already running. Call join() first before calling run() again.", + fmt::ptr(this)); return; } @@ -104,7 +107,9 @@ DhtRunner::run(const Config& config, Context&& context) auto local6 = config.bind6; if (not local4 and not local6) { if (context.logger) - context.logger->w("[runner %p] No address to bind specified in the configuration, using default addresses", fmt::ptr(this)); + context.logger + ->w("[runner %p] No address to bind specified in the configuration, using default addresses", + fmt::ptr(this)); local4.setFamily(AF_INET); local6.setFamily(AF_INET6); } @@ -118,14 +123,18 @@ DhtRunner::run(const Config& config, Context&& context) if (inConfig >> port) { if (local4.getPort() == 0) { if (context.logger) - context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", fmt::ptr(this), port); + context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", + fmt::ptr(this), + port); local4.setPort(port); } } if (inConfig >> port) { if (local6.getPort() == 0) { if (context.logger) - context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", fmt::ptr(this), port); + context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", + fmt::ptr(this), + port); local6.setPort(port); } } @@ -146,7 +155,7 @@ DhtRunner::run(const Config& config, Context&& context) if (not context.sock) { context.sock.reset(new net::UdpSocket(local4, local6, context.logger)); } - context.sock->setOnReceive([&] (net::PacketList&& pkts) { + context.sock->setOnReceive([&](net::PacketList&& pkts) { net::PacketList ret; { std::lock_guard lck(sock_mtx); @@ -169,12 +178,18 @@ DhtRunner::run(const Config& config, Context&& context) outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl; outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl; } - auto dht = std::make_unique(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger, std::move(context.rng)); - dht_ = std::make_unique(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger); + auto dht = std::make_unique(std::move(context.sock), + SecureDht::getConfig(config.dht_config), + context.logger, + std::move(context.rng)); + dht_ = std::make_unique(std::move(dht), + config.dht_config, + std::move(context.identityAnnouncedCb), + context.logger); } else { enableProxy(true); } - } catch(const std::exception& e) { + } catch (const std::exception& e) { config_ = {}; identityAnnouncedCb_ = {}; dht_.reset(); @@ -215,7 +230,8 @@ DhtRunner::run(const Config& config, Context&& context) if (not pending_ops_prio.empty()) return true; auto s = getStatus(); - if (not pending_ops.empty() and (s == NodeStatus::Connected or s == NodeStatus::Disconnected or running == State::Stopping)) + if (not pending_ops.empty() + and (s == NodeStatus::Connected or s == NodeStatus::Disconnected or running == State::Stopping)) return true; } return false; @@ -230,9 +246,8 @@ DhtRunner::run(const Config& config, Context&& context) if (config.proxy_server.empty()) { if (config.peer_discovery or config.peer_publish) { #ifdef OPENDHT_PEER_DISCOVERY - peerDiscovery_ = context.peerDiscovery ? - std::move(context.peerDiscovery) : - std::make_shared(); + peerDiscovery_ = context.peerDiscovery ? std::move(context.peerDiscovery) + : std::make_shared(); #else std::cerr << "Peer discovery requested but OpenDHT built without peer discovery support." << std::endl; #endif @@ -241,12 +256,13 @@ DhtRunner::run(const Config& config, Context&& context) #ifdef OPENDHT_PEER_DISCOVERY auto netId = config.dht_config.node_config.network; if (config.peer_discovery) { - peerDiscovery_->startDiscovery(PEER_DISCOVERY_DHT_SERVICE, [this, netId](NodeInsertionPack&& v, SockAddr&& addr){ - addr.setPort(v.port); - if (v.nodeId != dht_->getNodeId() && netId == v.net){ - bootstrap(v.nodeId, addr); - } - }); + peerDiscovery_->startDiscovery(PEER_DISCOVERY_DHT_SERVICE, + [this, netId](NodeInsertionPack&& v, SockAddr&& addr) { + addr.setPort(v.port); + if (v.nodeId != dht_->getNodeId() && netId == v.net) { + bootstrap(v.nodeId, addr); + } + }); } if (config.peer_publish) { msgpack::sbuffer sbuf_node; @@ -273,11 +289,9 @@ DhtRunner::run(const Config& config, Context&& context) statusCbs.emplace_back([this](NodeStatus status4, NodeStatus status6) { if (status4 == NodeStatus::Disconnected && status6 == NodeStatus::Disconnected) { peerDiscovery_->connectivityChanged(); - } - else if (status4 != NodeStatus::Connected || status6 != NodeStatus::Connected) { + } else if (status4 != NodeStatus::Connected || status6 != NodeStatus::Connected) { peerDiscovery_->stopConnectivityChanged(); } - }); } #endif @@ -285,15 +299,15 @@ DhtRunner::run(const Config& config, Context&& context) } void -DhtRunner::shutdown(ShutdownCallback cb, bool stop) { +DhtRunner::shutdown(ShutdownCallback cb, bool stop) +{ std::unique_lock lck(storage_mtx); auto expected = State::Running; if (not running.compare_exchange_strong(expected, State::Stopping)) { if (expected == State::Stopping and ongoing_ops) { if (cb) shutdownCallbacks_.emplace_back(std::move(cb)); - } - else if (cb) { + } else if (cb) { lck.unlock(); cb(); } @@ -304,7 +318,9 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) { ongoing_ops++; shutdownCallbacks_.emplace_back(std::move(cb)); pending_ops.emplace([=](SecureDht&) mutable { - auto onShutdown = [this]{ opEnded(); }; + auto onShutdown = [this] { + opEnded(); + }; if (dht_) dht_->shutdown(onShutdown, stop); else @@ -314,29 +330,35 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) { } void -DhtRunner::opEnded() { +DhtRunner::opEnded() +{ if (--ongoing_ops == 0) checkShutdown(); } DoneCallback -DhtRunner::bindOpDoneCallback(DoneCallback&& cb) { - return [this, cb = std::move(cb)](bool ok, const std::vector>& nodes){ - if (cb) cb(ok, nodes); +DhtRunner::bindOpDoneCallback(DoneCallback&& cb) +{ + return [this, cb = std::move(cb)](bool ok, const std::vector>& nodes) { + if (cb) + cb(ok, nodes); opEnded(); }; } DoneCallbackSimple -DhtRunner::bindOpDoneCallback(DoneCallbackSimple&& cb) { - return [this, cb = std::move(cb)](bool ok){ - if (cb) cb(ok); +DhtRunner::bindOpDoneCallback(DoneCallbackSimple&& cb) +{ + return [this, cb = std::move(cb)](bool ok) { + if (cb) + cb(ok); opEnded(); }; } bool -DhtRunner::checkShutdown() { +DhtRunner::checkShutdown() +{ decltype(shutdownCallbacks_) cbs; { std::lock_guard lck(storage_mtx); @@ -345,7 +367,8 @@ DhtRunner::checkShutdown() { cbs = std::move(shutdownCallbacks_); } for (auto& cb : cbs) - if (cb) cb(); + if (cb) + cb(); return true; } @@ -390,16 +413,18 @@ DhtRunner::join() } SockAddr -DhtRunner::getBound(sa_family_t af) const { +DhtRunner::getBound(sa_family_t af) const +{ std::lock_guard lck(dht_mtx); if (dht_) if (auto sock = dht_->getSocket()) return sock->getBound(af); - return SockAddr{}; + return SockAddr {}; } in_port_t -DhtRunner::getBoundPort(sa_family_t af) const { +DhtRunner::getBoundPort(sa_family_t af) const +{ std::lock_guard lck(dht_mtx); if (dht_) if (auto sock = dht_->getSocket()) @@ -417,24 +442,24 @@ DhtRunner::dumpTables() const InfoHash DhtRunner::getId() const { - return dht_ ? dht_->getId() : InfoHash{}; + return dht_ ? dht_->getId() : InfoHash {}; } std::shared_ptr DhtRunner::getPublicKey() const { - return dht_ ? dht_->getPublicKey() : std::shared_ptr{}; + return dht_ ? dht_->getPublicKey() : std::shared_ptr {}; } InfoHash DhtRunner::getNodeId() const { - return dht_ ? dht_->getNodeId() : InfoHash{}; + return dht_ ? dht_->getNodeId() : InfoHash {}; } - std::pair -DhtRunner::getStoreSize() const { +DhtRunner::getStoreSize() const +{ std::lock_guard lck(dht_mtx); if (!dht_) return {}; @@ -442,7 +467,8 @@ DhtRunner::getStoreSize() const { } void -DhtRunner::setStorageLimit(size_t limit) { +DhtRunner::setStorageLimit(size_t limit) +{ std::lock_guard lck(dht_mtx); if (!dht_) throw std::runtime_error("dht is not running"); @@ -450,7 +476,8 @@ DhtRunner::setStorageLimit(size_t limit) { } std::vector -DhtRunner::exportNodes() const { +DhtRunner::exportNodes() const +{ std::lock_guard lck(dht_mtx); if (!dht_) return {}; @@ -458,7 +485,8 @@ DhtRunner::exportNodes() const { } std::vector -DhtRunner::exportValues() const { +DhtRunner::exportValues() const +{ std::lock_guard lck(dht_mtx); if (!dht_) return {}; @@ -466,7 +494,8 @@ DhtRunner::exportValues() const { } void -DhtRunner::setLogger(const Sp& logger) { +DhtRunner::setLogger(const Sp& logger) +{ std::lock_guard lck(dht_mtx); logger_ = logger; if (dht_) @@ -474,26 +503,33 @@ DhtRunner::setLogger(const Sp& logger) { } void -DhtRunner::setLogFilter(const InfoHash& f) { +DhtRunner::setLogFilter(const InfoHash& f) +{ std::lock_guard lck(dht_mtx); if (dht_) dht_->setLogFilter(f); } void -DhtRunner::registerType(const ValueType& type) { +DhtRunner::registerType(const ValueType& type) +{ std::lock_guard lck(dht_mtx); dht_->registerType(type); } void -DhtRunner::importValues(const std::vector& values) { +DhtRunner::importValues(const std::vector& values) +{ std::lock_guard lck(dht_mtx); dht_->importValues(values); } unsigned -DhtRunner::getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const +DhtRunner::getNodesStats(sa_family_t af, + unsigned* good_return, + unsigned* dubious_return, + unsigned* cached_return, + unsigned* incoming_return) const { std::lock_guard lck(dht_mtx); const auto stats = dht_->getNodesStats(af); @@ -516,7 +552,8 @@ DhtRunner::getNodesStats(sa_family_t af) const } NodeInfo -DhtRunner::getNodeInfo() const { +DhtRunner::getNodeInfo() const +{ std::lock_guard lck(dht_mtx); NodeInfo info {}; if (dht_) { @@ -538,7 +575,7 @@ DhtRunner::getNodeInfo(std::function)> cb) { std::lock_guard lck(storage_mtx); ongoing_ops++; - pending_ops_prio.emplace([cb = std::move(cb), this](SecureDht& dht){ + pending_ops_prio.emplace([cb = std::move(cb), this](SecureDht& dht) { auto sinfo = std::make_shared(); auto& info = *sinfo; info.id = dht.getId(); @@ -616,7 +653,7 @@ DhtRunner::getPublicAddress(std::function&&)> cb, sa_ { std::lock_guard lck(storage_mtx); ongoing_ops++; - pending_ops_prio.emplace([cb = std::move(cb), this, af](SecureDht& dht){ + pending_ops_prio.emplace([cb = std::move(cb), this, af](SecureDht& dht) { cb(dht.getPublicAddress(af)); opEnded(); }); @@ -624,19 +661,22 @@ DhtRunner::getPublicAddress(std::function&&)> cb, sa_ } void -DhtRunner::registerCertificate(const std::shared_ptr& cert) { +DhtRunner::registerCertificate(const std::shared_ptr& cert) +{ std::lock_guard lck(dht_mtx); dht_->registerCertificate(cert); } void -DhtRunner::setLocalCertificateStore(CertificateStoreQueryLegacy&& query_method) { +DhtRunner::setLocalCertificateStore(CertificateStoreQueryLegacy&& query_method) +{ std::lock_guard lck(dht_mtx); if (dht_) dht_->setLocalCertificateStore(std::forward(query_method)); } -void DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) +void +DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) { std::lock_guard lck(dht_mtx); if (dht_) @@ -653,8 +693,10 @@ DhtRunner::loop_() { std::lock_guard lck(storage_mtx); auto s = getStatus(); - ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or s == NodeStatus::Disconnected or running == State::Stopping)) ? - std::move(pending_ops) : std::move(pending_ops_prio); + ops = (pending_ops_prio.empty() + && (s == NodeStatus::Connected or s == NodeStatus::Disconnected or running == State::Stopping)) + ? std::move(pending_ops) + : std::move(pending_ops_prio); } while (not ops.empty()) { ops.front()(*dht_); @@ -713,7 +755,7 @@ DhtRunner::loop_() if (nstatus4 != status4 || nstatus6 != status6) { status4 = nstatus4; status6 = nstatus6; - for (auto& cb : statusCbs){ + for (auto& cb : statusCbs) { cb(status4, status6); } } @@ -727,7 +769,8 @@ DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (dcb) dcb(false, {}); + if (dcb) + dcb(false, {}); return; } ongoing_ops++; @@ -743,11 +786,13 @@ DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb, get(InfoHash::get(key), std::move(vcb), std::move(dcb), std::move(f), std::move(w)); } void -DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) { +DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) +{ std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (done_cb) done_cb(false, {}); + if (done_cb) + done_cb(false, {}); return; } ongoing_ops++; @@ -814,14 +859,12 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (cb) cb(false, {}); + if (cb) + cb(false, {}); return; } ongoing_ops++; - pending_ops.emplace([=, - cb = std::move(cb), - sv = std::make_shared(std::move(value)) - ] (SecureDht& dht) mutable { + pending_ops.emplace([=, cb = std::move(cb), sv = std::make_shared(std::move(value))](SecureDht& dht) mutable { dht.put(hash, sv, bindOpDoneCallback(std::move(cb)), created, permanent); }); cv.notify_all(); @@ -833,7 +876,8 @@ DhtRunner::put(InfoHash hash, std::shared_ptr value, DoneCallback cb, tim std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (cb) cb(false, {}); + if (cb) + cb(false, {}); return; } ongoing_ops++; @@ -853,9 +897,7 @@ void DhtRunner::cancelPut(const InfoHash& h, Value::Id id) { std::lock_guard lck(storage_mtx); - pending_ops.emplace([=](SecureDht& dht) { - dht.cancelPut(h, id); - }); + pending_ops.emplace([=](SecureDht& dht) { dht.cancelPut(h, id); }); cv.notify_all(); } @@ -863,9 +905,7 @@ void DhtRunner::cancelPut(const InfoHash& h, const std::shared_ptr& value) { std::lock_guard lck(storage_mtx); - pending_ops.emplace([=](SecureDht& dht) { - dht.cancelPut(h, value->id); - }); + pending_ops.emplace([=](SecureDht& dht) { dht.cancelPut(h, value->id); }); cv.notify_all(); } @@ -875,14 +915,12 @@ DhtRunner::putSigned(InfoHash hash, std::shared_ptr value, DoneCallback c std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (cb) cb(false, {}); + if (cb) + cb(false, {}); return; } ongoing_ops++; - pending_ops.emplace([=, - cb = std::move(cb), - value = std::move(value) - ](SecureDht& dht) mutable { + pending_ops.emplace([=, cb = std::move(cb), value = std::move(value)](SecureDht& dht) mutable { dht.putSigned(hash, value, bindOpDoneCallback(std::move(cb)), permanent); }); cv.notify_all(); @@ -906,14 +944,12 @@ DhtRunner::putEncrypted(InfoHash hash, const PkId& to, std::shared_ptr va std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (cb) cb(false, {}); + if (cb) + cb(false, {}); return; } ongoing_ops++; - pending_ops.emplace([=, - cb = std::move(cb), - value = std::move(value) - ] (SecureDht& dht) mutable { + pending_ops.emplace([=, cb = std::move(cb), value = std::move(value)](SecureDht& dht) mutable { dht.putEncrypted(hash, to, value, bindOpDoneCallback(std::move(cb)), permanent); }); cv.notify_all(); @@ -925,14 +961,12 @@ DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr value std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (cb) cb(false, {}); + if (cb) + cb(false, {}); return; } ongoing_ops++; - pending_ops.emplace([=, - cb = std::move(cb), - value = std::move(value) - ] (SecureDht& dht) mutable { + pending_ops.emplace([=, cb = std::move(cb), value = std::move(value)](SecureDht& dht) mutable { dht.putEncrypted(hash, to, value, bindOpDoneCallback(std::move(cb)), permanent); }); cv.notify_all(); @@ -951,26 +985,29 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done } void -DhtRunner::putEncrypted(InfoHash hash, const std::shared_ptr& to, std::shared_ptr value, DoneCallback cb, bool permanent) +DhtRunner::putEncrypted(InfoHash hash, + const std::shared_ptr& to, + std::shared_ptr value, + DoneCallback cb, + bool permanent) { std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (cb) cb(false, {}); + if (cb) + cb(false, {}); return; } ongoing_ops++; - pending_ops.emplace([=, - cb = std::move(cb), - value = std::move(value) - ] (SecureDht& dht) mutable { + pending_ops.emplace([=, cb = std::move(cb), value = std::move(value)](SecureDht& dht) mutable { dht.putEncrypted(hash, *to, value, bindOpDoneCallback(std::move(cb)), permanent); }); cv.notify_all(); } void -DhtRunner::putEncrypted(InfoHash hash, const std::shared_ptr& to, Value&& value, DoneCallback cb, bool permanent) +DhtRunner::putEncrypted( + InfoHash hash, const std::shared_ptr& to, Value&& value, DoneCallback cb, bool permanent) { putEncrypted(hash, to, std::make_shared(std::move(value)), std::move(cb), permanent); } @@ -979,9 +1016,7 @@ void DhtRunner::bootstrap(const std::string& host, const std::string& service) { std::lock_guard lck(storage_mtx); - pending_ops_prio.emplace([host, service] (SecureDht& dht) mutable { - dht.addBootstrap(host, service); - }); + pending_ops_prio.emplace([host, service](SecureDht& dht) mutable { dht.addBootstrap(host, service); }); cv.notify_all(); } @@ -989,9 +1024,9 @@ void DhtRunner::bootstrap(const std::string& hostService) { std::lock_guard lck(storage_mtx); - pending_ops_prio.emplace([host_service = splitPort(hostService)] (SecureDht& dht) mutable { - dht.addBootstrap(host_service.first, host_service.second); - }); + auto [h, s] = splitPort(hostService); + pending_ops_prio.emplace( + [host = std::string(h), service = std::string(s)](SecureDht& dht) { dht.addBootstrap(host, service); }); cv.notify_all(); } @@ -999,9 +1034,7 @@ void DhtRunner::clearBootstrap() { std::lock_guard lck(storage_mtx); - pending_ops_prio.emplace([] (SecureDht& dht) mutable { - dht.clearBootstrap(); - }); + pending_ops_prio.emplace([](SecureDht& dht) mutable { dht.clearBootstrap(); }); cv.notify_all(); } @@ -1014,15 +1047,12 @@ DhtRunner::bootstrap(std::vector nodes, DoneCallbackSimple cb) } std::lock_guard lck(storage_mtx); ongoing_ops++; - pending_ops_prio.emplace([ - cb = bindOpDoneCallback(std::move(cb)), - nodes = std::move(nodes) - ] (SecureDht& dht) mutable { + pending_ops_prio.emplace([cb = bindOpDoneCallback(std::move(cb)), nodes = std::move(nodes)](SecureDht& dht) mutable { auto rem = cb ? std::make_shared>(nodes.size(), false) : nullptr; for (auto& node : nodes) { if (node.getPort() == 0) node.setPort(net::DHT_DEFAULT_PORT); - dht.pingNode(std::move(node), [rem,cb](bool ok) { + dht.pingNode(std::move(node), [rem, cb](bool ok) { auto& r = *rem; r.first--; r.second |= ok; @@ -1041,7 +1071,8 @@ DhtRunner::bootstrap(SockAddr addr, DoneCallbackSimple cb) std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); - if (cb) cb(false); + if (cb) + cb(false); return; } ongoing_ops++; @@ -1057,9 +1088,7 @@ DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address) std::lock_guard lck(storage_mtx); if (running != State::Running) return; - pending_ops_prio.emplace([id, address](SecureDht& dht) mutable { - dht.insertNode(id, address); - }); + pending_ops_prio.emplace([id, address](SecureDht& dht) mutable { dht.insertNode(id, address); }); cv.notify_all(); } @@ -1091,7 +1120,8 @@ DhtRunner::connectivityChanged() } void -DhtRunner::findCertificate(InfoHash hash, std::function&)> cb) { +DhtRunner::findCertificate(InfoHash hash, std::function&)> cb) +{ std::unique_lock lck(storage_mtx); if (running != State::Running) { lck.unlock(); @@ -1099,8 +1129,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function& crt){ + pending_ops.emplace([this, hash, cb = std::move(cb)](SecureDht& dht) { + dht.findCertificate(hash, [this, cb = std::move(cb)](const Sp& crt) { cb(crt); opEnded(); }); @@ -1108,7 +1138,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function&)> cb) +void +DhtRunner::findCertificate(PkId hash, std::function&)> cb) { std::unique_lock lck(storage_mtx); if (running != State::Running) { @@ -1116,8 +1147,8 @@ void DhtRunner::findCertificate(PkId hash, std::function& crt){ + pending_ops.emplace([this, hash, cb = std::move(cb)](SecureDht& dht) { + dht.findCertificate(hash, [this, cb = std::move(cb)](const Sp& crt) { cb(crt); opEnded(); }); @@ -1159,22 +1190,22 @@ DhtRunner::enableProxy(bool proxify) if (proxify) { // Init the proxy client auto dht_via_proxy = std::make_unique( - config_.server_ca, - config_.client_identity, - [this]{ - if (config_.threaded) { - std::lock_guard lck(storage_mtx); - pending_ops_prio.emplace([=](SecureDht&) mutable {}); - cv.notify_all(); - } - }, - config_.proxy_server, - config_.proxy_user_agent, - config_.push_node_id, - config_.push_token, - config_.push_topic, - config_.push_platform, - logger_); + config_.server_ca, + config_.client_identity, + [this] { + if (config_.threaded) { + std::lock_guard lck(storage_mtx); + pending_ops_prio.emplace([=](SecureDht&) mutable {}); + cv.notify_all(); + } + }, + config_.proxy_server, + config_.proxy_user_agent, + config_.push_node_id, + config_.push_token, + config_.push_topic, + config_.push_platform, + logger_); dht_ = std::make_unique(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_); // and use it use_proxy = proxify; @@ -1199,7 +1230,8 @@ DhtRunner::forwardAllMessages(bool forward) * Updates the push notification device token */ void -DhtRunner::setPushNotificationToken(const std::string& token) { +DhtRunner::setPushNotificationToken(const std::string& token) +{ std::lock_guard lck(dht_mtx); #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS) config_.push_token = token; @@ -1211,7 +1243,8 @@ DhtRunner::setPushNotificationToken(const std::string& token) { } void -DhtRunner::setPushNotificationTopic(const std::string& topic) { +DhtRunner::setPushNotificationTopic(const std::string& topic) +{ std::lock_guard lck(dht_mtx); #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS) config_.push_topic = topic; @@ -1223,7 +1256,8 @@ DhtRunner::setPushNotificationTopic(const std::string& topic) { } void -DhtRunner::setPushNotificationPlatform(const std::string& platform) { +DhtRunner::setPushNotificationPlatform(const std::string& platform) +{ std::lock_guard lck(dht_mtx); #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS) config_.push_platform = platform; @@ -1256,4 +1290,4 @@ DhtRunner::pushNotificationReceived([[maybe_unused]] const std::map(id); v->msgpack_unpack_body(msg.get()); if (v->recipient != key.getPublicKey().getId()) @@ -355,13 +360,13 @@ FieldValueIndex::FieldValueIndex(const Value& v, const Select& s) { if (not s.empty()) { auto selection = s.getSelection(); - std::transform(selection.begin(), selection.end(), std::inserter(index, index.end()), - [](const std::set::value_type& f) { - return std::make_pair(f, FieldValue {}); - }); + std::transform(selection.begin(), + selection.end(), + std::inserter(index, index.end()), + [](const std::set::value_type& f) { return std::make_pair(f, FieldValue {}); }); } else { index.clear(); - for (size_t f = 1 ; f < static_cast(Value::Field::COUNT) ; ++f) + for (size_t f = 1; f < static_cast(Value::Field::COUNT); ++f) index[static_cast(f)] = {}; } for (const auto& fvp : index) { @@ -374,13 +379,15 @@ FieldValueIndex::FieldValueIndex(const Value& v, const Select& s) index[f] = {f, v.type}; break; case Value::Field::OwnerPk: - index[f] = {f, v.owner ? v.owner->getId() : InfoHash() }; + index[f] = {f, v.owner ? v.owner->getId() : InfoHash()}; break; case Value::Field::SeqNum: index[f] = {f, v.seq}; break; case Value::Field::UserType: - index[f] = {f, Blob {v.user_type.begin(), v.user_type.end()}}; + index[f] = { + f, Blob {v.user_type.begin(), v.user_type.end()} + }; break; default: break; @@ -388,7 +395,9 @@ FieldValueIndex::FieldValueIndex(const Value& v, const Select& s) } } -bool FieldValueIndex::containedIn(const FieldValueIndex& other) const { +bool +FieldValueIndex::containedIn(const FieldValueIndex& other) const +{ if (index.size() > other.index.size()) return false; for (const auto& field : index) { @@ -399,7 +408,9 @@ bool FieldValueIndex::containedIn(const FieldValueIndex& other) const { return true; } -std::ostream& operator<<(std::ostream& os, const FieldValueIndex& fvi) { +std::ostream& +operator<<(std::ostream& os, const FieldValueIndex& fvi) +{ os << "Index["; for (auto v = fvi.index.begin(); v != fvi.index.end(); ++v) { switch (v->first) { @@ -438,7 +449,7 @@ FieldValueIndex::msgpack_unpack_fields(const std::set& fields, con unsigned j = 0; for (const auto& field : fields) { - auto& field_value = o.via.array.ptr[offset+(j++)]; + auto& field_value = o.via.array.ptr[offset + (j++)]; switch (field) { case Value::Field::Id: case Value::Field::ValueType: @@ -457,13 +468,16 @@ FieldValueIndex::msgpack_unpack_fields(const std::set& fields, con } } -void trim_str(std::string& str) { +void +trim_str(std::string& str) +{ auto first = std::min(str.size(), str.find_first_not_of(' ')); auto last = std::min(str.size(), str.find_last_not_of(' ')); str = str.substr(first, last - first + 1); } -Select::Select(std::string_view q_str) { +Select::Select(std::string_view q_str) +{ std::istringstream q_iss {std::string(q_str)}; std::string token {}; q_iss >> token; @@ -488,7 +502,8 @@ Select::Select(std::string_view q_str) { } } -Where::Where(std::string_view q_str) { +Where::Where(std::string_view q_str) +{ std::istringstream q_iss {std::string(q_str)}; std::string token {}; q_iss >> token; @@ -509,11 +524,9 @@ Where::Where(std::string_view q_str) { std::string s {}; std::istringstream convert {value_str}; convert >> v; - if (not convert - and value_str.size() > 1 - and value_str[0] == '"' - and value_str[value_str.size()-1] == '"') - s = value_str.substr(1, value_str.size()-2); + if (not convert and value_str.size() > 1 and value_str[0] == '"' + and value_str[value_str.size() - 1] == '"') + s = value_str.substr(1, value_str.size() - 2); else s = value_str; if (field_str == VALUE_KEY_ID) @@ -527,7 +540,8 @@ Where::Where(std::string_view q_str) { else if (field_str == "user_type") userType(s); else - throw std::invalid_argument(Query::QUERY_PARSE_ERROR + " (WHERE) incorrect token near: " + field_str); + throw std::invalid_argument(Query::QUERY_PARSE_ERROR + + " (WHERE) incorrect token near: " + field_str); } } } @@ -536,24 +550,25 @@ Where::Where(std::string_view q_str) { void Query::msgpack_unpack(const msgpack::object& o) { - if (o.type != msgpack::type::MAP) - throw msgpack::type_error(); + if (o.type != msgpack::type::MAP) + throw msgpack::type_error(); - auto rfilters = findMapValue(o, "w"sv); /* unpacking filters */ - if (rfilters) + auto rfilters = findMapValue(o, "w"sv); /* unpacking filters */ + if (rfilters) where.msgpack_unpack(*rfilters); - else - throw msgpack::type_error(); + else + throw msgpack::type_error(); - auto rfield_selector = findMapValue(o, "s"sv); /* unpacking field selectors */ - if (rfield_selector) + auto rfield_selector = findMapValue(o, "s"sv); /* unpacking field selectors */ + if (rfield_selector) select.msgpack_unpack(*rfield_selector); - else - throw msgpack::type_error(); + else + throw msgpack::type_error(); } -template -bool subset(const std::vector& fds, const std::vector& qfds) +template +bool +subset(const std::vector& fds, const std::vector& qfds) { for (const auto& fd : fds) { if (std::find_if(qfds.begin(), qfds.end(), [&fd](const T& _vfd) { return fd == _vfd; }) == qfds.end()) @@ -562,22 +577,28 @@ bool subset(const std::vector& fds, const std::vector& qfds) return true; } -bool Select::isSatisfiedBy(const Select& os) const { +bool +Select::isSatisfiedBy(const Select& os) const +{ /* empty, means all values are selected. */ - return fieldSelection_.empty() ? - os.fieldSelection_.empty() : - subset(fieldSelection_, os.fieldSelection_); + return fieldSelection_.empty() ? os.fieldSelection_.empty() : subset(fieldSelection_, os.fieldSelection_); } -bool Where::isSatisfiedBy(const Where& ow) const { +bool +Where::isSatisfiedBy(const Where& ow) const +{ return subset(ow.filters_, filters_); } -bool Query::isSatisfiedBy(const Query& q) const { +bool +Query::isSatisfiedBy(const Query& q) const +{ return none or (where.isSatisfiedBy(q.where) and select.isSatisfiedBy(q.select)); } -std::ostream& operator<<(std::ostream& s, const dht::Select& select) { +std::ostream& +operator<<(std::ostream& s, const dht::Select& select) +{ s << "SELECT "; if (select.fieldSelection_.empty()) s << '*'; @@ -608,10 +629,12 @@ std::ostream& operator<<(std::ostream& s, const dht::Select& select) { return s; } -std::ostream& operator<<(std::ostream& s, const dht::Where& where) { +std::ostream& +operator<<(std::ostream& s, const dht::Where& where) +{ if (not where.filters_.empty()) { s << "WHERE "; - for (auto f = where.filters_.begin() ; f != where.filters_.end() ; ++f) { + for (auto f = where.filters_.begin(); f != where.filters_.end(); ++f) { switch (f->getField()) { case Value::Field::Id: s << VALUE_KEY_ID << '=' << f->getInt(); @@ -639,5 +662,4 @@ std::ostream& operator<<(std::ostream& s, const dht::Where& where) { return s; } - -} +} // namespace dht