diff --git a/contract-tests/sse-contract-tests/include/entity_manager.hpp b/contract-tests/sse-contract-tests/include/entity_manager.hpp index 93a35f0b6..eb6f56170 100644 --- a/contract-tests/sse-contract-tests/include/entity_manager.hpp +++ b/contract-tests/sse-contract-tests/include/entity_manager.hpp @@ -7,10 +7,12 @@ #include +#include #include #include #include #include +#include #include class EventOutbox; @@ -18,7 +20,9 @@ class EventOutbox; class EntityManager { using Inbox = std::shared_ptr; using Outbox = std::shared_ptr; - using Entity = std::pair; + // Third element: suppress-next-error flag for caller-triggered restarts + // (the canceled read surfaces as an error on the read path). + using Entity = std::tuple>; std::unordered_map entities_; @@ -49,4 +53,15 @@ class EntityManager { * @return True if the entity was found and destroyed. */ bool destroy(std::string const& id); + /** + * Send a command to an entity (see the SSE contract test protocol). + * Currently supports "restart", which makes the client disconnect and + * reconnect immediately. + * @param id ID of the entity. + * @param command The command name. + * @return nullopt if no such entity exists; otherwise true if the + * command was recognized. + */ + std::optional command(std::string const& id, + std::string const& command); }; diff --git a/contract-tests/sse-contract-tests/src/entity_manager.cpp b/contract-tests/sse-contract-tests/src/entity_manager.cpp index 798d55ce5..0e32316da 100644 --- a/contract-tests/sse-contract-tests/src/entity_manager.cpp +++ b/contract-tests/sse-contract-tests/src/entity_manager.cpp @@ -44,13 +44,23 @@ std::optional EntityManager::create(ConfigParams const& params) { LD_LOG(logger_, LogLevel::kDebug) << std::move(msg); }); + auto suppress_next_error = std::make_shared(false); + client_builder.receiver([copy = poster](launchdarkly::sse::Event event) { copy->post_event(std::move(event)); }); - client_builder.errors([copy = poster](launchdarkly::sse::Error event) { - copy->post_error(event); - }); + client_builder.errors( + [copy = poster, suppress = suppress_next_error](launchdarkly::sse::Error event) { + // A caller-triggered restart surfaces as an error on the read + // path (the canceled read looks like a body timeout); like other + // services (e.g. okhttp's StreamClosedByCallerException + // suppression) we don't report it to the harness. + if (suppress->exchange(false)) { + return; + } + copy->post_error(event); + }); auto client = client_builder.build(); if (!client) { @@ -61,18 +71,34 @@ std::optional EntityManager::create(ConfigParams const& params) { client->async_connect(); - entities_.emplace(id, std::make_pair(client, poster)); + entities_.emplace(id, std::make_tuple(client, poster, suppress_next_error)); return id; } +std::optional EntityManager::command(std::string const& id, + std::string const& command) { + auto it = entities_.find(id); + if (it == entities_.end()) { + return std::nullopt; + } + + if (command == "restart") { + std::get<2>(it->second)->store(true); + std::get<0>(it->second)->async_restart("test harness restart command"); + return true; + } + + return false; +} + bool EntityManager::destroy(std::string const& id) { auto it = entities_.find(id); if (it == entities_.end()) { return false; } - it->second.first->async_shutdown(nullptr); - it->second.second->stop(); + std::get<0>(it->second)->async_shutdown(nullptr); + std::get<1>(it->second)->stop(); entities_.erase(it); diff --git a/contract-tests/sse-contract-tests/src/main.cpp b/contract-tests/sse-contract-tests/src/main.cpp index 0a9881466..664c46acc 100644 --- a/contract-tests/sse-contract-tests/src/main.cpp +++ b/contract-tests/sse-contract-tests/src/main.cpp @@ -37,7 +37,8 @@ int main(int argc, char* argv[]) { srv.add_capability("comments"); srv.add_capability("report"); srv.add_capability("post"); - srv.add_capability("reconnection"); + srv.add_capability("restart"); + srv.add_capability("server-directed-shutdown-request"); srv.add_capability("read-timeout"); net::signal_set signals{ioc, SIGINT, SIGTERM}; diff --git a/contract-tests/sse-contract-tests/src/session.cpp b/contract-tests/sse-contract-tests/src/session.cpp index 471121f64..e2da931f8 100644 --- a/contract-tests/sse-contract-tests/src/session.cpp +++ b/contract-tests/sse-contract-tests/src/session.cpp @@ -101,6 +101,33 @@ std::optional Session::generate_response(Request& req) { } } + if (req.method() == http::verb::post && + req.target().starts_with(kEntityPath)) { + std::string entity_id = req.target(); + boost::erase_first(entity_id, kEntityPath); + + std::string command; + try { + auto json = nlohmann::json::parse(req.body()); + command = json.at("command").get(); + } catch (nlohmann::json::exception& e) { + return bad_request("unable to parse command JSON"); + } + + auto handled = manager_.command(entity_id, command); + if (!handled) { + return not_found(req.target()); + } + if (!*handled) { + return bad_request("unrecognized command"); + } + + Response res{http::status::ok, req.version()}; + res.keep_alive(req.keep_alive()); + res.prepare_payload(); + return res; + } + if (req.method() == http::verb::delete_ && req.target().starts_with(kEntityPath)) { std::string entity_id = req.target();