fix sse behavior when reconnecting to streams

Author: cwaldren-ld

libs/server-sent-events/src/client.cpp

@@ -79,20 +79,20 @@ class FoxyClient : public Client,
           read_timeout_(read_timeout),
           write_timeout_(write_timeout),
           req_(std::move(req)),
-          session_(std::move(executor),
-                   launchdarkly::foxy::session_opts{
-                       ToOptRef(ssl_context_),
-                       connect_timeout.value_or(kNoTimeout)}),
           backoff_(
               initial_reconnect_delay.value_or(kDefaultInitialReconnectDelay),
               kDefaultMaxBackoffDelay),
           last_event_id_(std::nullopt),
-          backoff_timer_(session_.get_executor()),
+          backoff_timer_(std::move(executor)),
           event_receiver_(std::move(receiver)),
           logger_(std::move(logger)),
           errors_(std::move(errors)),
-          last_read_(std::nullopt) {
+          last_read_(std::nullopt),
+          shutting_down_(false),
+          body_parser_(std::nullopt),
+          session_(std::nullopt) {
         create_parser();
+        create_session();
     }
 
     /** Logs a message indicating that an async_read_some operation
@@ -132,6 +132,13 @@ class FoxyClient : public Client,
         body_parser_->get().body().on_event(event_receiver_);
     }
 
+    void create_session() {
+        session_.emplace(
+            backoff_timer_.get_executor(),
+            launchdarkly::foxy::session_opts{
+                ToOptRef(ssl_context_), connect_timeout_.value_or(kNoTimeout)});
+    }
+
     /**
      * Called whenever the connection needs to be reattempted, triggering
      * a timed wait for the current backoff duration.
@@ -158,6 +165,8 @@ class FoxyClient : public Client,
         logger_(msg.str());
 
         create_parser();
+        create_session();
+
         backoff_timer_.expires_from_now(backoff_.delay());
         backoff_timer_.async_wait(beast::bind_front_handler(
             &FoxyClient::on_backoff, shared_from_this()));
@@ -171,7 +180,7 @@ class FoxyClient : public Client,
     }
 
     void run() override {
-        session_.async_connect(
+        session_->async_connect(
             host_, port_,
             beast::bind_front_handler(&FoxyClient::on_connect,
                                       shared_from_this()));
@@ -190,10 +199,10 @@ class FoxyClient : public Client,
         } else {
             req_.erase("last-event-id");
         }
-        session_.opts.timeout = write_timeout_.value_or(kNoTimeout);
-        session_.async_write(req_,
-                             beast::bind_front_handler(&FoxyClient::on_write,
-                                                       shared_from_this()));
+        session_->opts.timeout = write_timeout_.value_or(kNoTimeout);
+        session_->async_write(req_,
+                              beast::bind_front_handler(&FoxyClient::on_write,
+                                                        shared_from_this()));
     }
 
     void on_write(boost::system::error_code ec, std::size_t amount) {
@@ -205,8 +214,8 @@ class FoxyClient : public Client,
             return do_backoff(ec.what());
         }
 
-        session_.opts.timeout = read_timeout_.value_or(kNoTimeout);
-        session_.async_read_header(
+        session_->opts.timeout = read_timeout_.value_or(kNoTimeout);
+        session_->async_read_header(
             *body_parser_, beast::bind_front_handler(&FoxyClient::on_headers,
                                                      shared_from_this()));
     }
@@ -222,7 +231,7 @@ class FoxyClient : public Client,
 
         if (!body_parser_->is_header_done()) {
             /* keep reading headers */
-            return session_.async_read_header(
+            return session_->async_read_header(
                 *body_parser_,
                 beast::bind_front_handler(&FoxyClient::on_headers,
                                           shared_from_this()));
@@ -244,7 +253,7 @@ class FoxyClient : public Client,
             backoff_.succeed();
 
             last_read_ = std::chrono::steady_clock::now();
-            return session_.async_read_some(
+            return session_->async_read_some(
                 *body_parser_,
                 beast::bind_front_handler(&FoxyClient::on_read_body,
                                           shared_from_this()));
@@ -290,13 +299,13 @@ class FoxyClient : public Client,
 
     void on_read_body(boost::system::error_code ec, std::size_t amount) {
         boost::ignore_unused(amount);
-        if (ec == boost::asio::error::operation_aborted) {
-            logger_("read HTTP response body aborted");
+        if (ec == boost::asio::error::operation_aborted && shutting_down_) {
+            logger_("read HTTP response body aborted (shutting down)");
             return;
         }
         if (!ec) {
             log_and_update_last_read(amount);
-            return session_.async_read_some(
+            return session_->async_read_some(
                 *body_parser_,
                 beast::bind_front_handler(&FoxyClient::on_read_body,
                                           shared_from_this()));
@@ -305,14 +314,16 @@ class FoxyClient : public Client,
     }
 
     void async_shutdown(std::function<void()> completion) override {
-        boost::asio::post(session_.get_executor(),
+        boost::asio::post(session_->get_executor(),
                           beast::bind_front_handler(&FoxyClient::do_shutdown,
                                                     shared_from_this(),
                                                     std::move(completion)));
     }
 
     void do_shutdown(std::function<void()> completion) {
-        session_.async_shutdown(beast::bind_front_handler(
+        shutting_down_ = true;
+        backoff_timer_.cancel();
+        session_->async_shutdown(beast::bind_front_handler(
             &FoxyClient::on_shutdown, std::move(completion)));
     }
 
@@ -381,12 +392,14 @@ class FoxyClient : public Client,
     Builder::ErrorCallback errors_;
 
     std::optional<http::response_parser<body>> body_parser_;
-    launchdarkly::foxy::client_session session_;
+    std::optional<launchdarkly::foxy::client_session> session_;
     std::optional<std::string> last_event_id_;
     Backoff backoff_;
     boost::asio::steady_timer backoff_timer_;
 
     std::optional<std::chrono::steady_clock::time_point> last_read_;
+
+    bool shutting_down_;
 };
 
 Builder::Builder(net::any_io_executor ctx, std::string url)