src/native/http2_stream_manager.c

/**
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 * SPDX-License-Identifier: Apache-2.0.
 */

#include "crt.h"
#include "http_connection_manager.h"
#include "http_request_response.h"
#include "http_request_utils.h"
#include "java_class_ids.h"

#include <http_proxy_options.h>
#include <jni.h>
#include <string.h>

#include <aws/common/condition_variable.h>
#include <aws/common/string.h>

#include <aws/io/channel_bootstrap.h>
#include <aws/io/event_loop.h>
#include <aws/io/logging.h>
#include <aws/io/socket.h>
#include <aws/io/tls_channel_handler.h>

#include <aws/http/connection.h>
#include <aws/http/connection_manager.h>
#include <aws/http/http.h>
#include <aws/http/http2_stream_manager.h>
#include <aws/http/proxy.h>

/* on 32-bit platforms, casting pointers to longs throws a warning we don't need */
#if UINTPTR_MAX == 0xffffffff
#    if defined(_MSC_VER)
#        pragma warning(push)
#        pragma warning(disable : 4305) /* 'type cast': truncation from 'jlong' to 'jni_tls_ctx_options *' */
#    else
#        pragma GCC diagnostic push
#        pragma GCC diagnostic ignored "-Wpointer-to-int-cast"
#        pragma GCC diagnostic ignored "-Wint-to-pointer-cast"
#    endif
#endif

/*
 * Stream manager binding, persists across the lifetime of the native object.
 */
struct aws_http2_stream_manager_binding {
    JavaVM *jvm;
    jweak java_http2_stream_manager;
    struct aws_http2_stream_manager *stream_manager;
};

static void s_destroy_manager_binding(struct aws_http2_stream_manager_binding *binding, JNIEnv *env) {
    if (binding == NULL) {
        return;
    }
    if (binding->java_http2_stream_manager != NULL) {
        (*env)->DeleteWeakGlobalRef(env, binding->java_http2_stream_manager);
    }

    aws_mem_release(aws_jni_get_allocator(), binding);
}

static void s_on_stream_manager_shutdown_complete_callback(void *user_data) {

    struct aws_http2_stream_manager_binding *binding = (struct aws_http2_stream_manager_binding *)user_data;
    /********** JNI ENV ACQUIRE **********/
    JNIEnv *env = aws_jni_acquire_thread_env(binding->jvm);
    if (env == NULL) {
        /* If we can't get an environment, then the JVM is probably shutting down.  Don't crash. */
        return;
    }

    AWS_LOGF_DEBUG(AWS_LS_HTTP_STREAM_MANAGER, "Java Stream Manager Shutdown Complete");
    jobject java_http2_stream_manager = (*env)->NewLocalRef(env, binding->java_http2_stream_manager);
    if (java_http2_stream_manager != NULL) {
        (*env)->CallVoidMethod(env, java_http2_stream_manager, http2_stream_manager_properties.onShutdownComplete);

        /* If exception raised from Java callback, but we already closed the stream manager, just move on */
        aws_jni_check_and_clear_exception(env);

        (*env)->DeleteLocalRef(env, java_http2_stream_manager);
    }

    /* We're done with this wrapper, free it. */
    s_destroy_manager_binding(binding, env);
    aws_jni_release_thread_env(binding->jvm, env);
    /********** JNI ENV RELEASE **********/
}

JNIEXPORT jlong JNICALL Java_software_amazon_awssdk_crt_http_Http2StreamManager_http2StreamManagerNew(
    JNIEnv *env,
    jclass jni_class,
    jobject stream_manager_jobject,
    jlong jni_client_bootstrap,
    jlong jni_socket_options,
    jlong jni_tls_ctx,
    jlong jni_tls_connection_options,
    jlongArray java_marshalled_settings,
    jbyteArray jni_endpoint,
    jint jni_port,
    jint jni_proxy_connection_type,
    jbyteArray jni_proxy_host,
    jint jni_proxy_port,
    jlong jni_proxy_tls_context,
    jint jni_proxy_authorization_type,
    jbyteArray jni_proxy_authorization_username,
    jbyteArray jni_proxy_authorization_password,
    jboolean jni_manual_window_management,
    jlong jni_monitoring_throughput_threshold_in_bytes_per_second,
    jint jni_monitoring_failure_interval_in_seconds,
    jint jni_max_conns,
    jint jni_ideal_concurrent_streams_per_connection,
    jint jni_max_concurrent_streams_per_connection,
    jboolean jni_prior_knowledge,
    jboolean jni_close_connection_on_server_error,
    jint jni_connection_ping_period_ms,
    jint jni_connection_ping_timeout_ms) {

    (void)jni_class;
    aws_cache_jni_ids(env);

    struct aws_client_bootstrap *client_bootstrap = (struct aws_client_bootstrap *)jni_client_bootstrap;
    struct aws_socket_options *socket_options = (struct aws_socket_options *)jni_socket_options;
    struct aws_tls_ctx *tls_ctx = (struct aws_tls_ctx *)jni_tls_ctx;
    struct aws_tls_connection_options *tls_connection_options =
        (struct aws_tls_connection_options *)jni_tls_connection_options;
    struct aws_http2_stream_manager_binding *binding = NULL;
    struct aws_allocator *allocator = aws_jni_get_allocator();

    if (!client_bootstrap) {
        aws_jni_throw_illegal_argument_exception(env, "ClientBootstrap can't be null");
        return (jlong)NULL;
    }

    if (!socket_options) {
        aws_jni_throw_illegal_argument_exception(env, "SocketOptions can't be null");
        return (jlong)NULL;
    }

    const size_t marshalled_len = (*env)->GetArrayLength(env, java_marshalled_settings);
    AWS_ASSERT(marshalled_len % 2 == 0);

    size_t num_initial_settings = marshalled_len / 2;
    struct aws_http2_setting *initial_settings =
        num_initial_settings ? aws_mem_calloc(allocator, num_initial_settings, sizeof(struct aws_http2_setting)) : NULL;

    jlong *marshalled_settings = (*env)->GetLongArrayElements(env, java_marshalled_settings, NULL);
    for (size_t i = 0; i < num_initial_settings; i++) {
        jlong id = marshalled_settings[i * 2];
        initial_settings[i].id = (uint32_t)id;
        jlong value = marshalled_settings[i * 2 + 1];
        /* We checked the value can fit into uint32_t in Java already */
        initial_settings[i].value = (uint32_t)value;
    }

    struct aws_byte_cursor endpoint = aws_jni_byte_cursor_from_jbyteArray_acquire(env, jni_endpoint);

    if (jni_port == 0) {
        aws_jni_throw_illegal_argument_exception(env, "Port must not be 0");
        goto cleanup;
    }

    if (jni_max_conns <= 0) {
        aws_jni_throw_illegal_argument_exception(env, "Max Connections must be > 0");
        goto cleanup;
    }

    uint32_t port = (uint32_t)jni_port;

    bool new_tls_conn_opts = (jni_tls_ctx != 0 && !tls_connection_options);

    struct aws_tls_connection_options tls_conn_options;
    AWS_ZERO_STRUCT(tls_conn_options);
    if (new_tls_conn_opts) {
        aws_tls_connection_options_init_from_ctx(&tls_conn_options, tls_ctx);
        aws_tls_connection_options_set_server_name(&tls_conn_options, allocator, &endpoint);
        tls_connection_options = &tls_conn_options;
    }

    binding = aws_mem_calloc(allocator, 1, sizeof(struct aws_http2_stream_manager_binding));
    AWS_FATAL_ASSERT(binding);
    binding->java_http2_stream_manager = (*env)->NewWeakGlobalRef(env, stream_manager_jobject);

    jint jvmresult = (*env)->GetJavaVM(env, &binding->jvm);
    (void)jvmresult;
    AWS_FATAL_ASSERT(jvmresult == 0);

    struct aws_http2_stream_manager_options manager_options = {
        .bootstrap = client_bootstrap,
        .initial_settings_array = initial_settings,
        .num_initial_settings = num_initial_settings,
        .socket_options = socket_options,
        .http2_prior_knowledge = jni_prior_knowledge,
        .tls_connection_options = tls_connection_options,
        .monitoring_options = NULL,
        .host = endpoint,
        .port = port,
        .shutdown_complete_callback = &s_on_stream_manager_shutdown_complete_callback,
        .shutdown_complete_user_data = binding,
        .enable_read_back_pressure = jni_manual_window_management,
        .close_connection_on_server_error = jni_close_connection_on_server_error,
        .connection_ping_period_ms = jni_connection_ping_period_ms,
        .connection_ping_timeout_ms = jni_connection_ping_timeout_ms,
        .ideal_concurrent_streams_per_connection = (size_t)jni_ideal_concurrent_streams_per_connection,
        .max_concurrent_streams_per_connection = (size_t)jni_max_concurrent_streams_per_connection,
        .max_connections = (size_t)jni_max_conns,
    };

    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);

    struct aws_tls_connection_options proxy_tls_conn_options;
    AWS_ZERO_STRUCT(proxy_tls_conn_options);

    aws_http_proxy_options_jni_init(
        env,
        &proxy_options,
        jni_proxy_connection_type,
        &proxy_tls_conn_options,
        jni_proxy_host,
        jni_proxy_port,
        jni_proxy_authorization_username,
        jni_proxy_authorization_password,
        jni_proxy_authorization_type,
        (struct aws_tls_ctx *)jni_proxy_tls_context);

    if (jni_proxy_host != NULL) {
        manager_options.proxy_options = &proxy_options;
    }

    binding->stream_manager = aws_http2_stream_manager_new(allocator, &manager_options);
    if (binding->stream_manager == NULL) {
        aws_jni_throw_runtime_exception(env, "Failed to create stream manager: %s", aws_error_str(aws_last_error()));
    }

    aws_http_proxy_options_jni_clean_up(
        env, &proxy_options, jni_proxy_host, jni_proxy_authorization_username, jni_proxy_authorization_password);

    if (new_tls_conn_opts) {
        aws_tls_connection_options_clean_up(&tls_conn_options);
    }

cleanup:
    aws_jni_byte_cursor_from_jbyteArray_release(env, jni_endpoint, endpoint);

    if (binding->stream_manager == NULL) {
        s_destroy_manager_binding(binding, env);
        binding = NULL;
    }

    return (jlong)binding;
}

/*
 * Stream manager binding, persists across the lifetime of the native object.
 */
struct aws_sm_acquire_stream_callback_data {
    JavaVM *jvm;
    struct http_stream_binding *stream_binding;
    jobject java_async_callback;
};

static void s_cleanup_sm_acquire_stream_callback_data(
    struct aws_sm_acquire_stream_callback_data *callback_data,
    JNIEnv *env) {

    if (callback_data->java_async_callback) {
        (*env)->DeleteGlobalRef(env, callback_data->java_async_callback);
    }
    aws_mem_release(aws_jni_get_allocator(), callback_data);
}

static struct aws_sm_acquire_stream_callback_data *s_new_sm_acquire_stream_callback_data(
    JNIEnv *env,
    struct aws_allocator *allocator,
    struct http_stream_binding *stream_binding,
    jobject async_callback) {
    struct aws_sm_acquire_stream_callback_data *callback_data =
        aws_mem_calloc(allocator, 1, sizeof(struct aws_sm_acquire_stream_callback_data));

    jint jvmresult = (*env)->GetJavaVM(env, &callback_data->jvm);
    AWS_FATAL_ASSERT(jvmresult == 0);
    callback_data->java_async_callback = async_callback ? (*env)->NewGlobalRef(env, async_callback) : NULL;
    AWS_FATAL_ASSERT(callback_data->java_async_callback != NULL);
    callback_data->stream_binding = stream_binding;

    return callback_data;
}

static void s_on_stream_acquired(struct aws_http_stream *stream, int error_code, void *user_data) {
    struct aws_sm_acquire_stream_callback_data *callback_data = user_data;
    /********** JNI ENV ACQUIRE **********/
    JNIEnv *env = aws_jni_acquire_thread_env(callback_data->jvm);
    if (env == NULL) {
        /* If we can't get an environment, then the JVM is probably shutting down.  Don't crash. */
        return;
    }
    if (error_code) {
        AWS_ASSERT(stream == NULL);
        jobject crt_exception = aws_jni_new_crt_exception_from_error_code(env, error_code);
        (*env)->CallVoidMethod(
            env, callback_data->java_async_callback, async_callback_properties.on_failure, crt_exception);
        (*env)->DeleteLocalRef(env, crt_exception);
        aws_http_stream_binding_release(env, callback_data->stream_binding);
    } else {
        /* Acquire for the native stream. The destroy callback for native stream will release the ref. */
        aws_http_stream_binding_acquire(callback_data->stream_binding);

        callback_data->stream_binding->native_stream = stream;
        jobject j_http_stream =
            aws_java_http_stream_from_native_new(env, callback_data->stream_binding, AWS_HTTP_VERSION_2);
        if (!j_http_stream) {
            jthrowable crt_exception = (*env)->ExceptionOccurred(env);
            AWS_ASSERT(crt_exception);
            (*env)->CallVoidMethod(
                env, callback_data->java_async_callback, async_callback_properties.on_failure, crt_exception);
            (*env)->DeleteLocalRef(env, crt_exception);
            (*env)->ExceptionClear(env);
            /* Release the refcount on binding for the java object that failed to be created. */
            aws_http_stream_binding_release(env, callback_data->stream_binding);
        } else {
            callback_data->stream_binding->java_http_stream_base = (*env)->NewGlobalRef(env, j_http_stream);
            (*env)->CallVoidMethod(
                env,
                callback_data->java_async_callback,
                async_callback_properties.on_success_with_object,
                callback_data->stream_binding->java_http_stream_base);
            (*env)->DeleteLocalRef(env, j_http_stream);
        }
    }
    AWS_FATAL_ASSERT(!aws_jni_check_and_clear_exception(env));
    s_cleanup_sm_acquire_stream_callback_data(callback_data, env);
    aws_jni_release_thread_env(callback_data->jvm, env);
    /********** JNI ENV RELEASE **********/
}

