test: moving test filters around

test/integration/BUILD

@@ -189,33 +189,6 @@ envoy_cc_test_library(
     ],
 )
 
-envoy_cc_test_library(
-    name = "pass_through_filter_lib",
-    srcs = [
-        "pass_through_filter.h",
-    ],
-    deps = [
-        "//include/envoy/http:filter_interface",
-        "//include/envoy/registry",
-    ],
-)
-
-envoy_cc_test_library(
-    name = "add_trailers_filter_config_lib",
-    srcs = [
-        "add_trailers_filter.cc",
-        "add_trailers_filter.h",
-        "add_trailers_filter_config.cc",
-        "add_trailers_filter_config.h",
-    ],
-    deps = [
-        "//include/envoy/http:filter_interface",
-        "//include/envoy/registry",
-        "//include/envoy/server:filter_config_interface",
-        "//source/extensions/filters/http/common:empty_http_filter_config_lib",
-    ],
-)
-
 envoy_cc_test_library(
     name = "http_integration_lib",
     srcs = [
@@ -225,15 +198,15 @@ envoy_cc_test_library(
         "http_integration.h",
     ],
     deps = [
-        ":add_trailers_filter_config_lib",
         ":integration_lib",
-        ":pass_through_filter_lib",
         ":test_host_predicate_lib",
         "//include/envoy/event:timer_interface",
         "//source/common/common:thread_annotations",
         "//source/extensions/filters/http/router:config",
         "//source/extensions/filters/network/http_connection_manager:config",
         "//test/common/upstream:utility_lib",
+        "//test/integration/filters:add_trailers_filter_config_lib",
+        "//test/integration/filters:pause_filter_lib",
         "//test/test_common:registry_lib",
     ],
 )

test/integration/filters/BUILD

@@ -0,0 +1,48 @@
+licenses(["notice"])  # Apache 2
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_test_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_test_library(
+    name = "add_trailers_filter_config_lib",
+    srcs = [
+        "add_trailers_filter.cc",
+    ],
+    deps = [
+        ":pass_through_filter_lib",
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//include/envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:empty_http_filter_config_lib",
+    ],
+)
+
+envoy_cc_test_library(
+    name = "pass_through_filter_lib",
+    srcs = [
+        "pass_through_filter.h",
+    ],
+    deps = [
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//source/extensions/filters/http/common:empty_http_filter_config_lib",
+    ],
+)
+
+envoy_cc_test_library(
+    name = "pause_filter_lib",
+    srcs = [
+        "pause_filter.cc",
+    ],
+    deps = [
+        ":pass_through_filter_lib",
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//source/common/network:connection_lib",
+    ],
+)

test/integration/filters/add_trailers_filter.cc

@@ -0,0 +1,50 @@
+#include <string>
+
+#include "envoy/http/filter.h"
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/http/common/empty_http_filter_config.h"
+
+#include "test/integration/filters/pass_through_filter.h"
+
+namespace Envoy {
+
+// A test filter that inserts trailers at the end of encode/decode
+class AddTrailersStreamFilter : public PassThroughFilter {
+public:
+  Http::FilterDataStatus decodeData(Buffer::Instance&, bool end_stream) {
+    if (end_stream) {
+      decoder_callbacks_->addDecodedTrailers().insertGrpcMessage().value(std::string("decode"));
+    }
+
+    return Http::FilterDataStatus::Continue;
+  }
+
+  Http::FilterDataStatus encodeData(Buffer::Instance&, bool end_stream) {
+    if (end_stream) {
+      encoder_callbacks_->addEncodedTrailers().insertGrpcMessage().value(std::string("encode"));
+    }
+
+    return Http::FilterDataStatus::Continue;
+  }
+};
+
+class AddTrailersStreamFilterConfig
+    : public Extensions::HttpFilters::Common::EmptyHttpFilterConfig {
+public:
+  AddTrailersStreamFilterConfig() : EmptyHttpFilterConfig("add-trailers-filter") {}
+
+  Http::FilterFactoryCb createFilter(const std::string&, Server::Configuration::FactoryContext&) {
+    return [](Http::FilterChainFactoryCallbacks& callbacks) -> void {
+      callbacks.addStreamFilter(std::make_shared<::Envoy::AddTrailersStreamFilter>());
+    };
+  }
+};
+
+// perform static registration
+static Registry::RegisterFactory<AddTrailersStreamFilterConfig,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy

test/integration/pass_through_filter.h → test/integration/filters/pass_through_filter.h

@@ -4,7 +4,6 @@
 
 namespace Envoy {
 
-// TODO(alyssawilk) move add_trailers_filter to use this.
 // A filter which passes all data through with Continue status.
 class PassThroughFilter : public Http::StreamFilter {
 public:

test/integration/filters/pause_filter.cc

@@ -0,0 +1,87 @@
+#include <string>
+
+#include "envoy/registry/registry.h"
+
+#include "common/network/connection_impl.h"
+
+#include "extensions/filters/http/common/empty_http_filter_config.h"
+
+#include "test/integration/filters/pass_through_filter.h"
+
+namespace Envoy {
+
+// This filter exists to synthetically test network backup by faking TCP
+// connection back-up when an encode is finished, and unblocking it when the
+// next stream starts to decode headers.
+// Allows regression tests for https://github.com/envoyproxy/envoy/issues/4541
+// TODO(alyssawilk) turn this up for more tests.
+class TestPauseFilter : public PassThroughFilter {
+public:
+  // Pass in a some global filter state to ensure the Network::Connection is
+  // blocked and unblocked exactly once.
+  TestPauseFilter(absl::Mutex& encode_lock, uint32_t& number_of_encode_calls_ref,
+                  uint32_t& number_of_decode_calls_ref)
+      : encode_lock_(encode_lock), number_of_encode_calls_ref_(number_of_encode_calls_ref),
+        number_of_decode_calls_ref_(number_of_decode_calls_ref) {}
+
+  Http::FilterDataStatus decodeData(Buffer::Instance& buf, bool end_stream) override {
+    if (end_stream) {
+      absl::WriterMutexLock m(&encode_lock_);
+      number_of_decode_calls_ref_++;
+      if (number_of_decode_calls_ref_ == 2) {
+        // If this is the second stream to decode headers, force low watermark state.
+        connection()->onLowWatermark();
+      }
+    }
+    return PassThroughFilter::decodeData(buf, end_stream);
+  }
+
+  Http::FilterDataStatus encodeData(Buffer::Instance& buf, bool end_stream) override {
+    if (end_stream) {
+      absl::WriterMutexLock m(&encode_lock_);
+      number_of_encode_calls_ref_++;
+      if (number_of_encode_calls_ref_ == 1) {
+        // If this is the first stream to encode headers, force high watermark state.
+        connection()->onHighWatermark();
+      }
+    }
+    return PassThroughFilter::encodeData(buf, end_stream);
+  }
+
+  Network::ConnectionImpl* connection() {
+    // As long as we're doing horrible things let's do *all* the horrible things.
+    // Assert the connection we have is a ConnectionImpl and const cast it so we
+    // can force watermark changes.
+    auto conn_impl = dynamic_cast<const Network::ConnectionImpl*>(decoder_callbacks_->connection());
+    return const_cast<Network::ConnectionImpl*>(conn_impl);
+  }
+
+  absl::Mutex& encode_lock_;
+  uint32_t& number_of_encode_calls_ref_;
+  uint32_t& number_of_decode_calls_ref_;
+};
+
+class TestPauseFilterConfig : public Extensions::HttpFilters::Common::EmptyHttpFilterConfig {
+public:
+  TestPauseFilterConfig() : EmptyHttpFilterConfig("pause-filter") {}
+
+  Http::FilterFactoryCb createFilter(const std::string&, Server::Configuration::FactoryContext&) {
+    return [&](Http::FilterChainFactoryCallbacks& callbacks) -> void {
+      // GUARDED_BY insists the lock be held when the guarded variables are passed by reference.
+      absl::WriterMutexLock m(&encode_lock_);
+      callbacks.addStreamFilter(std::make_shared<::Envoy::TestPauseFilter>(
+          encode_lock_, number_of_encode_calls_, number_of_decode_calls_));
+    };
+  }
+
+  absl::Mutex encode_lock_;
+  uint32_t number_of_encode_calls_ GUARDED_BY(encode_lock_) = 0;
+  uint32_t number_of_decode_calls_ GUARDED_BY(encode_lock_) = 0;
+};
+
+// perform static registration
+static Registry::RegisterFactory<TestPauseFilterConfig,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy