mirrors/opendht
1
0 Fork 0
mirror of https://github.com/savoirfairelinux/opendht.git synced 2026-11-08 12:28:42 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
f7243fca3771b974681a6f263104d27cacc7d162
opendht/src/dht_proxy_client.cpp
T
Adrien Béraud e5e5a45b91 op_cache, proxy_client: guard SearchCache::get and push refresh expiry 
SearchCache::get called the GetCallback unconditionally; Dht::query
passes none, which threw bad_function_call as soon as a listen cache
existed for the key. Return false without a callback so the network
path runs. A callback returning false means the consumer has enough
values, not a failure: report done(true) as the network path does.

DhtProxyClient push refresh expired every previously cached value once
the refresh get completed, including values that were just re-delivered
or edited, producing phantom expire callbacks. Track the ids received by
the get and only expire cached values missing from it, and skip the
reconciliation when the get failed so a transient error does not expire
values still present on the DHT.
2026-09-04 15:18:24 -04:00

1459 lines
54 KiB
C++
Raw Blame History

// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
// SPDX-License-Identifier: MIT
#include "dht_proxy_client.h"
#include "dhtrunner.h"
#include "op_cache.h"
#include "utils.h"
#include <llhttp.h>
#include <deque>
namespace dht {
struct DhtProxyClient::InfoState
{
std::atomic_uint ipv4 {0}, ipv6 {0};
std::atomic_bool cancel {false};
};
struct DhtProxyClient::OperationState
{
std::atomic_bool ok {true};
std::atomic_bool stop {false};
};
struct DhtProxyClient::Listener
{
// The proxy server is a single logical source that re-delivers values
// (push refresh, listen restart): don't count duplicates.
Listener(ValueCallback&& cb)
: cache(std::move(cb), false)
{}
OpValueCache cache;
CacheValueCallback cb;
Sp<OperationState> opstate;
std::shared_ptr<http::Request> request;
#ifdef OPENDHT_PUSH_NOTIFICATIONS
std::unique_ptr<asio::system_timer> refreshSubscriberTimer;
#endif
};
struct PermanentPut
{
PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::system_timer>&& j, const Sp<std::atomic_bool>& o)
: value(v)
, refreshPutTimer(std::move(j))
, ok(o)
{}
Sp<Value> value;
std::unique_ptr<asio::system_timer> refreshPutTimer;
Sp<std::atomic_bool> ok;
};
struct DhtProxyClient::ProxySearch
{
SearchCache ops {false};
std::unique_ptr<asio::steady_timer> opExpirationTimer;
std::map<size_t, Listener> listeners {};
std::map<Value::Id, PermanentPut> puts {};
std::set<Sp<Value>> pendingPuts {};
};
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) {
line_.clear();
line_.insert(line_.end(), buf_.begin(), it);
buf_.erase(buf_.begin(), it);
return true;
}
}
return false;
}
const std::string& line() const { return line_; }
private:
std::deque<char> buf_ {};
std::string line_ {};
};
std::string
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)]; });
return str;
}
DhtProxyClient::DhtProxyClient() {}
DhtProxyClient::DhtProxyClient(std::shared_ptr<dht::crypto::Certificate> serverCA,
dht::crypto::Identity clientIdentity,
std::function<void()> 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<dht::Logger> logger)
: DhtInterface(logger)
, proxyUrl_(serverHost)
, clientIdentity_(clientIdentity)
, serverCertificate_(serverCA)
, userAgent_(userAgent)
, pushClientId_(pushClientId)
, pushSessionId_(getRandomSessionId())
, deviceKey_(pushToken)
, notificationTopic_(pushTopic)
, platform_(pushPlatform)
, loopSignal_(signal)
, jsonReader_(Json::CharReaderBuilder {}.newCharReader())
{
localAddrv4_.setFamily(AF_INET);
localAddrv6_.setFamily(AF_INET6);
jsonBuilder_["commentStyle"] = "None";
jsonBuilder_["indentation"] = "";
if (logger_) {
if (serverCertificate_)
logger_->debug("[proxy:client] using ca certificate for ssl:\n{}",
serverCertificate_->toString(false /*chain*/));
if (clientIdentity_.first and clientIdentity_.second)
logger_->debug("[proxy:client] using client certificate for ssl:\n{}",
clientIdentity_.second->toString(false /*chain*/));
}
// run http client
httpClientThread_ = std::thread([this]() {
try {
if (logger_)
logger_->debug("[proxy:client] starting io_context");
// Ensures the httpContext_ won't run out of work
auto work = asio::make_work_guard(httpContext_);
httpContext_.run();
if (logger_)
logger_->debug("[proxy:client] http client io_context stopped");
} catch (const std::exception& ex) {
if (logger_)
logger_->error("[proxy:client] run error: {}", ex.what());
}
});
if (!proxyUrl_.empty())
startProxy();
}
void
DhtProxyClient::startProxy()
{
if (proxyUrl_.empty())
return;
if (logger_)
logger_->debug("[proxy:client] start proxy with {}", proxyUrl_);
nextProxyConfirmationTimer_ = std::make_unique<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
listenerRestartTimer_ = std::make_unique<asio::steady_timer>(httpContext_);
loopSignal_();
}
void
DhtProxyClient::handleProxyConfirm(const asio::error_code& ec)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->error("[proxy:client] confirm error: {}", ec.message());
return;
}
if (proxyUrl_.empty())
return;
getConnectivityStatus();
}
DhtProxyClient::~DhtProxyClient()
{
stop();
}
void
DhtProxyClient::stop()
{
if (not isDestroying_.exchange(true)) {
std::shared_ptr<http::Resolver> resolver;
{
std::lock_guard l(resolverLock_);
resolver = std::move(resolver_);
}
if (resolver)
resolver->cancel();
cancelAllListeners();
if (infoState_)
infoState_->cancel = true;
{
std::lock_guard lock(requestLock_);
for (auto& request : requests_)
request.second->cancel();
}
if (not httpContext_.stopped())
httpContext_.stop();
if (httpClientThread_.joinable())
httpClientThread_.join();
requests_.clear();
}
}
std::vector<Sp<Value>>
DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const
{
std::lock_guard lock(searchLock_);
auto s = searches_.find(k);
if (s == searches_.end())
return {};
return s->second.ops.get(filter);
}
Sp<Value>
DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const
{
std::lock_guard lock(searchLock_);
auto s = searches_.find(k);
if (s == searches_.end())
return {};
return s->second.ops.get(id);
}
void
DhtProxyClient::cancelAllListeners()
{
std::lock_guard lock(searchLock_);
if (logger_)
logger_->debug("[proxy:client] [listeners] [{} searches] cancel all", searches_.size());
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;
l->second.opstate->stop.store(true);
l->second.request->cancel();
// implicit request.reset()
s.second.listeners.erase(token);
});
}
}
void
DhtProxyClient::shutdown(ShutdownCallback cb, bool)
{
stop();
if (cb)
cb();
}
NodeStatus
DhtProxyClient::getStatus(sa_family_t af) const
{
std::lock_guard l(lockCurrentProxyInfos_);
switch (af) {
case AF_INET:
return statusIpv4_;
case AF_INET6:
return statusIpv6_;
default:
return NodeStatus::Disconnected;
}
}
bool
DhtProxyClient::isRunning(sa_family_t af) const
{
std::lock_guard l(lockCurrentProxyInfos_);
switch (af) {
case AF_INET:
return statusIpv4_ != NodeStatus::Disconnected;
case AF_INET6:
return statusIpv6_ != NodeStatus::Disconnected;
default:
return false;
}
}
time_point
DhtProxyClient::periodic(const uint8_t*, size_t, SockAddr, const time_point& /*now*/)
{
// Exec all currently stored callbacks
decltype(callbacks_) callbacks;
{
std::lock_guard lock(lockCallbacks_);
callbacks = std::move(callbacks_);
}
for (auto& callback : callbacks)
callback();
callbacks.clear();
return time_point::max();
}
void
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");
}
void
DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w)
{
if (logger_)
logger_->debug("[proxy:client] [get] [search {}]", key.to_view());
if (isDestroying_) {
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_method(restinio::http_method_get());
setHeaderFields(*request);
auto opstate = std::make_shared<OperationState>();
request->add_on_body_callback([this,
key,
opstate,
filter = Value::Filter::chain(std::move(f), w.getFilter()),
rxBuf = std::make_shared<LineSplit>(),
cb](const char* at, size_t length) {
try {
rxBuf->append(at, length);
// one value per body line
std::vector<Sp<Value>> values;
while (rxBuf->getLine('\n') and !opstate->stop) {
std::string err;
Json::Value json;
const auto& line = rxBuf->line();
if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)) {
opstate->ok.store(false);
return;
}
auto value = std::make_shared<Value>(json);
if ((not filter or filter(*value)) and cb)
values.emplace_back(std::move(value));
}
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)) {
opstate->stop.store(true);
}
});
}
loopSignal_();
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [get {}] body parsing error: {}", key.to_view(), e.what());
opstate->ok.store(false);
}
});
request->add_on_done_callback([this, reqid, opstate, donecb, key](const http::Response& response) {
if (response.status_code != 200) {
if (logger_)
logger_->error("[proxy:client] [get {}] failed with code={}", key.to_view(), response.status_code);
opstate->ok.store(false);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (donecb) {
{
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back([donecb, opstate]() {
donecb(opstate->ok, {});
opstate->stop.store(true);
});
}
loopSignal_();
}
if (not isDestroying_) {
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [get {}] error: {}", key.to_view(), e.what());
}
}
void
DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
{
if (not val or isDestroying_) {
if (cb)
cb(false, {});
return;
}
if (logger_)
logger_->debug("[proxy:client] [put] [search {}]", key.to_view());
std::shared_ptr<std::atomic_bool> ok;
if (permanent) {
std::lock_guard lock(searchLock_);
ok = std::make_shared<std::atomic_bool>(true);
auto& search = searches_[key];
if (val->id) {
auto id = val->id;
auto refreshPutTimer = std::make_unique<asio::system_timer>(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));
} 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);
}
void
DhtProxyClient::handleRefreshPut(const asio::error_code& ec, InfoHash key, Value::Id id)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->error("[proxy:client] [put] [refresh {}] {}", key.to_view(), ec.message());
return;
}
if (logger_)
logger_->debug("[proxy:client] [put] [refresh {}]", key.to_view());
std::lock_guard lock(searchLock_);
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);
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));
}
}
}
std::shared_ptr<http::Request>
DhtProxyClient::buildRequest(const std::string& target)
{
std::unique_lock l(resolverLock_);
auto resolver = resolver_;
l.unlock();
if (not resolver)
resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
auto request = target.empty() ? std::make_shared<http::Request>(httpContext_, resolver)
: std::make_shared<http::Request>(httpContext_, resolver, target);
if (serverCertificate_)
request->set_certificate_authority(serverCertificate_);
if (clientIdentity_.first and clientIdentity_.second)
request->set_identity(clientIdentity_);
request->set_header_field(restinio::http_field_t::user_agent, userAgent_);
return request;
}
void
DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb, time_point /*created*/, bool permanent)
{
if (logger_)
logger_->debug("[proxy:client] [put] [search {}] executing for {}", key, val->toString());
try {
auto request = buildRequest("/key/" + key.toString());
auto reqid = request->id();
request->set_method(restinio::http_method_post());
setHeaderFields(*request);
auto json = val->toJson();
if (permanent) {
if (deviceKey_.empty()) {
json["permanent"] = true;
} else {
#ifdef OPENDHT_PUSH_NOTIFICATIONS
Json::Value refresh;
getPushRequest(refresh);
json["permanent"] = refresh;
#else
json["permanent"] = true;
#endif
}
}
request->set_body(Json::writeString(jsonBuilder_, json));
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)) {
auto id = dht::Value(parsedValue).id;
val->id = id;
if (permanent) {
std::lock_guard lock(searchLock_);
auto& search = searches_[key];
auto it = search.pendingPuts.find(val);
if (it != search.pendingPuts.end()) {
auto sok = std::make_shared<std::atomic_bool>(ok);
auto refreshPutTimer = std::make_unique<asio::system_timer>(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));
search.pendingPuts.erase(it);
}
}
} else {
if (logger_)
logger_->error("[proxy:client] [put] failed to parse value from server: {}", err);
}
}
} else {
if (logger_)
logger_->error("[proxy:client] [put] failed with code={:d}", response.status_code);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (cb)
cb(ok);
if (not isDestroying_) {
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [put {}] error: {}", key, e.what());
opFailed();
}
}
/**
* Get data currently being put at the given hash.
*/
std::vector<Sp<Value>>
DhtProxyClient::getPut(const InfoHash& key) const
{
std::vector<Sp<Value>> ret;
auto search = searches_.find(key);
if (search != searches_.end()) {
ret.reserve(search->second.puts.size());
for (const auto& put : search->second.puts)
ret.emplace_back(put.second.value);
}
return ret;
}
/**
* Get data currently being put at the given hash with the given id.
*/
Sp<Value>
DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) const
{
auto search = searches_.find(key);
if (search == searches_.end())
return {};
auto val = search->second.puts.find(id);
if (val == search->second.puts.end())
return {};
return val->second.value;
}
/**
* Stop any put/announce operation at the given location,
* for the value with the given id.
*/
bool
DhtProxyClient::cancelPut(const InfoHash& key, const Value::Id& id)
{
auto search = searches_.find(key);
if (search == searches_.end())
return false;
if (logger_)
logger_->debug("[proxy:client] [put] [search {}] cancel", key.to_view());
return search->second.puts.erase(id) > 0;
}
NodeStats
DhtProxyClient::getNodesStats(sa_family_t af) const
{
return af == AF_INET ? stats4_ : stats6_;
}
void
DhtProxyClient::getProxyInfos()
{
if (logger_)
logger_->debug("[proxy:client] [info] requesting proxy server node information");
auto infoState = std::make_shared<InfoState>();
{
std::lock_guard l(lockCurrentProxyInfos_);
if (infoState_)
infoState_->cancel = true;
infoState_ = infoState;
if (statusIpv4_ == NodeStatus::Disconnected)
statusIpv4_ = NodeStatus::Connecting;
if (statusIpv6_ == NodeStatus::Disconnected)
statusIpv6_ = NodeStatus::Connecting;
}
if (logger_)
logger_->debug("[proxy:client] [status] sending request");
auto resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
queryProxyInfo(infoState, resolver, AF_INET);
queryProxyInfo(infoState, resolver, AF_INET6);
std::lock_guard l(resolverLock_);
resolver_ = resolver;
}
void
DhtProxyClient::queryProxyInfo(const Sp<InfoState>& infoState, const Sp<http::Resolver>& resolver, sa_family_t family)
{
if (logger_)
logger_->debug("[proxy:client] [status] query ipv{} info", family == AF_INET ? 4 : 6);
try {
auto request = std::make_shared<http::Request>(httpContext_, resolver, family);
if (serverCertificate_)
request->set_certificate_authority(serverCertificate_);
auto reqid = request->id();
request->set_method(restinio::http_method_get());
setHeaderFields(*request);
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_->error("[proxy:client] [status] ipv{} failed with 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);
} 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);
} else if (not infoState->cancel) {
if (auto req = response.request.lock()) {
if (auto conn = req->get_connection()) {
const auto& localAddr = conn->local_address();
proxyInfos["local_ip"] = localAddr.to_string();
}
}
onProxyInfos(proxyInfos, family);
}
}
if (not isDestroying_) {
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
if (infoState->cancel.load())
return;
{
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [status] error sending request: {}", e.what());
}
}
void
DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t family)
{
if (isDestroying_)
return;
std::unique_lock l(lockCurrentProxyInfos_);
auto oldStatus = std::max(statusIpv4_, statusIpv6_);
auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_;
auto& pubAddress = family == AF_INET ? publicAddressV4_ : publicAddressV6_;
auto& localAddress = family == AF_INET ? localAddrv4_ : localAddrv6_;
if (not proxyInfos.isMember("node_id")) {
if (logger_)
logger_->error("[proxy:client] [info] request failed for {}", family == AF_INET ? "ipv4" : "ipv6");
status = NodeStatus::Disconnected;
if (pubAddress) {
pubAddress = {};
}
} else {
if (logger_)
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"]);
stats6_ = NodeStats(proxyInfos["ipv6"]);
auto publicIp = parsePublicAddress(proxyInfos["public_ip"]);
bool pub = pubAddress != publicIp;
pubAddress = publicIp;
if (proxyInfos.isMember("local_ip")) {
std::string localIp = proxyInfos["local_ip"].asString();
if (localAddress.toString() != localIp) {
localAddress.setAddress(localIp.c_str());
pub = (bool) localAddress;
}
}
if (pub && publicAddressChangedCb_) {
std::vector<SockAddr> addresses;
if (publicAddressV4_)
addresses.emplace_back(publicAddressV4_);
if (publicAddressV6_)
addresses.emplace_back(publicAddressV6_);
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back(
[cb = publicAddressChangedCb_, addresses = std::move(addresses)]() { cb(std::move(addresses)); });
}
if (stats4_.good_nodes + stats6_.good_nodes)
status = NodeStatus::Connected;
else if (stats4_.dubious_nodes + stats6_.dubious_nodes)
status = NodeStatus::Connecting;
else
status = NodeStatus::Disconnected;
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [info] error processing: {}", e.what());
}
}
auto newStatus = std::max(statusIpv4_, statusIpv6_);
if (newStatus == NodeStatus::Connected) {
if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting || launchConnectedCbs_) {
launchConnectedCbs_ = false;
listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
if (not onConnectCallbacks_.empty()) {
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back([cbs = std::move(onConnectCallbacks_)]() mutable {
while (not cbs.empty()) {
cbs.front()();
cbs.pop();
}
});
}
}
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_->expires_at(std::chrono::steady_clock::now() + std::chrono::minutes(1));
nextProxyConfirmationTimer_->async_wait(
std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
}
l.unlock();
loopSignal_();
}
SockAddr
DhtProxyClient::parsePublicAddress(const Json::Value& val)
{
auto public_ip = val.asString();
auto [host, service] = splitPort(public_ip);
auto sa = SockAddr::resolve(std::string(host));
if (sa.empty())
return {};
return sa.front().getMappedIPv4();
}
std::vector<SockAddr>
DhtProxyClient::getPublicAddress(sa_family_t family)
{
std::lock_guard l(lockCurrentProxyInfos_);
std::vector<SockAddr> result;
if (publicAddressV6_ && family != AF_INET)
result.emplace_back(publicAddressV6_);
if (publicAddressV4_ && family != AF_INET6)
result.emplace_back(publicAddressV4_);
return result;
}
size_t
DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filter, Where where)
{
if (logger_)
logger_->debug("[proxy:client] [listen] [search {}]", key.to_view());
if (isDestroying_)
return 0;
std::lock_guard lock(searchLock_);
auto& search = searches_[key];
auto query = std::make_shared<Query>(Select {}, std::move(where));
return search.ops
.listen(cb, query, std::move(filter), [this, key](Sp<Query>, ValueCallback cb, SyncCallback) -> size_t {
// Find search
auto search = searches_.find(key);
if (search == searches_.end()) {
if (logger_)
logger_->error("[proxy:client] [listen] [search {}] search not found", key.to_view());
return 0;
}
if (logger_)
logger_->debug("[proxy:client] [listen] [search {}] sending {}",
key.to_view(),
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<OperationState>();
l->second.opstate = opstate;
l->second.cb = [this, key, token, opstate](const std::vector<Sp<Value>>& 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<asio::system_timer>(httpContext_);
l->second.refreshSubscriberTimer->expires_at(std::chrono::system_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(fmt::format("/key/{}/listen", key.to_view()));
} else {
method = ListenMethod::SUBSCRIBE;
header.method(restinio::http_method_subscribe());
header.request_target(fmt::format("/key/{}", key.to_view()));
}
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<OperationState> opstate)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->error("[proxy:client] [resubscribe {}] {}", key.to_view(), ec.message());
return;
}
if (opstate->stop)
return;
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()) {
resubscribe(key, token, l->second);
} else {
if (logger_)
logger_->error("[proxy:client] [resubscribe {}] token not found", key.to_view());
}
}
}
bool
DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken)
{
if (logger_)
logger_->debug("[proxy:client] [search {}] cancel listen {}", key.to_view(), gtoken);
std::lock_guard lock(searchLock_);
// find the listener in cache
auto it = searches_.find(key);
if (it == searches_.end())
return false;
auto& ops = it->second.ops;
bool canceled = ops.cancelListen(gtoken, std::chrono::steady_clock::now());
// define real cancel listen only once
if (not it->second.opExpirationTimer)
it->second.opExpirationTimer = std::make_unique<asio::steady_timer>(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));
return canceled;
}
void
DhtProxyClient::handleExpireListener(const asio::error_code& ec, const InfoHash& key)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->error("[proxy:client] [listen {}] error in cancel: {}", key.to_view(), ec.message());
return;
}
if (logger_)
logger_->debug("[proxy:client] [listen {}] expire listener", key.to_view());
std::lock_guard lock(searchLock_);
auto search = searches_.find(key);
if (search == searches_.end())
return;
// everytime a new expiry is set, a previous gets aborted
time_point next = search->second.ops.expire(std::chrono::steady_clock::now(), [&](size_t ltoken) {
auto it = search->second.listeners.find(ltoken);
if (it == search->second.listeners.end())
return;
auto& listener = it->second;
listener.opstate->stop = true;
if (not deviceKey_.empty()) {
// UNSUBSCRIBE
auto request = buildRequest("/key/" + key.toString());
auto reqid = request->id();
try {
request->set_method(restinio::http_method_unsubscribe());
setHeaderFields(*request);
Json::Value body;
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) {
if (response.status_code != 200) {
if (logger_)
logger_->error("[proxy:client] [unsubscribe {}] failed with code={}",
key.to_view(),
response.status_code);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (not isDestroying_) {
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [unsubscribe {}] failed: {}", key.to_view(), e.what());
}
} else {
// stop the request
if (listener.request) {
listener.request->cancel();
listener.request.reset();
}
}
search->second.listeners.erase(it);
if (logger_)
logger_->debug("[proxy:client] [listen:cancel] [search {}] {} listener remaining",
key.to_view(),
search->second.listeners.size());
});
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));
}
if (search->second.listeners.empty()) {
searches_.erase(search);
}
}
void
DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
const CacheValueCallback& cb,
const Sp<OperationState>& opstate,
Listener& listener,
ListenMethod method)
{
if (logger_)
logger_->error("[proxy:client] [listen] sendListen: {}", (int) method);
try {
auto request = buildRequest();
listener.request = request;
auto reqid = request->id();
request->set_header(header);
setHeaderFields(*request);
if (method == ListenMethod::LISTEN)
request->set_connection_type(restinio::http_connection_header_t::keep_alive);
#ifdef OPENDHT_PUSH_NOTIFICATIONS
std::string body;
if (method != ListenMethod::LISTEN)
body = fillBody(method == ListenMethod::RESUBSCRIBE);
request->set_body(body);
#endif
auto rxBuf = std::make_shared<LineSplit>();
request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length) {
try {
auto& b = *rxBuf;
b.append(at, length);
// one value per body line
while (b.getLine('\n') and !opstate->stop) {
std::string err;
Json::Value json;
const auto& line = b.line();
if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)) {
opstate->ok.store(false);
return;
}
if (json.size() == 0) { // it's the end
break;
}
auto value = std::make_shared<Value>(json);
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()))
opstate->stop.store(true);
});
}
loopSignal_();
}
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [listen] request #{} error in parsing: {}", reqid, e.what());
opstate->ok.store(false);
}
});
request->add_on_done_callback([this, opstate, reqid](const http::Response& response) {
if (response.status_code != 200) {
if (logger_)
logger_->error("[proxy:client] [listen] send request #{} failed with code={}",
reqid,
response.status_code);
opstate->ok.store(false);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (not isDestroying_) {
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
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) {
if (logger_)
logger_->error("[proxy:client] [listen] request failed: {}", e.what());
}
}
void
DhtProxyClient::opFailed()
{
if (isDestroying_)
return;
if (logger_)
logger_->error("[proxy:client] proxy request failed");
{
std::lock_guard l(lockCurrentProxyInfos_);
statusIpv4_ = NodeStatus::Disconnected;
statusIpv6_ = NodeStatus::Disconnected;
}
getConnectivityStatus();
loopSignal_();
}
void
DhtProxyClient::getConnectivityStatus()
{
if (logger_)
logger_->d("[proxy:client] [connectivity] get status");
if (!isDestroying_)
getProxyInfos();
}
void
DhtProxyClient::restartListeners(const asio::error_code& ec)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->e("[proxy:client] restart error: %s", ec.message().c_str());
return;
}
if (isDestroying_)
return;
if (logger_)
logger_->d("[proxy:client] [listeners] refresh permanent puts");
std::lock_guard lock(searchLock_);
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);
if (!put.second.refreshPutTimer) {
put.second.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_);
}
put.second.refreshPutTimer->expires_at(std::chrono::system_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) {
doPut(key, pput, {}, time_point::max(), true);
}
}
if (not deviceKey_.empty()) {
if (logger_)
logger_->debug("[proxy:client] [listeners] resubscribe due to a connectivity change");
// Connectivity changed, refresh all subscribe
for (auto& search : searches_)
for (auto& listener : search.second.listeners)
if (!listener.second.opstate->ok)
resubscribe(search.first, listener.first, listener.second);
return;
}
if (logger_)
logger_->debug("[proxy:client] [listeners] restarting listeners");
for (auto& search : searches_) {
for (auto& l : search.second.listeners) {
auto& listener = l.second;
if (auto opstate = listener.opstate)
opstate->stop = true;
listener.request->cancel();
listener.request.reset();
}
}
for (auto& search : searches_) {
for (auto& l : search.second.listeners) {
auto& listener = l.second;
auto opstate = listener.opstate;
// Redo listen
opstate->stop.store(false);
opstate->ok.store(true);
auto cb = listener.cb;
// define header
restinio::http_request_header_t header;
header.method(restinio::http_method_get());
header.request_target("/key/" + search.first.toString() + "/listen");
sendListen(header, cb, opstate, listener, ListenMethod::LISTEN);
}
}
}
PushNotificationResult
DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::string, std::string>& notification)
{
auto ret = PushNotificationResult::IgnoredNoOp;
#ifdef OPENDHT_PUSH_NOTIFICATIONS
{
// If a push notification is received, the proxy is up and running
std::lock_guard l(lockCurrentProxyInfos_);
auto oldStatus = std::max(statusIpv4_, statusIpv6_);
if (oldStatus != NodeStatus::Connected)
launchConnectedCbs_ = true;
statusIpv4_ = NodeStatus::Connected;
statusIpv6_ = NodeStatus::Connected;
}
auto launchLoop = false;
try {
auto sessionId = notification.find("s");
if (sessionId != notification.end() and sessionId->second != pushSessionId_) {
if (logger_)
logger_->debug("[proxy:client] [push] ignoring push for other session");
return PushNotificationResult::IgnoredWrongSession;
}
std::lock_guard lock(searchLock_);
auto timeout = notification.find("timeout");
if (timeout != notification.cend()) {
InfoHash key(timeout->second);
auto& search = searches_.at(key);
auto vidIt = notification.find("vid");
if (vidIt != notification.end()) {
// Refresh put
auto vid = std::stoull(vidIt->second);
auto& put = search.puts.at(vid);
if (!put.refreshPutTimer)
put.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
std::chrono::system_clock::now());
else
put.refreshPutTimer->expires_at(std::chrono::system_clock::now());
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;
}
} else {
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 (...) {
}
auto key = InfoHash(notification.at("key"));
auto& search = searches_.at(key);
auto expired = notification.find("exp");
for (auto& list : search.listeners) {
if (list.second.opstate->stop)
continue;
if (logger_)
logger_->debug("[proxy:client] [push] [search {}] received", key);
auto token = list.first;
auto opstate = list.second.opstate;
if (expired == notification.end()) {
auto cb = list.second.cb;
auto oldValues = list.second.cache.getValues();
auto receivedIds = std::make_shared<std::set<Value::Id>>();
get(
key,
[cb, sendTime, receivedIds](const std::vector<Sp<Value>>& vals) {
for (const auto& v : vals)
receivedIds->insert(v->id);
return cb(vals, false, sendTime);
},
[cb, oldValues, sendTime, receivedIds](bool ok) {
if (!ok)
return;
// Decrement old values refcount to expire values not
// present in the new list
std::vector<Sp<Value>> toExpire;
for (const auto& v : oldValues)
if (receivedIds->find(v->id) == receivedIds->end())
toExpire.emplace_back(v);
if (!toExpire.empty())
cb(toExpire, true, sendTime);
});
ret = PushNotificationResult::Values;
} else {
std::stringstream ss(expired->second);
std::vector<Value::Id> ids;
while (ss.good()) {
std::string substr;
getline(ss, substr, ',');
ids.emplace_back(std::stoull(substr));
}
{
std::lock_guard lockCb(lockCallbacks_);
callbacks_.emplace_back([this, key, token, opstate, ids, sendTime]() {
if (opstate->stop)
return;
std::lock_guard lock(searchLock_);
auto s = searches_.find(key);
if (s == searches_.end())
return;
auto l = s->second.listeners.find(token);
if (l == s->second.listeners.end())
return;
if (not opstate->stop and not l->second.cache.onValuesExpired(ids, sendTime))
opstate->stop = true;
});
}
launchLoop = true;
ret = PushNotificationResult::ValuesExpired;
}
}
}
} catch (const std::out_of_range& e) {
ret = PushNotificationResult::IgnoredNoOp;
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:client] [push] receive error: {}", e.what());
ret = PushNotificationResult::Error;
}
if (launchLoop)
loopSignal_();
#endif
return ret;
}
void
DhtProxyClient::resubscribe([[maybe_unused]] const InfoHash& key,
[[maybe_unused]] const size_t token,
[[maybe_unused]] Listener& listener)
{
#ifdef OPENDHT_PUSH_NOTIFICATIONS
if (deviceKey_.empty())
return;
if (logger_)
logger_->debug("[proxy:client] [resubscribe] [search {}]", key);
auto opstate = listener.opstate;
listener.request.reset(); // This will update ok to false
opstate->ok = true;
restinio::http_request_header_t header;
header.method(restinio::http_method_subscribe());
header.request_target("/key/" + key.toString());
if (!listener.refreshSubscriberTimer) {
listener.refreshSubscriberTimer = std::make_unique<asio::system_timer>(httpContext_);
}
listener.refreshSubscriberTimer->expires_at(std::chrono::system_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
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
void
DhtProxyClient::getPushRequest(Json::Value& body) const
{
body["key"] = deviceKey_;
body["client_id"] = pushClientId_;
body["session_id"] = pushSessionId_;
body["topic"] = notificationTopic_;
body["platform"] = platform_;
}
std::string
DhtProxyClient::fillBody(bool resubscribe)
{
// Fill body with
// {
// "key":"device_key",
// }
Json::Value body;
getPushRequest(body);
if (resubscribe) {
// This is the first listen, we want to retrieve previous values.
body["refresh"] = true;
}
auto content = Json::writeString(jsonBuilder_, body) + "\n";
std::replace(content.begin(), content.end(), '\n', ' ');
return content;
}
#endif // OPENDHT_PUSH_NOTIFICATIONS
void
DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token)
{
#ifdef OPENDHT_PUSH_NOTIFICATIONS
{
std::unique_lock l(lockCurrentProxyInfos_);
if (deviceKey_ == token)
return;
deviceKey_ = token;
if (!(statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected))
return;
}
if (logger_)
logger_->debug("[proxy:client] [push] token changed, resubscribing");
std::lock_guard lock(searchLock_);
for (auto& search : searches_) {
for (auto& listener : search.second.listeners) {
resubscribe(search.first, listener.first, listener.second);
}
}
#endif
}
} // namespace dht
Reference in new issue View Git Blame Copy Permalink