dhtproxy: fix listener destruction thus, restart

include/opendht/http.h

@@ -150,6 +150,7 @@ public:
         SENDING,
         HEADER_RECEIVED,
         RECEIVING,
+        FINISHING,
         DONE
     };
     using OnStatusCb = std::function<void(unsigned int status_code)>;
@@ -184,7 +185,6 @@ public:
     void add_on_state_change_callback(OnStateChangeCb cb);
 
     void send();
-    void end();
 
 private:
     struct Callbacks {
@@ -224,6 +224,7 @@ private:
 
     std::mutex cbs_mutex_;
     std::unique_ptr<Callbacks> cbs_;
+    State state_;
 
     std::string service_;
     std::string host_;

src/dht_proxy_client.cpp

@@ -1101,7 +1101,7 @@ DhtProxyClient::restartListeners()
             auto& listener = l.second;
             if (auto opstate = listener.opstate)
                 opstate->stop = true;
-            listener.request->end();
+            listener.request.reset();
         }
     }
     for (auto& search: searches_) {
@@ -1222,8 +1222,8 @@ DhtProxyClient::resubscribe(const InfoHash& key, const size_t token, Listener& l
 
     auto opstate = listener.opstate;
     opstate->stop = true;
-    if (listener.request) {
-        listener.request->end();
+    if (listener.request){
+        listener.request.reset();
     }
     opstate.reset(new OperationState());
 

src/http.cpp

@@ -276,12 +276,7 @@ Request::Request(asio::io_context& ctx, asio::ip::basic_resolver_results<asio::i
 
 Request::~Request()
 {
-}
-
-void
-Request::end()
-{
-    conn_.reset();
+    terminate(asio::error::eof);
 }
 
 unsigned int
@@ -390,7 +385,7 @@ Request::add_on_state_change_callback(OnStateChangeCb cb)
 void
 Request::notify_state_change(const State state)
 {
-
+    state_ = state;
     std::lock_guard<std::mutex> lock(cbs_mutex_);
     if (cbs_->on_state_change)
         cbs_->on_state_change(state, response_);
@@ -551,6 +546,14 @@ Request::post()
 void
 Request::terminate(const asio::error_code& ec)
 {
+    if (state_ == State::FINISHING or state_ == State::DONE)
+        return;
+
+    notify_state_change(State::FINISHING);
+
+    // destroy the socket with the io_context to cancel all its scheduled operations
+    conn_.reset();
+
     // set response outcome, ignore enf of file and abort
     if (!ec or ec == asio::error::eof or ec == asio::error::operation_aborted)
         response_.status_code = 200;