TEST_P(TcpTunnelingIntegrationTest, TcpProxyDownstreamFlush) {
  // Use a very large size to make sure it is larger than the kernel socket read buffer.
  const uint32_t size = 50 * 1024 * 1024;
  config_helper_.setBufferLimits(size / 4, size / 4);
  initialize();

  setUpConnection(fake_upstream_connection_);

  tcp_client_->readDisable(true);
  std::string data(size, 'a');
  if (upstreamProtocol() == Http::CodecType::HTTP1) {
    ASSERT_TRUE(tcp_client_->write("hello", false));
    ASSERT_TRUE(upstream_request_->waitForData(*dispatcher_, 5));

    upstream_request_->encodeData(data, true);
    ASSERT_TRUE(fake_upstream_connection_->close());
  } else {
    ASSERT_TRUE(tcp_client_->write("", true));

    // This ensures that readDisable(true) has been run on its thread
    // before tcp_client_ starts writing.
    ASSERT_TRUE(upstream_request_->waitForEndStream(*dispatcher_));

    upstream_request_->encodeData(data, true);
  }

  test_server_->waitForCounterGe("cluster.cluster_0.upstream_flow_control_paused_reading_total", 1);
  tcp_client_->readDisable(false);
  tcp_client_->waitForData(data);
  tcp_client_->waitForHalfClose();
  if (upstreamProtocol() == Http::CodecType::HTTP1) {
    tcp_client_->close();
  }
}
TEST_P(TcpTunnelingIntegrationTest, TcpProxyDownstreamFlush) {
   
  const uint32_t size = 50 * 1024 * 1024;
  config_helper_.setBufferLimits(size / 4, size / 4);
  initialize();

  setUpConnection(fake_upstream_connection_);

  tcp_client_->readDisable(true);
  std::string data(size, 'a');
  if (upstreamProtocol() == Http::CodecType::HTTP1) {
    ASSERT_TRUE(tcp_client_->write("hello", false));
    ASSERT_TRUE(upstream_request_->waitForData(*dispatcher_, 5));

    upstream_request_->encodeData(data, true);
    ASSERT_TRUE(fake_upstream_connection_->close());
  } else {
    ASSERT_TRUE(tcp_client_->write("", true));

     
     
    ASSERT_TRUE(upstream_request_->waitForEndStream(*dispatcher_));

    upstream_request_->encodeData(data, true);
  }

  test_server_->waitForCounterGe("cluster.cluster_0.upstream_flow_control_paused_reading_total", 1);
  tcp_client_->readDisable(false);
  tcp_client_->waitForData(data);
  tcp_client_->waitForHalfClose();
  if (upstreamProtocol() == Http::CodecType::HTTP1) {
    tcp_client_->close();
  }
}