JNIEXPORT void JNICALL Java_software_amazon_awssdk_crt_http_Http2StreamManager_http2StreamManagerAcquireStream(
    JNIEnv *env,
    jclass jni_class,
    jlong jni_stream_manager,
    jbyteArray marshalled_request,
    jobject jni_http_request_body_stream,
    jobject jni_http_response_callback_handler,
    jobject java_async_callback) {
    (void)jni_class;
    aws_cache_jni_ids(env);

    struct aws_http2_stream_manager_binding *sm_binding = (struct aws_http2_stream_manager_binding *)jni_stream_manager;
    struct aws_http2_stream_manager *stream_manager = sm_binding->stream_manager;

    if (!stream_manager) {
        aws_jni_throw_illegal_argument_exception(env, "Stream Manager can't be null");
        return;
    }

    if (!jni_http_response_callback_handler) {
        aws_jni_throw_illegal_argument_exception(
            env, "Http2StreamManager.acquireStream: Invalid jni_http_response_callback_handler");
        return;
    }
    if (!java_async_callback) {
        aws_jni_throw_illegal_argument_exception(env, "Http2StreamManager.acquireStream: Invalid async callback");
        return;
    }

    struct http_stream_binding *stream_binding = aws_http_stream_binding_new(env, jni_http_response_callback_handler);
    if (!stream_binding) {
        /* Exception already thrown */
        return;
    }

    stream_binding->native_request =
        aws_http_request_new_from_java_http_request(env, marshalled_request, jni_http_request_body_stream);
    if (stream_binding->native_request == NULL) {
        /* Exception already thrown */
        aws_http_stream_binding_release(env, stream_binding);
        return;
    }

    struct aws_http_make_request_options request_options = {
        .self_size = sizeof(request_options),
        .request = stream_binding->native_request,
        /* Set Callbacks */
        .on_response_headers = aws_java_http_stream_on_incoming_headers_fn,
        .on_response_header_block_done = aws_java_http_stream_on_incoming_header_block_done_fn,
        .on_response_body = aws_java_http_stream_on_incoming_body_fn,
        .on_complete = aws_java_http_stream_on_stream_complete_fn,
        .on_destroy = aws_java_http_stream_on_stream_destroy_fn,
        .user_data = stream_binding,
    };

    struct aws_allocator *allocator = aws_jni_get_allocator();
    struct aws_sm_acquire_stream_callback_data *callback_data =
        s_new_sm_acquire_stream_callback_data(env, allocator, stream_binding, java_async_callback);

    struct aws_http2_stream_manager_acquire_stream_options acquire_options = {
        .options = &request_options,
        .callback = s_on_stream_acquired,
        .user_data = callback_data,
    };

    aws_http2_stream_manager_acquire_stream(sm_binding->stream_manager, &acquire_options);
}

JNIEXPORT void JNICALL Java_software_amazon_awssdk_crt_http_Http2StreamManager_http2StreamManagerRelease(
    JNIEnv *env,
    jclass jni_class,
    jlong jni_stream_manager) {
    (void)jni_class;
    aws_cache_jni_ids(env);

    struct aws_http2_stream_manager_binding *sm_binding = (struct aws_http2_stream_manager_binding *)jni_stream_manager;
    struct aws_http2_stream_manager *stream_manager = sm_binding->stream_manager;

    if (!stream_manager) {
        aws_jni_throw_runtime_exception(env, "Stream Manager can't be null");
        return;
    }

    AWS_LOGF_DEBUG(AWS_LS_HTTP_CONNECTION, "Releasing StreamManager: id: %p", (void *)stream_manager);
    aws_http2_stream_manager_release(stream_manager);
}

JNIEXPORT jobject JNICALL Java_software_amazon_awssdk_crt_http_Http2StreamManager_http2StreamManagerFetchMetrics(
    JNIEnv *env,
    jclass jni_class,
    jlong jni_stream_manager) {
    (void)jni_class;
    aws_cache_jni_ids(env);

    struct aws_http2_stream_manager_binding *sm_binding = (struct aws_http2_stream_manager_binding *)jni_stream_manager;
    struct aws_http2_stream_manager *stream_manager = sm_binding->stream_manager;

    if (!stream_manager) {
        aws_jni_throw_runtime_exception(env, "Stream Manager can't be null");
        return NULL;
    }

    struct aws_http_manager_metrics metrics;
    aws_http2_stream_manager_fetch_metrics(stream_manager, &metrics);

    return (*env)->NewObject(
        env,
        http_manager_metrics_properties.http_manager_metrics_class,
        http_manager_metrics_properties.constructor_method_id,
        (jlong)metrics.available_concurrency,
        (jlong)metrics.pending_concurrency_acquires,
        (jlong)metrics.leased_concurrency);
}