diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 875c2f2a9971aa904b83bd1319930d17dec8b30f..bb1abc874b349c694a42018bf947448a60b0754c 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -716,6 +716,22 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi loopSignal_(); return true; }; + + if (not deviceKey_.empty()) + // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason) + l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] { + if (state->cancel) + return; + std::lock_guard<std::mutex> 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, l->second); + } + } + }); + auto pushNotifToken = std::make_shared<proxy::ListenToken>(0); auto vcb = l->second.cb; l->second.pushNotifToken = pushNotifToken; @@ -1009,6 +1025,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener) { #if OPENDHT_PUSH_NOTIFICATIONS if (deviceKey_.empty()) return; + scheduler.syncTime(); DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str()); // Subscribe restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString()); @@ -1023,6 +1040,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener) state->ok = true; listener.req = req; listener.pushNotifToken = pushNotifToken; + scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); listener.thread = std::thread([this, req, pushNotifToken, state]() { fillBodyToGetToken(req); auto settings = std::make_shared<restbed::Settings>();