engine: add drainConnections API to platform layer (#1731)

Description: this PR uses the main interface API exposed in #1729 and surfaces it for use in the platform layer
Risk Level: low - new API. However actual usage is high risk and should be monitored carefully.
Testing: swift and Kotlin integration tests.

Signed-off-by: Jose Nino <jnino@lyft.com>
Signed-off-by: JP Simard <jp@jpsim.com>

mobile/library/common/jni/jni_interface.cc

@@ -945,3 +945,11 @@ Java_io_envoyproxy_envoymobile_engine_JniLibrary_registerStringAccessor(JNIEnv*
   env->DeleteLocalRef(jcls_JvmStringAccessorContext);
   return result;
 }
+
+extern "C" JNIEXPORT void JNICALL
+Java_io_envoyproxy_envoymobile_engine_JniLibrary_drainConnections(JNIEnv* env,
+                                                                  jclass, // class
+                                                                  jlong engine) {
+  jni_log("[Envoy]", "drainConnections");
+  drain_connections(engine);
+}

mobile/library/java/io/envoyproxy/envoymobile/engine/AndroidEngineImpl.java

@@ -82,4 +82,9 @@ public int recordHistogramValue(String elements, Map<String, String> tags, int v
   public int registerStringAccessor(String accessorName, EnvoyStringAccessor accessor) {
     return envoyEngine.registerStringAccessor(accessorName, accessor);
   }
+
+  @Override
+  public void drainConnections() {
+    envoyEngine.drainConnections();
+  }
 }

mobile/library/java/io/envoyproxy/envoymobile/engine/EnvoyEngine.java

@@ -111,4 +111,9 @@ int runWithTemplate(String configurationYAML, EnvoyConfiguration envoyConfigurat
    * This is a noop if called before the underlying EnvoyEngine has started.
    */
   void flushStats();
+
+  /**
+   * Drain all connections owned by this Engine.
+   */
+  void drainConnections();
 }

mobile/library/java/io/envoyproxy/envoymobile/engine/EnvoyEngineImpl.java

@@ -189,4 +189,9 @@ public int recordHistogramValue(String elements, Map<String, String> tags, int v
   public int registerStringAccessor(String accessor_name, EnvoyStringAccessor accessor) {
     return JniLibrary.registerStringAccessor(accessor_name, new JvmStringAccessorContext(accessor));
   }
+
+  @Override
+  public void drainConnections() {
+    JniLibrary.drainConnections(engineHandle);
+  }
 }

mobile/library/java/io/envoyproxy/envoymobile/engine/JniLibrary.java

@@ -285,4 +285,9 @@ protected static native int recordHistogramValue(long engine, String elements, b
    */
   protected static native int registerStringAccessor(String accessorName,
                                                      JvmStringAccessorContext context);
+
+  /**
+   * Drain all connections owned by this Engine.
+   */
+  protected static native int drainConnections(long engine);
 }

mobile/library/kotlin/io/envoyproxy/envoymobile/Engine.kt

@@ -27,4 +27,9 @@ interface Engine {
    * This is a noop if called before the underlying EnvoyEngine has started.
    */
   fun flushStats()
+
+  /**
+   * Drain all connections owned by this Engine.
+   */
+  fun drainConnections()
 }

mobile/library/kotlin/io/envoyproxy/envoymobile/EngineImpl.kt

@@ -47,4 +47,8 @@ class EngineImpl constructor(
   override fun flushStats() {
     envoyEngine.flushStats()
   }
+
+  override fun drainConnections() {
+    envoyEngine.drainConnections()
+  }
 }

mobile/library/kotlin/io/envoyproxy/envoymobile/mocks/MockEnvoyEngine.kt

@@ -42,4 +42,6 @@ internal class MockEnvoyEngine : EnvoyEngine {
   override fun registerStringAccessor(accessorName: String, accessor: EnvoyStringAccessor): Int = 0
 
   override fun flushStats() = Unit
+
+  override fun drainConnections() = Unit
 }

mobile/library/objective-c/EnvoyEngine.h

@@ -466,6 +466,8 @@ extern const int kEnvoyFailure;
 
 - (void)terminate;
 
+- (void)drainConnections;
+
 @end
 
 #pragma mark - EnvoyLogger

mobile/library/objective-c/EnvoyEngineImpl.m

@@ -544,6 +544,10 @@ - (void)terminate {
   terminate_engine(_engineHandle);
 }
 
+- (void)drainConnections {
+  drain_connections(_engineHandle);
+}
+
 #pragma mark - Private
 
 - (void)startObservingLifecycleNotifications {

mobile/library/swift/Engine.swift

@@ -17,4 +17,7 @@ public protocol Engine: AnyObject {
 
   /// Terminates the running engine.
   func terminate()
+
+  /// Drain all connections owned by this Engine.
+  func drainConnections()
 }

mobile/library/swift/EngineImpl.swift

@@ -65,4 +65,8 @@ extension EngineImpl: Engine {
   func terminate() {
     self.engine.terminate()
   }
+
+  func drainConnections() {
+    self.engine.drainConnections()
+  }
 }

mobile/library/swift/mocks/MockEnvoyEngine.swift

@@ -91,4 +91,6 @@ extension MockEnvoyEngine: EnvoyEngine {
   }
 
   func terminate() {}
+
+  func drainConnections() {}
 }

mobile/test/kotlin/integration/BUILD

@@ -71,6 +71,20 @@ envoy_mobile_jni_kt_test(
     ],
 )
 
