lazy read disable for the http1 codec

source/common/http/http1/codec_impl.cc

@@ -1143,13 +1143,38 @@ void ServerConnectionImpl::onBody(Buffer::Instance& data) {
   }
 }
 
+Http::Status ServerConnectionImpl::dispatch(Buffer::Instance& data) {
+  if (active_request_ != nullptr && active_request_->remote_complete_) {
+    // Active downstream request remote complete but there is some new data comming then try to
+    // disable the connection reading. Connection reading will be re-enabled after the current
+    // active downstream request has completed.
+    // This ensures that the new comming data can be consumed after the current active downstream
+    // request has completed.
+    active_request_->response_encoder_.readDisable(true);
+    return okStatus();
+  }
+
+  Http::Status status = ConnectionImpl::dispatch(data);
+
+  if (active_request_ != nullptr && active_request_->remote_complete_) {
+    // Active downstream request remote complete but there is some remaining data in the read buffer
+    // then try to disable the connection reading. Connection reading will be re-enabled after the
+    // current active downstream request has completed.
+    // This ensures that the remaining data can be consumed after the current active downstream
+    // request has completed.
+    if (data.length() > 0) {
+      active_request_->response_encoder_.readDisable(true);
+    }
+  }
+  return status;
+}
+
 ParserStatus ServerConnectionImpl::onMessageCompleteBase() {
   ASSERT(!handling_upgrade_);
   if (active_request_) {
 
     // The request_decoder should be non-null after we've called the newStream on callbacks.
     ASSERT(active_request_->request_decoder_);
-    active_request_->response_encoder_.readDisable(true);
     active_request_->remote_complete_ = true;
 
     if (deferred_end_stream_headers_) {

source/common/http/http1/codec_impl.h

@@ -81,8 +81,6 @@ class StreamEncoderImpl : public virtual StreamEncoder,
   void setIsResponseToConnectRequest(bool value) { is_response_to_connect_request_ = value; }
   void setDetails(absl::string_view details) { details_ = details; }
 
-  void clearReadDisableCallsForTests() { read_disable_calls_ = 0; }
-
   const StreamInfo::BytesMeterSharedPtr& bytesMeter() override { return bytes_meter_; }
 
 protected:
@@ -477,6 +475,7 @@ class ServerConnectionImpl : public ServerConnection, public ConnectionImpl {
   Status onUrl(const char* data, size_t length) override;
   Status onStatus(const char*, size_t) override { return okStatus(); }
   // ConnectionImpl
+  Http::Status dispatch(Buffer::Instance& data) override;
   void onEncodeComplete() override;
   StreamInfo::BytesMeter& getBytesMeter() override {
     if (active_request_) {

test/integration/fake_upstream.cc

@@ -298,29 +298,9 @@ void FakeStream::finishGrpcStream(Grpc::Status::GrpcStatus status) {
       {"grpc-status", std::to_string(static_cast<uint32_t>(status))}});
 }
 
-// The TestHttp1ServerConnectionImpl outlives its underlying Network::Connection
-// so must not access the Connection on teardown. To achieve this, clear the
-// read disable calls to avoid checking / editing the Connection blocked state.
 class TestHttp1ServerConnectionImpl : public Http::Http1::ServerConnectionImpl {
 public:
   using Http::Http1::ServerConnectionImpl::ServerConnectionImpl;
-
-  Http::Http1::ParserStatus onMessageCompleteBase() override {
-    auto rc = ServerConnectionImpl::onMessageCompleteBase();
-
-    if (activeRequest() && activeRequest()->request_decoder_) {
-      // Undo the read disable from the base class - we have many tests which
-      // waitForDisconnect after a full request has been read which will not
-      // receive the disconnect if reading is disabled.
-      activeRequest()->response_encoder_.readDisable(false);
-    }
-    return rc;
-  }
-  ~TestHttp1ServerConnectionImpl() override {
-    if (activeRequest()) {
-      activeRequest()->response_encoder_.clearReadDisableCallsForTests();
-    }
-  }
 };
 
 class TestHttp2ServerConnectionImpl : public Http::Http2::ServerConnectionImpl {