simplify WebTransportImpl::terminateSessionStreams

Summary: rewrites WebTransportImpl::terminateSessionStreams for simplicity

Reviewed By: joanna-jo

Differential Revision: D83162525

fbshipit-source-id: 8dfc2057e0716ff3a826fcff3ff25eb3ed745d6d

Author: hanidamlaj

proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp

@@ -495,15 +495,14 @@ TEST_F(HTTPTransactionWebTransportTest, StreamDetachWithOpenStreams) {
                        });
         readCancelled = true;
       });
-  folly::CancellationCallback writeCancel(
-      res->writeHandle->getCancelToken(), [&writeCancelled, &res, this] {
-        VLOG(4) << "writeCancelled";
-        EXPECT_CALL(transport_, resetWebTransportEgress(0, WT_APP_ERROR_2));
-        res->writeHandle->resetStream(WT_APP_ERROR_2);
-        writeCancelled = true;
-      });
+  folly::CancellationCallback writeCancel(res->writeHandle->getCancelToken(),
+                                          [&] {
+                                            VLOG(4) << "writeCancelled";
+                                            writeCancelled = true;
+                                          });
   HTTPException ex(HTTPException::Direction::INGRESS_AND_EGRESS, "aborted");
   handler_.expectError();
+  EXPECT_CALL(transport_, resetWebTransportEgress(0, _));
   EXPECT_CALL(
       transport_,
       stopReadingWebTransportIngress(0, makeOpt(WebTransport::kInternalError)));

proxygen/lib/http/webtransport/WebTransportImpl.cpp

@@ -33,36 +33,23 @@ void WebTransportImpl::destroy() {
 
 void WebTransportImpl::terminateSessionStreams(uint32_t errorCode,
                                                const std::string& reason) {
-  // These loops are dicey for possible erasure from callbacks
-  for (auto ingressStreamIt = wtIngressStreams_.begin();
-       ingressStreamIt != wtIngressStreams_.end();) {
-    auto id = ingressStreamIt->first;
-    auto& stream = ingressStreamIt->second;
-    ingressStreamIt++;
+  auto wtIngressStreams = std::move(wtIngressStreams_);
+  for (auto& [id, stream] : wtIngressStreams) {
     // Deliver an error to the application if needed
+    VLOG(4) << "aborting wt ingress id=" << id << " err=" << errorCode
+            << "; open=" << int(stream.open());
     if (stream.open()) {
-      VLOG(4) << "aborting WT ingress id=" << id << " with error=" << errorCode;
       stream.deliverReadError(WebTransport::Exception(errorCode, reason));
       stopReadingWebTransportIngress(id, errorCode);
-    } else {
-      VLOG(4) << "WT ingress already complete for id=" << id;
     }
   }
-  wtIngressStreams_.clear();
-  for (auto egressStreamIt = wtEgressStreams_.begin();
-       egressStreamIt != wtEgressStreams_.end();) {
-    auto id = egressStreamIt->first;
-    auto& stream = egressStreamIt->second;
-    egressStreamIt++;
+
+  auto wtEgressStreams = std::move(wtEgressStreams_);
+  for (auto& [id, stream] : wtEgressStreams) {
     // Deliver an error to the application
     stream.onStopSending(errorCode);
-    // The handler may have run and reset this stream, removing it from
-    // wtEgressStreams_, otherwise we have to reset it.
-    if (wtEgressStreams_.find(id) != wtEgressStreams_.end()) {
-      resetWebTransportEgress(id, errorCode);
-    }
+    resetWebTransportEgress(id, errorCode);
   }
-  wtEgressStreams_.clear();
 }
 
 void WebTransportImpl::onMaxData(uint64_t maxData) noexcept {

proxygen/lib/http/webtransport/WebTransportImpl.h

@@ -109,7 +109,6 @@ class WebTransportImpl : public WebTransport {
   ~WebTransportImpl() override = default;
 
   void destroy();
-  void onMaxData(uint64_t maxData) noexcept;
 
   // WT API
   folly::Expected<WebTransport::StreamWriteHandle*, WebTransport::ErrorCode>
@@ -379,6 +378,7 @@ class WebTransportImpl : public WebTransport {
     sendFlowController_ = FlowController(sendWindow);
     recvFlowController_ = FlowController(recvWindow);
   }
+  void onMaxData(uint64_t maxData) noexcept;
 };
 
 } // namespace proxygen