Monitoring and idle connection reaping resurrected

src/main/java/software/amazon/awssdk/crt/http/HttpClientConnectionManager.java

@@ -109,6 +109,14 @@ private HttpClientConnectionManager(HttpClientConnectionManagerOptions options)
             proxyAuthorizationPassword = proxyOptions.getAuthorizationPassword();
         }
 
+        HttpMonitoringOptions monitoringOptions = options.getMonitoringOptions();
+        long monitoringThroughputThresholdInBytesPerSecond = 0;
+        int monitoringFailureIntervalInSeconds = 0;
+        if (monitoringOptions != null) {
+            monitoringThroughputThresholdInBytesPerSecond = monitoringOptions.getMinThroughputBytesPerSecond();
+            monitoringFailureIntervalInSeconds = monitoringOptions.getAllowableThroughputFailureIntervalSeconds();
+        }
+
         acquireNativeHandle(httpClientConnectionManagerNew(this,
                                             clientBootstrap.getNativeHandle(),
                                             socketOptions.getNativeHandle(),
@@ -123,7 +131,10 @@ private HttpClientConnectionManager(HttpClientConnectionManagerOptions options)
                                             proxyAuthorizationType,
                                             proxyAuthorizationUsername != null ? proxyAuthorizationUsername.getBytes(UTF8) : null,
                                             proxyAuthorizationPassword != null ? proxyAuthorizationPassword.getBytes(UTF8) : null,
-                                            options.isManualWindowManagement()));
+                                            options.isManualWindowManagement(),
+                                            options.getMaxConnectionIdleInMilliseconds(),
+                                            monitoringThroughputThresholdInBytesPerSecond,
+                                            monitoringFailureIntervalInSeconds));
 
         /* we don't need to add a reference to socketOptions since it's copied during connection manager construction */
          addReferenceTo(clientBootstrap);
@@ -261,7 +272,10 @@ private static native long httpClientConnectionManagerNew(HttpClientConnectionMa
                                                         int proxyAuthorizationType,
                                                         byte[] proxyAuthorizationUsername,
                                                         byte[] proxyAuthorizationPassword,
-                                                        boolean isManualWindowManagement) throws CrtRuntimeException;
+                                                        boolean isManualWindowManagement,
+                                                        long maxConnectionIdleInMilliseconds,
+                                                        long monitoringThroughputThresholdInBytesPerSecond,
+                                                        int monitoringFailureIntervalInSeconds) throws CrtRuntimeException;
 
     private static native void httpClientConnectionManagerRelease(long conn_manager) throws CrtRuntimeException;
 

src/main/java/software/amazon/awssdk/crt/http/HttpClientConnectionManagerOptions.java

@@ -37,6 +37,8 @@ public class HttpClientConnectionManagerOptions {
     private int maxConnections = DEFAULT_MAX_CONNECTIONS;
     private HttpProxyOptions proxyOptions;
     private boolean manualWindowManagement = false;
+    private HttpMonitoringOptions monitoringOptions;
+    private long maxConnectionIdleInMilliseconds = 0;
 
     public HttpClientConnectionManagerOptions() {
     }
@@ -203,5 +205,36 @@ public HttpClientConnectionManagerOptions withManualWindowManagement(boolean man
         this.manualWindowManagement = manualWindowManagement;
         return this;
     }
+
+    /**
+     * Sets maximum amount of time, in milliseconds, that the connection can be idle in the manager before
+     * getting culled by the manager
+     * @param maxConnectionIdleInMilliseconds How long to allow connections to be idle before reaping them
+     * @return this
+     */
+    public HttpClientConnectionManagerOptions withMaxConnectionIdleInMilliseconds(long maxConnectionIdleInMilliseconds) {
+        this.maxConnectionIdleInMilliseconds = maxConnectionIdleInMilliseconds;
+        return this;
+    }
+
+    /**
+     * @return How long to allow connections to be idle before reaping them
+     */
+    public long getMaxConnectionIdleInMilliseconds() { return maxConnectionIdleInMilliseconds; }
+
+    /**
+     * Sets the monitoring options for connections in the connection pool
+     * @param monitoringOptions Monitoring options for this connection manager, or null to disable monitoring
+     * @return this
+     */
+    public HttpClientConnectionManagerOptions withMonitoringOptions(HttpMonitoringOptions monitoringOptions) {
+        this.monitoringOptions = monitoringOptions;
+        return this;
+    }
+
+    /**
+     * @return the monitoring options for connections in the connection pool
+     */
+    public HttpMonitoringOptions getMonitoringOptions() { return monitoringOptions; }
 }
 

