Merge pull request #1496 from majestrate/rpc-kill-endpoint-sessions-2020-11-28

add llarp.endpoint rpc command

llarp/rpc/rpc_server.cpp

@@ -118,6 +118,62 @@ namespace llarp::rpc
                 defer.reply(data);
               });
             })
+        .add_request_command(
+            "endpoint",
+            [&](lokimq::Message& msg) {
+              HandleJSONRequest(msg, [r = m_Router](nlohmann::json obj, ReplyFunction_t reply) {
+                if (r->IsServiceNode())
+                {
+                  reply(CreateJSONError("not supported"));
+                  return;
+                }
+                std::string endpoint = "default";
+                std::unordered_set<service::Address> kills;
+                {
+                  const auto itr = obj.find("endpoint");
+                  if (itr != obj.end())
+                    endpoint = itr->get<std::string>();
+                }
+                {
+                  const auto itr = obj.find("kill");
+                  if (itr != obj.end())
+                  {
+                    if (itr->is_array())
+                    {
+                      for (auto kill_itr = itr->begin(); kill_itr != itr->end(); ++kill_itr)
+                      {
+                        if (kill_itr->is_string())
+                          kills.emplace(kill_itr->get<std::string>());
+                      }
+                    }
+                    else if (itr->is_string())
+                    {
+                      kills.emplace(itr->get<std::string>());
+                    }
+                  }
+                }
+                if (kills.empty())
+                {
+                  reply(CreateJSONError("no action taken"));
+                  return;
+                }
+                LogicCall(r->logic(), [r, endpoint, kills, reply]() {
+                  auto ep = r->hiddenServiceContext().GetEndpointByName(endpoint);
+                  if (ep == nullptr)
+                  {
+                    reply(CreateJSONError("no endpoint with name " + endpoint));
+                    return;
+                  }
+                  std::size_t removed = 0;
+                  for (auto kill : kills)
+                  {
+                    removed += ep->RemoveAllConvoTagsFor(std::move(kill));
+                  }
+                  reply(CreateJSONResponse(
+                      "removed " + std::to_string(removed) + " flow" + (removed == 1 ? "" : "s")));
+                });
+              });
+            })
         .add_request_command(
             "exit",
             [&](oxenmq::Message& msg) {

llarp/service/address.hpp

@@ -104,4 +104,17 @@ namespace llarp
   }  // namespace service
 }  // namespace llarp
 
+namespace std
+{
+  template <>
+  struct hash<llarp::service::Address>
+  {
+    size_t
+    operator()(const llarp::service::Address& addr) const
+    {
+      return llarp::service::Address::Hash{}(addr);
+    }
+  };
+}  // namespace std
+
 #endif

llarp/service/endpoint.cpp

@@ -345,6 +345,25 @@ namespace llarp
       itr->second.lastUsed = Now();
     }
 
+    size_t
+    Endpoint::RemoveAllConvoTagsFor(service::Address remote)
+    {
+      size_t removed = 0;
+      auto& sessions = Sessions();
+      auto itr = sessions.begin();
+      while (itr != sessions.end())
+      {
+        if (itr->second.remote.Addr() == remote)
+        {
+          itr = sessions.erase(itr);
+          removed++;
+        }
+        else
+          ++itr;
+      }
+      return removed;
+    }
+
     bool
     Endpoint::GetSenderFor(const ConvoTag& tag, ServiceInfo& si) const
     {

llarp/service/endpoint.hpp

@@ -273,6 +273,9 @@ namespace llarp
 
       using PendingBufferQueue = std::deque<PendingBuffer>;
 
+      size_t
+      RemoveAllConvoTagsFor(service::Address remote);
+
       bool
       WantsOutboundSession(const Address&) const override;