python: support DESTDIR in pip install step

Use --root $DESTDIR when the environment variable is set so that
staged installs (e.g. 'make DESTDIR=/tmp/pkg install') place the
Python package under the staging root.

Also pass --no-build-isolation and --prefix to avoid rebuilding in an
isolated venv and to honour CMAKE_INSTALL_PREFIX.

include/opendht/proxy.h

@@ -10,14 +10,6 @@ namespace proxy {
 constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
 constexpr const std::chrono::hours OP_MARGIN {2};   // two hours
 
-/**
- * Deadline for a subscribe request to complete. Unlike a listen, a subscribe is a short
- * request: the server answers immediately and the device then relies on push notifications.
- * Nothing else observes it, so without a deadline a subscribe stalled in name resolution,
- * connection or TLS handshake would never complete nor fail.
- */
-constexpr const std::chrono::seconds SUBSCRIBE_TIMEOUT {20};
-
 using ListenToken = uint64_t;
 
 } // namespace proxy

python/CMakeLists.txt

@@ -28,8 +28,18 @@ install(CODE [[
         if(NOT _pip_check EQUAL 0)
             message(WARNING "Skipping Python package install: pip not available for ${Python3_EXECUTABLE}.")
         else()
+            set(_pip_install_args
+                install
+                --no-deps
+                --no-build-isolation
+                --break-system-packages
+                --prefix ${CMAKE_INSTALL_PREFIX})
+            if(DEFINED ENV{DESTDIR} AND NOT "$ENV{DESTDIR}" STREQUAL "")
+                list(APPEND _pip_install_args --root "$ENV{DESTDIR}")
+            endif()
+            list(APPEND _pip_install_args .)
             execute_process(
-                COMMAND ${Python3_EXECUTABLE} -m pip install --no-deps --break-system-packages .
+                COMMAND ${Python3_EXECUTABLE} -m pip ${_pip_install_args}
                 WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/python
                 OUTPUT_VARIABLE _pip_output
                 ERROR_VARIABLE _pip_error

src/dht_proxy_client.cpp

@@ -1075,12 +1075,6 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
         request->set_body(body);
 #endif
         auto rxBuf = std::make_shared<LineSplit>();
-        // Completion flag shared with the subscribe deadline armed below. The request
-        // callbacks replace each other rather than accumulating, so the flag is raised from
-        // the single on_done callback instead of a dedicated one.
-        std::shared_ptr<std::atomic_bool> subscribeDone;
-        if (method != ListenMethod::LISTEN)
-            subscribeDone = std::make_shared<std::atomic_bool>(false);
         request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length) {
             try {
                 auto& b = *rxBuf;
@@ -1119,9 +1113,7 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
                 opstate->ok.store(false);
             }
         });
-        request->add_on_done_callback([this, opstate, reqid, subscribeDone](const http::Response& response) {
-            if (subscribeDone)
-                subscribeDone->store(true);
+        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={}",
@@ -1146,29 +1138,6 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
                 request->get_connection()->set_keepalive(seconds);
             }
         });
-        if (subscribeDone) {
-            // Give the subscribe a deadline. Nothing else would ever release it: the
-            // keepalive above is only armed once the server has answered, and no other
-            // request in this client sets a timeout. A subscribe that never completes
-            // leaves the listener registered locally while the server never learns about
-            // it, so the device stops being notified without anything reporting an error.
-            // Cancelling runs the on_done callback, which clears the request, and failing
-            // the operation lets restartListeners() send the subscription again.
-            std::weak_ptr<http::Request> wreq = request;
-            request->timeout(proxy::SUBSCRIBE_TIMEOUT,
-                             [this, reqid, opstate, subscribeDone, wreq](const asio::error_code& ec) {
-                                 if (ec == asio::error::operation_aborted or subscribeDone->load()
-                                     or isDestroying_)
-                                     return;
-                                 if (logger_)
-                                     logger_->error("[proxy:client] [listen] subscribe request #{} timed out",
-                                                    reqid);
-                                 opstate->ok.store(false);
-                                 if (auto req = wreq.lock())
-                                     req->cancel();
-                                 opFailed();
-                             });
-        }
         request->send();
     } catch (const std::exception& e) {
         if (logger_)