src/main/java/software/amazon/awssdk/crt/http/HttpMonitoringOptions.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package software.amazon.awssdk.crt.http;
+
+/**
+ * This class provides access to basic http connection monitoring controls in lieu of the more traditional
+ * timeouts.
+ *
+ * The user can set a throughput threshold (in bytes per second) for the a connection to be considered healthy.  If
+ * the connection falls below this threshold for a configurable amount of time, then the connection is considered
+ * unhealthy and shut down.  Throughput/health is only measured when the connection has work (read or write) that
+ * needs to be done.
+ */
+public class HttpMonitoringOptions {
+
+    /**
+     * minimum amount of throughput, in bytes per second, for a connection to be considered healthy.
+     */
+    private long minThroughputBytesPerSecond;
+
+    /**
+     * How long, in seconds, a connection is allowed to be unhealthy before getting shut down.  Must be at least
+     * two.
+     */
+    private int allowableThroughputFailureIntervalSeconds;
+
+    /**
+     * Creates a new set of monitoring options
+     */
+    public HttpMonitoringOptions() {
+    }
+
+    /**
+     * Sets a throughput threshold for connections.  Throughput below this value will be considered unhealthy.
+     * @param minThroughputBytesPerSecond minimum amount of throughput, in bytes per second, for a connection to be
+     *                                    considered healthy.
+     */
+    public void setMinThroughputBytesPerSecond(long minThroughputBytesPerSecond) {
+        if (minThroughputBytesPerSecond < 0) {
+            throw new IllegalArgumentException("Http monitoring minimum throughput must be non-negative");
+        }
+        this.minThroughputBytesPerSecond = minThroughputBytesPerSecond;
+    }
+
+    /**
+     * @return minimum amount of throughput, in bytes per second, for a connection to be considered healthy.
+     */
+    public long getMinThroughputBytesPerSecond() { return minThroughputBytesPerSecond; }
+
+    /**
+     * Sets how long, in seconds, a connection is allowed to be unhealthy before getting shut down.  Must be at
+     * least two.
+     * @param allowableThroughputFailureIntervalSeconds How long, in seconds, a connection is allowed to be unhealthy
+     *                                                  before getting shut down.
+     */
+    public void setAllowableThroughputFailureIntervalSeconds(int allowableThroughputFailureIntervalSeconds) {
+        if (allowableThroughputFailureIntervalSeconds < 2) {
+            throw new IllegalArgumentException("Http monitoring failure interval must be at least two");
+        }
+        this.allowableThroughputFailureIntervalSeconds = allowableThroughputFailureIntervalSeconds;
+    }
+
+    /**
+     * @return How long, in seconds, a connection is allowed to be unhealthy before getting shut down.  Must be at
+     * least two.
+     */
+    public int getAllowableThroughputFailureIntervalSeconds() { return allowableThroughputFailureIntervalSeconds; }
+
+
+}

src/native/http_connection_manager.c

@@ -145,7 +145,10 @@ JNIEXPORT jlong JNICALL Java_software_amazon_awssdk_crt_http_HttpClientConnectio
     jint jni_proxy_authorization_type,
     jbyteArray jni_proxy_authorization_username,
     jbyteArray jni_proxy_authorization_password,
-    jboolean jni_manual_window_management) {
+    jboolean jni_manual_window_management,
+    jlong jni_max_connection_idle_in_milliseconds,
+    jlong jni_monitoring_throughput_threshold_in_bytes_per_second,
+    jint jni_monitoring_failure_interval_in_seconds) {
 
     (void)jni_class;
 
@@ -218,11 +221,21 @@ JNIEXPORT jlong JNICALL Java_software_amazon_awssdk_crt_http_HttpClientConnectio
     manager_options.monitoring_options = NULL;
     /* TODO: this variable needs to be renamed in aws-c-http. Come back and change it next revision. */
     manager_options.enable_read_back_pressure = jni_manual_window_management;
+    manager_options.max_connection_idle_in_milliseconds = jni_max_connection_idle_in_milliseconds;
 
     if (use_tls) {
         manager_options.tls_connection_options = &tls_conn_options;
     }
 
+    struct aws_http_connection_monitoring_options monitoring_options;
+    AWS_ZERO_STRUCT(monitoring_options);
+    if (jni_monitoring_throughput_threshold_in_bytes_per_second >= 0 && jni_monitoring_failure_interval_in_seconds >= 2) {
+        monitoring_options.minimum_throughput_bytes_per_second = jni_monitoring_throughput_threshold_in_bytes_per_second;
+        monitoring_options.allowable_throughput_failure_interval_seconds = jni_monitoring_failure_interval_in_seconds;
+
+        manager_options.monitoring_options = &monitoring_options;
+    }
+
     struct aws_http_proxy_options proxy_options;
     AWS_ZERO_STRUCT(proxy_options);