+envoy_mobile_jni_kt_test(
+    name = "drain_connections_test",
+    srcs = [
+        "DrainConnectionsTest.kt",
+    ],
+    native_deps = [
+        "//library/common/jni:libjava_jni_lib.so",
+        "//library/common/jni:java_jni_lib.jnilib",
+    ],
+    deps = [
+        "//library/kotlin/io/envoyproxy/envoymobile:envoy_interfaces_lib",
+    ],
+)
+
 envoy_mobile_jni_kt_test(
     name = "grpc_receive_error_test",
     srcs = [

mobile/test/kotlin/integration/DrainConnectionsTest.kt

@@ -0,0 +1,118 @@
+package test.kotlin.integration
+
+import io.envoyproxy.envoymobile.Custom
+import io.envoyproxy.envoymobile.EngineBuilder
+import io.envoyproxy.envoymobile.RequestHeadersBuilder
+import io.envoyproxy.envoymobile.RequestMethod
+import io.envoyproxy.envoymobile.ResponseHeaders
+import io.envoyproxy.envoymobile.UpstreamHttpProtocol
+import io.envoyproxy.envoymobile.engine.JniLibrary
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.fail
+import org.junit.Test
+
+private val apiListenerType = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.EnvoyMobileHttpConnectionManager"
+private val assertionFilterType = "type.googleapis.com/envoymobile.extensions.filters.http.assertion.Assertion"
+private val config =
+"""
+static_resources:
+  listeners:
+  - name: base_api_listener
+    address:
+      socket_address:
+        protocol: TCP
+        address: 0.0.0.0
+        port_value: 10000
+    api_listener:
+      api_listener:
+        "@type": $apiListenerType
+        config:
+          stat_prefix: hcm
+          route_config:
+            name: api_router
+            virtual_hosts:
+              - name: api
+                domains:
+                  - "*"
+                routes:
+                  - match:
+                      prefix: "/"
+                    direct_response:
+                      status: 200
+          http_filters:
+            - name: envoy.filters.http.assertion
+              typed_config:
+                "@type": $assertionFilterType
+                match_config:
+                  http_request_headers_match:
+                    headers:
+                      - name: ":authority"
+                        exact_match: example.com
+            - name: envoy.router
+              typed_config:
+                "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
+"""
+
+class DrainConnectionsTest {
+
+  init {
+    JniLibrary.loadTestLibrary()
+  }
+
+  @Test
+  fun `successful request after connection drain`() {
+    val headersExpectation = CountDownLatch(2)
+
+    val engine = EngineBuilder(Custom(config)).build()
+    val client = engine.streamClient()
+
+    val requestHeaders = RequestHeadersBuilder(
+      method = RequestMethod.GET,
+      scheme = "https",
+      authority = "example.com",
+      path = "/test"
+    )
+      .addUpstreamHttpProtocol(UpstreamHttpProtocol.HTTP2)
+      .build()
+
+    var resultHeaders1: ResponseHeaders? = null
+    var resultEndStream1: Boolean? = null
+    client.newStreamPrototype()
+      .setOnResponseHeaders { responseHeaders, endStream, _ ->
+        resultHeaders1 = responseHeaders
+        resultEndStream1 = endStream
+        headersExpectation.countDown()
+      }
+      .setOnError { _, _ -> fail("Unexpected error") }
+      .start()
+      .sendHeaders(requestHeaders, true)
+
+    headersExpectation.await(10, TimeUnit.SECONDS)
+
+    engine.drainConnections()
+
+    var resultHeaders2: ResponseHeaders? = null
+    var resultEndStream2: Boolean? = null
+    client.newStreamPrototype()
+      .setOnResponseHeaders { responseHeaders, endStream, _ ->
+        resultHeaders2 = responseHeaders
+        resultEndStream2 = endStream
+        headersExpectation.countDown()
+      }
+      .setOnError { _, _ -> fail("Unexpected error") }
+      .start()
+      .sendHeaders(requestHeaders, true)
+
+    headersExpectation.await(10, TimeUnit.SECONDS)
+
+    engine.terminate()
+
+    assertThat(headersExpectation.count).isEqualTo(0)
+    assertThat(resultHeaders1!!.httpStatus).isEqualTo(200)
+    assertThat(resultEndStream1).isTrue()
+    assertThat(resultHeaders2!!.httpStatus).isEqualTo(200)
+    assertThat(resultEndStream2).isTrue()
+  }
+}

mobile/test/swift/integration/BUILD

@@ -12,6 +12,16 @@ envoy_mobile_swift_test(
     ],
 )
 
+envoy_mobile_swift_test(
+    name = "drain_connections_test",
+    srcs = [
+        "DrainConnectionsTest.swift",
+    ],
+    deps = [
+        "//library/objective-c:envoy_engine_objc_lib",
+    ],
+)
+
 envoy_mobile_swift_test(
     name = "direct_response_contains_headers_integration_test",
     srcs = [

mobile/test/swift/integration/DrainConnectionsTest.swift

@@ -0,0 +1,101 @@
+import Envoy
+import EnvoyEngine
+import Foundation
+import XCTest
+
+final class DrainConnectionsTest: XCTestCase {
+  func testDrainConnections() {
+    // swiftlint:disable:next line_length
+    let emhcmType = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.EnvoyMobileHttpConnectionManager"
+    // swiftlint:disable:next line_length
+    let assertionFilterType = "type.googleapis.com/envoymobile.extensions.filters.http.assertion.Assertion"
+    let config =
+"""
+static_resources:
+  listeners:
+  - name: base_api_listener
+    address:
+      socket_address:
+        protocol: TCP
+        address: 0.0.0.0
+        port_value: 10000
+    api_listener:
+      api_listener:
+        "@type": \(emhcmType)
+        config:
+          stat_prefix: hcm
+          route_config:
+            name: api_router
+            virtual_hosts:
+              - name: api
+                domains:
+                  - "*"
+                routes:
+                  - match:
+                      prefix: "/"
+                    direct_response:
+                      status: 200
+          http_filters:
+            - name: envoy.filters.http.assertion
+              typed_config:
+                "@type": \(assertionFilterType)
+                match_config:
+                  http_request_headers_match:
+                    headers:
+                      - name: ":authority"
+                        exact_match: example.com
+            - name: envoy.router
+              typed_config:
+                "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
+"""
+    let engine = EngineBuilder(yaml: config)
+      .addLogLevel(.debug)
+      .build()
+
+    let client = engine
+      .streamClient()
+
+    let requestHeaders = RequestHeadersBuilder(method: .get, scheme: "https",
+                                               authority: "example.com", path: "/test")
+      .addUpstreamHttpProtocol(.http2)
+      .build()
+
+    let expectation1 =
+      self.expectation(description: "Run called with expected http status first request")
+
+    client
+      .newStreamPrototype()
+      .setOnResponseHeaders { responseHeaders, endStream, _ in
+         XCTAssertEqual(200, responseHeaders.httpStatus)
+         XCTAssertTrue(endStream)
+         expectation1.fulfill()
+      }
+      .setOnError { _, _ in
+        XCTFail("Unexpected error")
+      }
+      .start()
+      .sendHeaders(requestHeaders, endStream: true)
+
+    XCTAssertEqual(XCTWaiter.wait(for: [expectation1], timeout: 1), .completed)
+
+    engine.drainConnections()
+
+    let expectation2 =
+      self.expectation(description: "Run called with expected http status first request")
+
+    client
+      .newStreamPrototype()
+      .setOnResponseHeaders { responseHeaders, endStream, _ in
+         XCTAssertEqual(200, responseHeaders.httpStatus)
+         XCTAssertTrue(endStream)
+         expectation2.fulfill()
+      }
+      .setOnError { _, _ in
+        XCTFail("Unexpected error")
+      }
+      .start()
+      .sendHeaders(requestHeaders, endStream: true)
+
+    XCTAssertEqual(XCTWaiter.wait(for: [expectation2], timeout: 1), .completed)
+  }
+}