diff options
| author | orivej <orivej@yandex-team.ru> | 2022-02-10 16:45:01 +0300 |
|---|---|---|
| committer | Daniil Cherednik <dcherednik@yandex-team.ru> | 2022-02-10 16:45:01 +0300 |
| commit | 2d37894b1b037cf24231090eda8589bbb44fb6fc (patch) | |
| tree | be835aa92c6248212e705f25388ebafcf84bc7a1 /contrib/restricted/aws/aws-c-event-stream/source | |
| parent | 718c552901d703c502ccbefdfc3c9028d608b947 (diff) | |
| download | ydb-2d37894b1b037cf24231090eda8589bbb44fb6fc.tar.gz | |
Restoring authorship annotation for <orivej@yandex-team.ru>. Commit 2 of 2.
Diffstat (limited to 'contrib/restricted/aws/aws-c-event-stream/source')
5 files changed, 2932 insertions, 2932 deletions
diff --git a/contrib/restricted/aws/aws-c-event-stream/source/event_stream.c b/contrib/restricted/aws/aws-c-event-stream/source/event_stream.c index 9a308a5148..224690f100 100644 --- a/contrib/restricted/aws/aws-c-event-stream/source/event_stream.c +++ b/contrib/restricted/aws/aws-c-event-stream/source/event_stream.c @@ -1,6 +1,6 @@ -/** - * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. - * SPDX-License-Identifier: Apache-2.0. +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. */ #include <aws/event-stream/event_stream.h> @@ -8,7 +8,7 @@ #include <aws/checksums/crc.h> #include <aws/common/encoding.h> -#include <aws/io/io.h> +#include <aws/io/io.h> #include <inttypes.h> @@ -42,24 +42,24 @@ static struct aws_error_info s_errors[] = { AWS_ERROR_EVENT_STREAM_MESSAGE_PARSER_ILLEGAL_STATE, "message parser encountered an illegal state", LIB_NAME), - AWS_DEFINE_ERROR_INFO( - AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED, - "event stream rpc connection has been closed", - LIB_NAME), - AWS_DEFINE_ERROR_INFO( - AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR, - "event stream rpc connection has encountered a protocol error", - LIB_NAME), - AWS_DEFINE_ERROR_INFO( - AWS_ERROR_EVENT_STREAM_RPC_STREAM_CLOSED, - "event stream rpc connection stream is closed.", - LIB_NAME), - AWS_DEFINE_ERROR_INFO( - AWS_ERROR_EVENT_STREAM_RPC_STREAM_NOT_ACTIVATED, - "event stream rpc stream continuation was not successfully activated before use. Call " - "aws_event_stream_rpc_client_continuation_activate()" - " before using a stream continuation token.", - LIB_NAME), + AWS_DEFINE_ERROR_INFO( + AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED, + "event stream rpc connection has been closed", + LIB_NAME), + AWS_DEFINE_ERROR_INFO( + AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR, + "event stream rpc connection has encountered a protocol error", + LIB_NAME), + AWS_DEFINE_ERROR_INFO( + AWS_ERROR_EVENT_STREAM_RPC_STREAM_CLOSED, + "event stream rpc connection stream is closed.", + LIB_NAME), + AWS_DEFINE_ERROR_INFO( + AWS_ERROR_EVENT_STREAM_RPC_STREAM_NOT_ACTIVATED, + "event stream rpc stream continuation was not successfully activated before use. Call " + "aws_event_stream_rpc_client_continuation_activate()" + " before using a stream continuation token.", + LIB_NAME), }; static struct aws_error_info_list s_list = { @@ -67,56 +67,56 @@ static struct aws_error_info_list s_list = { .count = sizeof(s_errors) / sizeof(struct aws_error_info), }; -static bool s_event_stream_library_initialized = false; - -static struct aws_log_subject_info s_event_stream_log_subject_infos[] = { - DEFINE_LOG_SUBJECT_INFO( - AWS_LS_EVENT_STREAM_GENERAL, - "event-stream-general", - "Subject for aws-c-event-stream logging that defies categorization."), - DEFINE_LOG_SUBJECT_INFO( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "event-stream-channel-handler", - "Subject for event-stream channel handler related logging."), - DEFINE_LOG_SUBJECT_INFO( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "event-stream-rpc-server", - "Subject for event-stream rpc server."), - DEFINE_LOG_SUBJECT_INFO( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "event-stream-rpc-client", - "Subject for event-stream rpc client."), -}; - -static struct aws_log_subject_info_list s_event_stream_log_subject_list = { - .subject_list = s_event_stream_log_subject_infos, - .count = AWS_ARRAY_SIZE(s_event_stream_log_subject_infos), -}; - -void aws_event_stream_library_init(struct aws_allocator *allocator) { - if (!s_event_stream_library_initialized) { - s_event_stream_library_initialized = true; - aws_io_library_init(allocator); +static bool s_event_stream_library_initialized = false; + +static struct aws_log_subject_info s_event_stream_log_subject_infos[] = { + DEFINE_LOG_SUBJECT_INFO( + AWS_LS_EVENT_STREAM_GENERAL, + "event-stream-general", + "Subject for aws-c-event-stream logging that defies categorization."), + DEFINE_LOG_SUBJECT_INFO( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "event-stream-channel-handler", + "Subject for event-stream channel handler related logging."), + DEFINE_LOG_SUBJECT_INFO( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "event-stream-rpc-server", + "Subject for event-stream rpc server."), + DEFINE_LOG_SUBJECT_INFO( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "event-stream-rpc-client", + "Subject for event-stream rpc client."), +}; + +static struct aws_log_subject_info_list s_event_stream_log_subject_list = { + .subject_list = s_event_stream_log_subject_infos, + .count = AWS_ARRAY_SIZE(s_event_stream_log_subject_infos), +}; + +void aws_event_stream_library_init(struct aws_allocator *allocator) { + if (!s_event_stream_library_initialized) { + s_event_stream_library_initialized = true; + aws_io_library_init(allocator); aws_register_error_info(&s_list); - aws_register_log_subject_info_list(&s_event_stream_log_subject_list); + aws_register_log_subject_info_list(&s_event_stream_log_subject_list); + } +} + +void aws_event_stream_library_clean_up(void) { + if (s_event_stream_library_initialized) { + s_event_stream_library_initialized = false; + aws_unregister_error_info(&s_list); + aws_io_library_clean_up(); } } -void aws_event_stream_library_clean_up(void) { - if (s_event_stream_library_initialized) { - s_event_stream_library_initialized = false; - aws_unregister_error_info(&s_list); - aws_io_library_clean_up(); - } -} - #define TOTAL_LEN_OFFSET 0 #define PRELUDE_CRC_OFFSET (sizeof(uint32_t) + sizeof(uint32_t)) #define HEADER_LEN_OFFSET sizeof(uint32_t) /* Computes the byte length necessary to store the headers represented in the headers list. * returns that length. */ -uint32_t aws_event_stream_compute_headers_required_buffer_len(const struct aws_array_list *headers) { +uint32_t aws_event_stream_compute_headers_required_buffer_len(const struct aws_array_list *headers) { if (!headers || !aws_array_list_length(headers)) { return 0; } @@ -148,7 +148,7 @@ uint32_t aws_event_stream_compute_headers_required_buffer_len(const struct aws_a /* adds the headers represented in the headers list to the buffer. returns the new buffer offset for use elsewhere. Assumes buffer length is at least the length of the return value from compute_headers_length() */ -size_t aws_event_stream_write_headers_to_buffer(const struct aws_array_list *headers, uint8_t *buffer) { +size_t aws_event_stream_write_headers_to_buffer(const struct aws_array_list *headers, uint8_t *buffer) { if (!headers || !aws_array_list_length(headers)) { return 0; } @@ -205,12 +205,12 @@ size_t aws_event_stream_write_headers_to_buffer(const struct aws_array_list *hea return buffer_alias - buffer; } -int aws_event_stream_read_headers_from_buffer( - struct aws_array_list *headers, - const uint8_t *buffer, - size_t headers_len) { +int aws_event_stream_read_headers_from_buffer( + struct aws_array_list *headers, + const uint8_t *buffer, + size_t headers_len) { - if (AWS_UNLIKELY(headers_len > AWS_EVENT_STREAM_MAX_HEADERS_SIZE)) { + if (AWS_UNLIKELY(headers_len > AWS_EVENT_STREAM_MAX_HEADERS_SIZE)) { return aws_raise_error(AWS_ERROR_EVENT_STREAM_MESSAGE_FIELD_SIZE_EXCEEDED); } @@ -291,9 +291,9 @@ int aws_event_stream_message_init( size_t payload_len = payload ? payload->len : 0; - uint32_t headers_length = aws_event_stream_compute_headers_required_buffer_len(headers); + uint32_t headers_length = aws_event_stream_compute_headers_required_buffer_len(headers); - if (AWS_UNLIKELY(headers_length > AWS_EVENT_STREAM_MAX_HEADERS_SIZE)) { + if (AWS_UNLIKELY(headers_length > AWS_EVENT_STREAM_MAX_HEADERS_SIZE)) { return aws_raise_error(AWS_ERROR_EVENT_STREAM_MESSAGE_FIELD_SIZE_EXCEEDED); } @@ -304,7 +304,7 @@ int aws_event_stream_message_init( return aws_raise_error(AWS_ERROR_OVERFLOW_DETECTED); } - if (AWS_UNLIKELY(total_length > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE)) { + if (AWS_UNLIKELY(total_length > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE)) { return aws_raise_error(AWS_ERROR_EVENT_STREAM_MESSAGE_FIELD_SIZE_EXCEEDED); } @@ -326,7 +326,7 @@ int aws_event_stream_message_init( buffer_offset += sizeof(running_crc); if (headers_length) { - buffer_offset += aws_event_stream_write_headers_to_buffer(headers, buffer_offset); + buffer_offset += aws_event_stream_write_headers_to_buffer(headers, buffer_offset); } if (payload) { @@ -349,7 +349,7 @@ int aws_event_stream_message_from_buffer( struct aws_event_stream_message *message, struct aws_allocator *alloc, struct aws_byte_buf *buffer) { - AWS_ASSERT(buffer); + AWS_ASSERT(buffer); message->alloc = alloc; message->owns_buffer = 0; @@ -364,7 +364,7 @@ int aws_event_stream_message_from_buffer( return aws_raise_error(AWS_ERROR_EVENT_STREAM_BUFFER_LENGTH_MISMATCH); } - if (AWS_UNLIKELY(message_length > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE)) { + if (AWS_UNLIKELY(message_length > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE)) { return aws_raise_error(AWS_ERROR_EVENT_STREAM_MESSAGE_FIELD_SIZE_EXCEEDED); } @@ -440,7 +440,7 @@ uint32_t aws_event_stream_message_prelude_crc(const struct aws_event_stream_mess } int aws_event_stream_message_headers(const struct aws_event_stream_message *message, struct aws_array_list *headers) { - return aws_event_stream_read_headers_from_buffer( + return aws_event_stream_read_headers_from_buffer( headers, message->message_buffer + AWS_EVENT_STREAM_PRELUDE_LENGTH, aws_event_stream_message_headers_len(message)); @@ -580,16 +580,16 @@ int aws_event_stream_message_to_debug_str(FILE *fd, const struct aws_event_strea } int aws_event_stream_headers_list_init(struct aws_array_list *headers, struct aws_allocator *allocator) { - AWS_ASSERT(headers); - AWS_ASSERT(allocator); + AWS_ASSERT(headers); + AWS_ASSERT(allocator); return aws_array_list_init_dynamic(headers, allocator, 4, sizeof(struct aws_event_stream_header_value_pair)); } void aws_event_stream_headers_list_cleanup(struct aws_array_list *headers) { - if (AWS_UNLIKELY(!headers || !aws_array_list_is_valid(headers))) { - return; - } + if (AWS_UNLIKELY(!headers || !aws_array_list_is_valid(headers))) { + return; + } for (size_t i = 0; i < aws_array_list_length(headers); ++i) { struct aws_event_stream_header_value_pair *header = NULL; @@ -652,43 +652,43 @@ int aws_event_stream_add_string_header( return s_add_variable_len_header(headers, &header, name, name_len, (uint8_t *)value, value_len, copy); } -struct aws_event_stream_header_value_pair aws_event_stream_create_string_header( - struct aws_byte_cursor name, - struct aws_byte_cursor value) { - AWS_PRECONDITION(name.len < INT8_MAX); - AWS_PRECONDITION(value.len < INT16_MAX); - - struct aws_event_stream_header_value_pair header = { - .header_value_type = AWS_EVENT_STREAM_HEADER_STRING, - .header_value.variable_len_val = value.ptr, - .header_value_len = (uint16_t)value.len, - .header_name_len = (uint8_t)name.len, - .value_owned = 0, - }; - - memcpy(header.header_name, name.ptr, name.len); - - return header; -} - -struct aws_event_stream_header_value_pair aws_event_stream_create_int32_header( - struct aws_byte_cursor name, - int32_t value) { - AWS_PRECONDITION(name.len < INT8_MAX); - - struct aws_event_stream_header_value_pair header = { - .header_value_type = AWS_EVENT_STREAM_HEADER_INT32, - .header_value_len = (uint16_t)sizeof(int32_t), - .header_name_len = (uint8_t)name.len, - .value_owned = 0, - }; - - memcpy(header.header_name, name.ptr, name.len); - aws_write_u32((uint32_t)value, header.header_value.static_val); - - return header; -} - +struct aws_event_stream_header_value_pair aws_event_stream_create_string_header( + struct aws_byte_cursor name, + struct aws_byte_cursor value) { + AWS_PRECONDITION(name.len < INT8_MAX); + AWS_PRECONDITION(value.len < INT16_MAX); + + struct aws_event_stream_header_value_pair header = { + .header_value_type = AWS_EVENT_STREAM_HEADER_STRING, + .header_value.variable_len_val = value.ptr, + .header_value_len = (uint16_t)value.len, + .header_name_len = (uint8_t)name.len, + .value_owned = 0, + }; + + memcpy(header.header_name, name.ptr, name.len); + + return header; +} + +struct aws_event_stream_header_value_pair aws_event_stream_create_int32_header( + struct aws_byte_cursor name, + int32_t value) { + AWS_PRECONDITION(name.len < INT8_MAX); + + struct aws_event_stream_header_value_pair header = { + .header_value_type = AWS_EVENT_STREAM_HEADER_INT32, + .header_value_len = (uint16_t)sizeof(int32_t), + .header_name_len = (uint8_t)name.len, + .value_owned = 0, + }; + + memcpy(header.header_name, name.ptr, name.len); + aws_write_u32((uint32_t)value, header.header_value.static_val); + + return header; +} + int aws_event_stream_add_byte_header(struct aws_array_list *headers, const char *name, uint8_t name_len, int8_t value) { struct aws_event_stream_header_value_pair header = {.header_name_len = name_len, .header_value_len = 1, @@ -1226,8 +1226,8 @@ static int s_verify_prelude_state( if (AWS_LIKELY(decoder->running_crc == decoder->prelude.prelude_crc)) { if (AWS_UNLIKELY( - decoder->prelude.headers_len > AWS_EVENT_STREAM_MAX_HEADERS_SIZE || - decoder->prelude.total_len > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE)) { + decoder->prelude.headers_len > AWS_EVENT_STREAM_MAX_HEADERS_SIZE || + decoder->prelude.total_len > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE)) { aws_raise_error(AWS_ERROR_EVENT_STREAM_MESSAGE_FIELD_SIZE_EXCEEDED); char error_message[] = "Maximum message field size exceeded"; @@ -1240,10 +1240,10 @@ static int s_verify_prelude_state( return AWS_OP_ERR; } - /* Should only call on_prelude() after passing crc check and limitation check, otherwise call on_prelude() with - * incorrect prelude is error prune. */ - decoder->on_prelude(decoder, &decoder->prelude, decoder->user_context); - + /* Should only call on_prelude() after passing crc check and limitation check, otherwise call on_prelude() with + * incorrect prelude is error prune. */ + decoder->on_prelude(decoder, &decoder->prelude, decoder->user_context); + decoder->running_crc = aws_checksums_crc32( decoder->working_buffer + PRELUDE_CRC_OFFSET, (int)sizeof(decoder->prelude.prelude_crc), diff --git a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_channel_handler.c b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_channel_handler.c index e3c21f5aa4..03b3129621 100644 --- a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_channel_handler.c +++ b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_channel_handler.c @@ -1,547 +1,547 @@ -/* - * Copyright 2010-2020 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. - */ - -#include <aws/event-stream/event_stream.h> -#include <aws/event-stream/event_stream_channel_handler.h> - -#include <aws/checksums/crc.h> - -#include <aws/io/channel.h> - -#include <inttypes.h> - -static const size_t s_default_payload_size = 1024; - -/* an event stream message has overhead of - * [msg len (uint32_t)] - * [headers len (uint32_t)] - * [prelude crc (uint32_t)] - * ... headers and payload .... - * [message crc (uint32_t)] - */ -static const size_t s_message_overhead_size = AWS_EVENT_STREAM_PRELUDE_LENGTH + AWS_EVENT_STREAM_TRAILER_LENGTH; - -struct aws_event_stream_channel_handler { - struct aws_channel_handler handler; - struct aws_byte_buf message_buf; - uint32_t running_crc; - uint32_t current_message_len; - aws_event_stream_channel_handler_on_message_received_fn *on_message_received; - void *user_data; - size_t initial_window_size; - bool manual_window_management; -}; - -static int s_process_read_message( - struct aws_channel_handler *handler, - struct aws_channel_slot *slot, - struct aws_io_message *message) { - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: received message of size %zu", - (void *)handler, - message->message_data.len); - struct aws_event_stream_channel_handler *event_stream_handler = handler->impl; - - struct aws_byte_cursor message_cursor = aws_byte_cursor_from_buf(&message->message_data); - - int error_code = AWS_ERROR_SUCCESS; - while (message_cursor.len) { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: processing chunk of size %zu", - (void *)handler, - message_cursor.len); - - /* first read only the prelude so we can do checks before reading the entire buffer. */ - if (event_stream_handler->message_buf.len < AWS_EVENT_STREAM_PRELUDE_LENGTH) { - size_t remaining_prelude = AWS_EVENT_STREAM_PRELUDE_LENGTH - event_stream_handler->message_buf.len; - size_t to_copy = aws_min_size(message_cursor.len, remaining_prelude); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: processing prelude, %zu bytes of an expected 12.", - (void *)handler, - to_copy); - - if (!aws_byte_buf_write(&event_stream_handler->message_buf, message_cursor.ptr, to_copy)) { - error_code = aws_last_error(); - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: writing to prelude buffer failed with error %s", - (void *)handler, - aws_error_debug_str(error_code)); - goto finished; - } - - aws_byte_cursor_advance(&message_cursor, to_copy); - } - - /* we need to get the prelude so we can get the message length to know how much to read and also - * to check the prelude CRC to protect against bit-flips causing us to read to much memory */ - if (event_stream_handler->message_buf.len == AWS_EVENT_STREAM_PRELUDE_LENGTH) { - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: processing prelude buffer", (void *)handler); - - struct aws_byte_cursor prelude_cursor = aws_byte_cursor_from_buf(&event_stream_handler->message_buf); - - event_stream_handler->running_crc = - aws_checksums_crc32(prelude_cursor.ptr, sizeof(uint32_t) + sizeof(uint32_t), 0); - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: calculated prelude CRC of %" PRIu32, - (void *)handler, - event_stream_handler->running_crc); - - aws_byte_cursor_read_be32(&prelude_cursor, &event_stream_handler->current_message_len); - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: read total message length of %" PRIu32, - (void *)handler, - event_stream_handler->current_message_len); - if (event_stream_handler->current_message_len > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: message length of %" PRIu32 " exceeds the max size of %zu", - (void *)handler, - event_stream_handler->current_message_len, - (size_t)AWS_EVENT_STREAM_MAX_MESSAGE_SIZE); - aws_raise_error(AWS_ERROR_EVENT_STREAM_MESSAGE_FIELD_SIZE_EXCEEDED); - error_code = aws_last_error(); - goto finished; - } - - /* advance past the headers field since we don't really care about it at this point */ - aws_byte_cursor_advance(&prelude_cursor, sizeof(uint32_t)); - - uint32_t prelude_crc = 0; - aws_byte_cursor_read_be32(&prelude_cursor, &prelude_crc); - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: read prelude CRC of %" PRIu32, - (void *)handler, - prelude_crc); - - /* make sure the checksum matches before processing any further */ - if (event_stream_handler->running_crc != prelude_crc) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: prelude CRC mismatch. calculated %" PRIu32 " but the crc for the message was %" PRIu32, - (void *)handler, - event_stream_handler->running_crc, - prelude_crc); - aws_raise_error(AWS_ERROR_EVENT_STREAM_PRELUDE_CHECKSUM_FAILURE); - error_code = aws_last_error(); - goto finished; - } - } - - /* read whatever is remaining from the message */ - if (event_stream_handler->message_buf.len < event_stream_handler->current_message_len) { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: processing remaining message buffer", (void *)handler); - size_t remaining = event_stream_handler->current_message_len - event_stream_handler->message_buf.len; - size_t to_copy = aws_min_size(message_cursor.len, remaining); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: of the remaining %zu, processing %zu from the " - "current message.", - (void *)handler, - remaining, - to_copy); - - struct aws_byte_cursor to_append = aws_byte_cursor_advance(&message_cursor, to_copy); - if (aws_byte_buf_append_dynamic(&event_stream_handler->message_buf, &to_append)) { - error_code = aws_last_error(); - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Appending to the message buffer failed with error %s.", - (void *)handler, - aws_error_debug_str(error_code)); - - goto finished; - } - } - - /* If we read the entire message, parse it and give it back to the subscriber. Keep in mind, once we're to this - * point the aws_event_stream API handles the rest of the message parsing and validation. */ - if (event_stream_handler->message_buf.len == event_stream_handler->current_message_len) { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: An entire message has been read. Parsing the message now.", - (void *)handler); - struct aws_event_stream_message received_message; - AWS_ZERO_STRUCT(received_message); - - if (aws_event_stream_message_from_buffer( - &received_message, event_stream_handler->handler.alloc, &event_stream_handler->message_buf)) { - error_code = aws_last_error(); - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Parsing the message failed with error %s.", - (void *)handler, - aws_error_debug_str(error_code)); - goto finished; - } - - size_t message_size = event_stream_handler->message_buf.len; - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: Invoking on_message_received callback.", (void *)handler); - event_stream_handler->on_message_received( - &received_message, AWS_ERROR_SUCCESS, event_stream_handler->user_data); - aws_event_stream_message_clean_up(&received_message); - event_stream_handler->current_message_len = 0; - event_stream_handler->running_crc = 0; - aws_byte_buf_reset(&event_stream_handler->message_buf, true); - - if (!event_stream_handler->manual_window_management) { - aws_channel_slot_increment_read_window(slot, message_size); - } - } - } - -finished: - if (error_code) { - event_stream_handler->on_message_received(NULL, error_code, event_stream_handler->user_data); - aws_channel_shutdown(slot->channel, error_code); - } - aws_mem_release(message->allocator, message); - return AWS_OP_SUCCESS; -} - -struct message_write_data { - struct aws_allocator *allocator; - struct aws_channel_task task; - struct aws_event_stream_channel_handler *handler; - struct aws_event_stream_message *message; - aws_event_stream_channel_handler_on_message_written_fn *on_message_written; - void *user_data; -}; - -static void s_on_message_write_completed_fn( - struct aws_channel *channel, - struct aws_io_message *message, - int err_code, - void *user_data) { - (void)channel; - (void)message; - - struct message_write_data *message_data = user_data; - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "channel=%p: Message write completed. Invoking " - "on_message_written callback.", - (void *)channel); - message_data->on_message_written(message_data->message, err_code, message_data->user_data); - aws_mem_release(message_data->allocator, message_data); -} - -static void s_write_handler_message(struct aws_channel_task *task, void *arg, enum aws_task_status status) { - (void)task; - - struct message_write_data *message_data = arg; - - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "static: Write message task invoked."); - if (status == AWS_TASK_STATUS_RUN_READY) { - struct aws_event_stream_message *message = message_data->message; - struct aws_event_stream_channel_handler *handler = message_data->handler; - - struct aws_byte_cursor message_cur = aws_byte_cursor_from_array( - aws_event_stream_message_buffer(message), aws_event_stream_message_total_length(message)); - - while (message_cur.len) { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: writing message chunk of size %zu.", - (void *)&handler->handler, - message_cur.len); - - /* io messages from the pool are allowed to be smaller than the requested size. */ - struct aws_io_message *io_message = aws_channel_acquire_message_from_pool( - handler->handler.slot->channel, AWS_IO_MESSAGE_APPLICATION_DATA, message_cur.len); - - if (!io_message) { - int error_code = aws_last_error(); - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Error occurred while acquiring io message %s.", - (void *)&handler->handler, - aws_error_debug_str(error_code)); - - message_data->on_message_written(message, error_code, message_data->user_data); - aws_mem_release(message_data->allocator, message_data); - aws_channel_shutdown(handler->handler.slot->channel, error_code); - break; - } - - aws_byte_buf_write_to_capacity(&io_message->message_data, &message_cur); - - /* if that was the end of the buffer we want to write, attach the completion callback to that io message */ - if (message_cur.len == 0) { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Message completely written to all io buffers.", - (void *)&handler->handler); - io_message->on_completion = s_on_message_write_completed_fn; - io_message->user_data = message_data; - } - - /* note if this fails the io message will not be queued and as a result will not have it's completion - * callback invoked. */ - if (aws_channel_slot_send_message(handler->handler.slot, io_message, AWS_CHANNEL_DIR_WRITE)) { - aws_mem_release(io_message->allocator, io_message); - int error_code = aws_last_error(); - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Error occurred while sending message to channel %s.", - (void *)&handler->handler, - aws_error_debug_str(error_code)); - message_data->on_message_written(message, error_code, message_data->user_data); - aws_mem_release(message_data->allocator, message_data); - aws_channel_shutdown(handler->handler.slot->channel, error_code); - break; - } - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: Message sent to channel", (void *)&handler->handler); - } - } else { - AWS_LOGF_WARN(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "static: Channel was shutdown. Message not sent"); - message_data->on_message_written( - message_data->message, AWS_ERROR_IO_OPERATION_CANCELLED, message_data->user_data); - aws_mem_release(message_data->allocator, message_data); - } -} - -int aws_event_stream_channel_handler_write_message( - struct aws_channel_handler *channel_handler, - struct aws_event_stream_message *message, - aws_event_stream_channel_handler_on_message_written_fn *on_message_written, - void *user_data) { - AWS_PRECONDITION(channel_handler); - AWS_PRECONDITION(message); - AWS_PRECONDITION(on_message_written); - - struct aws_event_stream_channel_handler *handler = channel_handler->impl; - - struct message_write_data *write_data = - aws_mem_calloc(handler->handler.alloc, 1, sizeof(struct message_write_data)); - - if (!write_data) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Error occurred while allocating callback data %s.", - (void *)channel_handler, - aws_error_debug_str(aws_last_error())); - aws_channel_shutdown(channel_handler->slot->channel, aws_last_error()); - return AWS_OP_ERR; - } - - write_data->handler = handler; - write_data->user_data = user_data; - write_data->message = message; - write_data->on_message_written = on_message_written; - write_data->allocator = handler->handler.alloc; - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: Scheduling message write task", (void *)channel_handler); - aws_channel_task_init( - &write_data->task, s_write_handler_message, write_data, "aws_event_stream_channel_handler_write_message"); - aws_channel_schedule_task_now(handler->handler.slot->channel, &write_data->task); - - return AWS_OP_SUCCESS; -} - -void *aws_event_stream_channel_handler_get_user_data(struct aws_channel_handler *channel_handler) { - struct aws_event_stream_channel_handler *handler = channel_handler->impl; - return handler->user_data; -} - -struct window_update_data { - struct aws_allocator *allocator; - struct aws_channel_task task; - struct aws_event_stream_channel_handler *handler; - size_t window_update_size; -}; - -static void s_update_window_task(struct aws_channel_task *task, void *arg, enum aws_task_status status) { - (void)task; - struct window_update_data *update_data = arg; - - if (status == AWS_TASK_STATUS_RUN_READY) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "static: updating window. increment of %zu", - update_data->window_update_size); - aws_channel_slot_increment_read_window(update_data->handler->handler.slot, update_data->window_update_size); - } - - aws_mem_release(update_data->allocator, update_data); -} - -void aws_event_stream_channel_handler_increment_read_window( - struct aws_channel_handler *channel_handler, - size_t window_update_size) { - AWS_PRECONDITION(channel_handler); - - struct aws_event_stream_channel_handler *handler = channel_handler->impl; - - if (!handler->manual_window_management) { - return; - } - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: A user requested window update and manual window management is specified. Updating size of %zu", - (void *)channel_handler, - window_update_size); - - if (aws_channel_thread_is_callers_thread(handler->handler.slot->channel)) { - if (aws_channel_slot_increment_read_window(handler->handler.slot, window_update_size)) { - aws_channel_shutdown(handler->handler.slot->channel, aws_last_error()); - return; - } - } - - struct window_update_data *update_data = - aws_mem_calloc(handler->handler.alloc, 1, sizeof(struct window_update_data)); - - if (!update_data) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Error occurred while allocating update window data %s.", - (void *)channel_handler, - aws_error_debug_str(aws_last_error())); - aws_channel_shutdown(handler->handler.slot->channel, aws_last_error()); - return; - } - - update_data->allocator = handler->handler.alloc; - update_data->handler = handler; - update_data->window_update_size = window_update_size; - - aws_channel_task_init( - &update_data->task, - s_update_window_task, - update_data, - "aws_event_stream_channel_handler_increment_read_window"); - aws_channel_schedule_task_now(handler->handler.slot->channel, &update_data->task); -} - -static int s_process_write_message( - struct aws_channel_handler *handler, - struct aws_channel_slot *slot, - struct aws_io_message *message) { - (void)handler; - (void)slot; - (void)message; - AWS_FATAL_ASSERT(!"The event-stream-channel-handler is not designed to be a mid-channel handler."); - return AWS_OP_ERR; -} - -static int s_increment_read_window(struct aws_channel_handler *handler, struct aws_channel_slot *slot, size_t size) { - (void)handler; - return aws_channel_slot_increment_read_window(slot, size); -} - -static size_t s_initial_window_size(struct aws_channel_handler *handler) { - struct aws_event_stream_channel_handler *message_handler = handler->impl; - return message_handler->initial_window_size; -} - -static size_t s_message_overhead(struct aws_channel_handler *handler) { - (void)handler; - return s_message_overhead_size; -} - -static void s_destroy(struct aws_channel_handler *handler) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: destroying event-stream message channel handler.", - (void *)handler); - struct aws_event_stream_channel_handler *event_stream_handler = handler->impl; - aws_byte_buf_clean_up(&event_stream_handler->message_buf); - aws_mem_release(handler->alloc, event_stream_handler); -} - -static int s_shutdown( - struct aws_channel_handler *handler, - struct aws_channel_slot *slot, - enum aws_channel_direction dir, - int error_code, - bool free_scarce_resources_immediately) { - (void)handler; - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: shutdown called on event-stream channel handler with error %s.", - (void *)handler, - aws_error_debug_str(error_code)); - - return aws_channel_slot_on_handler_shutdown_complete(slot, dir, error_code, free_scarce_resources_immediately); -} - -static struct aws_channel_handler_vtable vtable = { - .destroy = s_destroy, - .increment_read_window = s_increment_read_window, - .initial_window_size = s_initial_window_size, - .process_read_message = s_process_read_message, - .process_write_message = s_process_write_message, - .message_overhead = s_message_overhead, - .shutdown = s_shutdown, -}; - -struct aws_channel_handler *aws_event_stream_channel_handler_new( - struct aws_allocator *allocator, - const struct aws_event_stream_channel_handler_options *handler_options) { - AWS_PRECONDITION(allocator); - AWS_PRECONDITION(handler_options); - AWS_PRECONDITION(handler_options->on_message_received); - - AWS_LOGF_INFO(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "static: creating new event-stream message channel handler."); - - struct aws_event_stream_channel_handler *event_stream_handler = - aws_mem_calloc(allocator, 1, sizeof(struct aws_event_stream_channel_handler)); - - if (!event_stream_handler) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "static: Error occurred while allocating handler %s.", - aws_error_debug_str(aws_last_error())); - return NULL; - } - - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "static: new handler is %p", (void *)&event_stream_handler->handler); - - if (aws_byte_buf_init( - &event_stream_handler->message_buf, allocator, s_default_payload_size + s_message_overhead_size)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, - "id=%p: Error occurred while allocating scratch buffer %s.", - (void *)&event_stream_handler->handler, - aws_error_debug_str(aws_last_error())); - aws_mem_release(allocator, event_stream_handler); - return NULL; - } - - event_stream_handler->on_message_received = handler_options->on_message_received; - event_stream_handler->user_data = handler_options->user_data; - event_stream_handler->initial_window_size = - handler_options->initial_window_size > 0 ? handler_options->initial_window_size : SIZE_MAX; - event_stream_handler->manual_window_management = handler_options->manual_window_management; - event_stream_handler->handler.vtable = &vtable; - event_stream_handler->handler.alloc = allocator; - event_stream_handler->handler.impl = event_stream_handler; - - return &event_stream_handler->handler; -} +/* + * Copyright 2010-2020 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. + */ + +#include <aws/event-stream/event_stream.h> +#include <aws/event-stream/event_stream_channel_handler.h> + +#include <aws/checksums/crc.h> + +#include <aws/io/channel.h> + +#include <inttypes.h> + +static const size_t s_default_payload_size = 1024; + +/* an event stream message has overhead of + * [msg len (uint32_t)] + * [headers len (uint32_t)] + * [prelude crc (uint32_t)] + * ... headers and payload .... + * [message crc (uint32_t)] + */ +static const size_t s_message_overhead_size = AWS_EVENT_STREAM_PRELUDE_LENGTH + AWS_EVENT_STREAM_TRAILER_LENGTH; + +struct aws_event_stream_channel_handler { + struct aws_channel_handler handler; + struct aws_byte_buf message_buf; + uint32_t running_crc; + uint32_t current_message_len; + aws_event_stream_channel_handler_on_message_received_fn *on_message_received; + void *user_data; + size_t initial_window_size; + bool manual_window_management; +}; + +static int s_process_read_message( + struct aws_channel_handler *handler, + struct aws_channel_slot *slot, + struct aws_io_message *message) { + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: received message of size %zu", + (void *)handler, + message->message_data.len); + struct aws_event_stream_channel_handler *event_stream_handler = handler->impl; + + struct aws_byte_cursor message_cursor = aws_byte_cursor_from_buf(&message->message_data); + + int error_code = AWS_ERROR_SUCCESS; + while (message_cursor.len) { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: processing chunk of size %zu", + (void *)handler, + message_cursor.len); + + /* first read only the prelude so we can do checks before reading the entire buffer. */ + if (event_stream_handler->message_buf.len < AWS_EVENT_STREAM_PRELUDE_LENGTH) { + size_t remaining_prelude = AWS_EVENT_STREAM_PRELUDE_LENGTH - event_stream_handler->message_buf.len; + size_t to_copy = aws_min_size(message_cursor.len, remaining_prelude); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: processing prelude, %zu bytes of an expected 12.", + (void *)handler, + to_copy); + + if (!aws_byte_buf_write(&event_stream_handler->message_buf, message_cursor.ptr, to_copy)) { + error_code = aws_last_error(); + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: writing to prelude buffer failed with error %s", + (void *)handler, + aws_error_debug_str(error_code)); + goto finished; + } + + aws_byte_cursor_advance(&message_cursor, to_copy); + } + + /* we need to get the prelude so we can get the message length to know how much to read and also + * to check the prelude CRC to protect against bit-flips causing us to read to much memory */ + if (event_stream_handler->message_buf.len == AWS_EVENT_STREAM_PRELUDE_LENGTH) { + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: processing prelude buffer", (void *)handler); + + struct aws_byte_cursor prelude_cursor = aws_byte_cursor_from_buf(&event_stream_handler->message_buf); + + event_stream_handler->running_crc = + aws_checksums_crc32(prelude_cursor.ptr, sizeof(uint32_t) + sizeof(uint32_t), 0); + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: calculated prelude CRC of %" PRIu32, + (void *)handler, + event_stream_handler->running_crc); + + aws_byte_cursor_read_be32(&prelude_cursor, &event_stream_handler->current_message_len); + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: read total message length of %" PRIu32, + (void *)handler, + event_stream_handler->current_message_len); + if (event_stream_handler->current_message_len > AWS_EVENT_STREAM_MAX_MESSAGE_SIZE) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: message length of %" PRIu32 " exceeds the max size of %zu", + (void *)handler, + event_stream_handler->current_message_len, + (size_t)AWS_EVENT_STREAM_MAX_MESSAGE_SIZE); + aws_raise_error(AWS_ERROR_EVENT_STREAM_MESSAGE_FIELD_SIZE_EXCEEDED); + error_code = aws_last_error(); + goto finished; + } + + /* advance past the headers field since we don't really care about it at this point */ + aws_byte_cursor_advance(&prelude_cursor, sizeof(uint32_t)); + + uint32_t prelude_crc = 0; + aws_byte_cursor_read_be32(&prelude_cursor, &prelude_crc); + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: read prelude CRC of %" PRIu32, + (void *)handler, + prelude_crc); + + /* make sure the checksum matches before processing any further */ + if (event_stream_handler->running_crc != prelude_crc) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: prelude CRC mismatch. calculated %" PRIu32 " but the crc for the message was %" PRIu32, + (void *)handler, + event_stream_handler->running_crc, + prelude_crc); + aws_raise_error(AWS_ERROR_EVENT_STREAM_PRELUDE_CHECKSUM_FAILURE); + error_code = aws_last_error(); + goto finished; + } + } + + /* read whatever is remaining from the message */ + if (event_stream_handler->message_buf.len < event_stream_handler->current_message_len) { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: processing remaining message buffer", (void *)handler); + size_t remaining = event_stream_handler->current_message_len - event_stream_handler->message_buf.len; + size_t to_copy = aws_min_size(message_cursor.len, remaining); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: of the remaining %zu, processing %zu from the " + "current message.", + (void *)handler, + remaining, + to_copy); + + struct aws_byte_cursor to_append = aws_byte_cursor_advance(&message_cursor, to_copy); + if (aws_byte_buf_append_dynamic(&event_stream_handler->message_buf, &to_append)) { + error_code = aws_last_error(); + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Appending to the message buffer failed with error %s.", + (void *)handler, + aws_error_debug_str(error_code)); + + goto finished; + } + } + + /* If we read the entire message, parse it and give it back to the subscriber. Keep in mind, once we're to this + * point the aws_event_stream API handles the rest of the message parsing and validation. */ + if (event_stream_handler->message_buf.len == event_stream_handler->current_message_len) { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: An entire message has been read. Parsing the message now.", + (void *)handler); + struct aws_event_stream_message received_message; + AWS_ZERO_STRUCT(received_message); + + if (aws_event_stream_message_from_buffer( + &received_message, event_stream_handler->handler.alloc, &event_stream_handler->message_buf)) { + error_code = aws_last_error(); + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Parsing the message failed with error %s.", + (void *)handler, + aws_error_debug_str(error_code)); + goto finished; + } + + size_t message_size = event_stream_handler->message_buf.len; + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: Invoking on_message_received callback.", (void *)handler); + event_stream_handler->on_message_received( + &received_message, AWS_ERROR_SUCCESS, event_stream_handler->user_data); + aws_event_stream_message_clean_up(&received_message); + event_stream_handler->current_message_len = 0; + event_stream_handler->running_crc = 0; + aws_byte_buf_reset(&event_stream_handler->message_buf, true); + + if (!event_stream_handler->manual_window_management) { + aws_channel_slot_increment_read_window(slot, message_size); + } + } + } + +finished: + if (error_code) { + event_stream_handler->on_message_received(NULL, error_code, event_stream_handler->user_data); + aws_channel_shutdown(slot->channel, error_code); + } + aws_mem_release(message->allocator, message); + return AWS_OP_SUCCESS; +} + +struct message_write_data { + struct aws_allocator *allocator; + struct aws_channel_task task; + struct aws_event_stream_channel_handler *handler; + struct aws_event_stream_message *message; + aws_event_stream_channel_handler_on_message_written_fn *on_message_written; + void *user_data; +}; + +static void s_on_message_write_completed_fn( + struct aws_channel *channel, + struct aws_io_message *message, + int err_code, + void *user_data) { + (void)channel; + (void)message; + + struct message_write_data *message_data = user_data; + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "channel=%p: Message write completed. Invoking " + "on_message_written callback.", + (void *)channel); + message_data->on_message_written(message_data->message, err_code, message_data->user_data); + aws_mem_release(message_data->allocator, message_data); +} + +static void s_write_handler_message(struct aws_channel_task *task, void *arg, enum aws_task_status status) { + (void)task; + + struct message_write_data *message_data = arg; + + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "static: Write message task invoked."); + if (status == AWS_TASK_STATUS_RUN_READY) { + struct aws_event_stream_message *message = message_data->message; + struct aws_event_stream_channel_handler *handler = message_data->handler; + + struct aws_byte_cursor message_cur = aws_byte_cursor_from_array( + aws_event_stream_message_buffer(message), aws_event_stream_message_total_length(message)); + + while (message_cur.len) { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: writing message chunk of size %zu.", + (void *)&handler->handler, + message_cur.len); + + /* io messages from the pool are allowed to be smaller than the requested size. */ + struct aws_io_message *io_message = aws_channel_acquire_message_from_pool( + handler->handler.slot->channel, AWS_IO_MESSAGE_APPLICATION_DATA, message_cur.len); + + if (!io_message) { + int error_code = aws_last_error(); + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Error occurred while acquiring io message %s.", + (void *)&handler->handler, + aws_error_debug_str(error_code)); + + message_data->on_message_written(message, error_code, message_data->user_data); + aws_mem_release(message_data->allocator, message_data); + aws_channel_shutdown(handler->handler.slot->channel, error_code); + break; + } + + aws_byte_buf_write_to_capacity(&io_message->message_data, &message_cur); + + /* if that was the end of the buffer we want to write, attach the completion callback to that io message */ + if (message_cur.len == 0) { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Message completely written to all io buffers.", + (void *)&handler->handler); + io_message->on_completion = s_on_message_write_completed_fn; + io_message->user_data = message_data; + } + + /* note if this fails the io message will not be queued and as a result will not have it's completion + * callback invoked. */ + if (aws_channel_slot_send_message(handler->handler.slot, io_message, AWS_CHANNEL_DIR_WRITE)) { + aws_mem_release(io_message->allocator, io_message); + int error_code = aws_last_error(); + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Error occurred while sending message to channel %s.", + (void *)&handler->handler, + aws_error_debug_str(error_code)); + message_data->on_message_written(message, error_code, message_data->user_data); + aws_mem_release(message_data->allocator, message_data); + aws_channel_shutdown(handler->handler.slot->channel, error_code); + break; + } + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: Message sent to channel", (void *)&handler->handler); + } + } else { + AWS_LOGF_WARN(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "static: Channel was shutdown. Message not sent"); + message_data->on_message_written( + message_data->message, AWS_ERROR_IO_OPERATION_CANCELLED, message_data->user_data); + aws_mem_release(message_data->allocator, message_data); + } +} + +int aws_event_stream_channel_handler_write_message( + struct aws_channel_handler *channel_handler, + struct aws_event_stream_message *message, + aws_event_stream_channel_handler_on_message_written_fn *on_message_written, + void *user_data) { + AWS_PRECONDITION(channel_handler); + AWS_PRECONDITION(message); + AWS_PRECONDITION(on_message_written); + + struct aws_event_stream_channel_handler *handler = channel_handler->impl; + + struct message_write_data *write_data = + aws_mem_calloc(handler->handler.alloc, 1, sizeof(struct message_write_data)); + + if (!write_data) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Error occurred while allocating callback data %s.", + (void *)channel_handler, + aws_error_debug_str(aws_last_error())); + aws_channel_shutdown(channel_handler->slot->channel, aws_last_error()); + return AWS_OP_ERR; + } + + write_data->handler = handler; + write_data->user_data = user_data; + write_data->message = message; + write_data->on_message_written = on_message_written; + write_data->allocator = handler->handler.alloc; + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "id=%p: Scheduling message write task", (void *)channel_handler); + aws_channel_task_init( + &write_data->task, s_write_handler_message, write_data, "aws_event_stream_channel_handler_write_message"); + aws_channel_schedule_task_now(handler->handler.slot->channel, &write_data->task); + + return AWS_OP_SUCCESS; +} + +void *aws_event_stream_channel_handler_get_user_data(struct aws_channel_handler *channel_handler) { + struct aws_event_stream_channel_handler *handler = channel_handler->impl; + return handler->user_data; +} + +struct window_update_data { + struct aws_allocator *allocator; + struct aws_channel_task task; + struct aws_event_stream_channel_handler *handler; + size_t window_update_size; +}; + +static void s_update_window_task(struct aws_channel_task *task, void *arg, enum aws_task_status status) { + (void)task; + struct window_update_data *update_data = arg; + + if (status == AWS_TASK_STATUS_RUN_READY) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "static: updating window. increment of %zu", + update_data->window_update_size); + aws_channel_slot_increment_read_window(update_data->handler->handler.slot, update_data->window_update_size); + } + + aws_mem_release(update_data->allocator, update_data); +} + +void aws_event_stream_channel_handler_increment_read_window( + struct aws_channel_handler *channel_handler, + size_t window_update_size) { + AWS_PRECONDITION(channel_handler); + + struct aws_event_stream_channel_handler *handler = channel_handler->impl; + + if (!handler->manual_window_management) { + return; + } + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: A user requested window update and manual window management is specified. Updating size of %zu", + (void *)channel_handler, + window_update_size); + + if (aws_channel_thread_is_callers_thread(handler->handler.slot->channel)) { + if (aws_channel_slot_increment_read_window(handler->handler.slot, window_update_size)) { + aws_channel_shutdown(handler->handler.slot->channel, aws_last_error()); + return; + } + } + + struct window_update_data *update_data = + aws_mem_calloc(handler->handler.alloc, 1, sizeof(struct window_update_data)); + + if (!update_data) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Error occurred while allocating update window data %s.", + (void *)channel_handler, + aws_error_debug_str(aws_last_error())); + aws_channel_shutdown(handler->handler.slot->channel, aws_last_error()); + return; + } + + update_data->allocator = handler->handler.alloc; + update_data->handler = handler; + update_data->window_update_size = window_update_size; + + aws_channel_task_init( + &update_data->task, + s_update_window_task, + update_data, + "aws_event_stream_channel_handler_increment_read_window"); + aws_channel_schedule_task_now(handler->handler.slot->channel, &update_data->task); +} + +static int s_process_write_message( + struct aws_channel_handler *handler, + struct aws_channel_slot *slot, + struct aws_io_message *message) { + (void)handler; + (void)slot; + (void)message; + AWS_FATAL_ASSERT(!"The event-stream-channel-handler is not designed to be a mid-channel handler."); + return AWS_OP_ERR; +} + +static int s_increment_read_window(struct aws_channel_handler *handler, struct aws_channel_slot *slot, size_t size) { + (void)handler; + return aws_channel_slot_increment_read_window(slot, size); +} + +static size_t s_initial_window_size(struct aws_channel_handler *handler) { + struct aws_event_stream_channel_handler *message_handler = handler->impl; + return message_handler->initial_window_size; +} + +static size_t s_message_overhead(struct aws_channel_handler *handler) { + (void)handler; + return s_message_overhead_size; +} + +static void s_destroy(struct aws_channel_handler *handler) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: destroying event-stream message channel handler.", + (void *)handler); + struct aws_event_stream_channel_handler *event_stream_handler = handler->impl; + aws_byte_buf_clean_up(&event_stream_handler->message_buf); + aws_mem_release(handler->alloc, event_stream_handler); +} + +static int s_shutdown( + struct aws_channel_handler *handler, + struct aws_channel_slot *slot, + enum aws_channel_direction dir, + int error_code, + bool free_scarce_resources_immediately) { + (void)handler; + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: shutdown called on event-stream channel handler with error %s.", + (void *)handler, + aws_error_debug_str(error_code)); + + return aws_channel_slot_on_handler_shutdown_complete(slot, dir, error_code, free_scarce_resources_immediately); +} + +static struct aws_channel_handler_vtable vtable = { + .destroy = s_destroy, + .increment_read_window = s_increment_read_window, + .initial_window_size = s_initial_window_size, + .process_read_message = s_process_read_message, + .process_write_message = s_process_write_message, + .message_overhead = s_message_overhead, + .shutdown = s_shutdown, +}; + +struct aws_channel_handler *aws_event_stream_channel_handler_new( + struct aws_allocator *allocator, + const struct aws_event_stream_channel_handler_options *handler_options) { + AWS_PRECONDITION(allocator); + AWS_PRECONDITION(handler_options); + AWS_PRECONDITION(handler_options->on_message_received); + + AWS_LOGF_INFO(AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, "static: creating new event-stream message channel handler."); + + struct aws_event_stream_channel_handler *event_stream_handler = + aws_mem_calloc(allocator, 1, sizeof(struct aws_event_stream_channel_handler)); + + if (!event_stream_handler) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "static: Error occurred while allocating handler %s.", + aws_error_debug_str(aws_last_error())); + return NULL; + } + + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "static: new handler is %p", (void *)&event_stream_handler->handler); + + if (aws_byte_buf_init( + &event_stream_handler->message_buf, allocator, s_default_payload_size + s_message_overhead_size)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_CHANNEL_HANDLER, + "id=%p: Error occurred while allocating scratch buffer %s.", + (void *)&event_stream_handler->handler, + aws_error_debug_str(aws_last_error())); + aws_mem_release(allocator, event_stream_handler); + return NULL; + } + + event_stream_handler->on_message_received = handler_options->on_message_received; + event_stream_handler->user_data = handler_options->user_data; + event_stream_handler->initial_window_size = + handler_options->initial_window_size > 0 ? handler_options->initial_window_size : SIZE_MAX; + event_stream_handler->manual_window_management = handler_options->manual_window_management; + event_stream_handler->handler.vtable = &vtable; + event_stream_handler->handler.alloc = allocator; + event_stream_handler->handler.impl = event_stream_handler; + + return &event_stream_handler->handler; +} diff --git a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc.c b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc.c index b44ec7a68c..7cffee8e8a 100644 --- a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc.c +++ b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc.c @@ -1,112 +1,112 @@ -/** - * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. - * SPDX-License-Identifier: Apache-2.0. - */ -#include <aws/event-stream/event_stream_rpc.h> - -#include <inttypes.h> - -const struct aws_byte_cursor aws_event_stream_rpc_message_type_name = - AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(":message-type"); -const struct aws_byte_cursor aws_event_stream_rpc_message_flags_name = - AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(":message-flags"); -const struct aws_byte_cursor aws_event_stream_rpc_stream_id_name = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(":stream-id"); -const struct aws_byte_cursor aws_event_stream_rpc_operation_name = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("operation"); - -/* just a convenience function for fetching message metadata from the event stream headers on a single iteration. */ -int aws_event_stream_rpc_extract_message_metadata( - const struct aws_array_list *message_headers, - int32_t *stream_id, - int32_t *message_type, - int32_t *message_flags, - struct aws_byte_buf *operation_name) { - size_t length = aws_array_list_length(message_headers); - bool message_type_found = 0; - bool message_flags_found = 0; - bool stream_id_found = 0; - bool operation_name_found = 0; - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_GENERAL, "processing message headers for rpc protocol. %zu headers to process.", length); - - for (size_t i = 0; i < length; ++i) { - struct aws_event_stream_header_value_pair *header = NULL; - aws_array_list_get_at_ptr(message_headers, (void **)&header, i); - struct aws_byte_buf name_buf = aws_event_stream_header_name(header); - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "processing header name " PRInSTR, AWS_BYTE_BUF_PRI(name_buf)); - - /* check type first since that's cheaper than a string compare */ - if (header->header_value_type == AWS_EVENT_STREAM_HEADER_INT32) { - - struct aws_byte_buf stream_id_field = aws_byte_buf_from_array( - aws_event_stream_rpc_stream_id_name.ptr, aws_event_stream_rpc_stream_id_name.len); - if (aws_byte_buf_eq_ignore_case(&name_buf, &stream_id_field)) { - *stream_id = aws_event_stream_header_value_as_int32(header); - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "stream id header value %" PRId32, *stream_id); - stream_id_found += 1; - goto found; - } - - struct aws_byte_buf message_type_field = aws_byte_buf_from_array( - aws_event_stream_rpc_message_type_name.ptr, aws_event_stream_rpc_message_type_name.len); - if (aws_byte_buf_eq_ignore_case(&name_buf, &message_type_field)) { - *message_type = aws_event_stream_header_value_as_int32(header); - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "message type header value %" PRId32, *message_type); - message_type_found += 1; - goto found; - } - - struct aws_byte_buf message_flags_field = aws_byte_buf_from_array( - aws_event_stream_rpc_message_flags_name.ptr, aws_event_stream_rpc_message_flags_name.len); - if (aws_byte_buf_eq_ignore_case(&name_buf, &message_flags_field)) { - *message_flags = aws_event_stream_header_value_as_int32(header); - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "message flags header value %" PRId32, *message_flags); - message_flags_found += 1; - goto found; - } - } - - if (header->header_value_type == AWS_EVENT_STREAM_HEADER_STRING) { - struct aws_byte_buf operation_field = aws_byte_buf_from_array( - aws_event_stream_rpc_operation_name.ptr, aws_event_stream_rpc_operation_name.len); - - if (aws_byte_buf_eq_ignore_case(&name_buf, &operation_field)) { - *operation_name = aws_event_stream_header_value_as_string(header); - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_GENERAL, - "operation name header value" PRInSTR, - AWS_BYTE_BUF_PRI(*operation_name)); - operation_name_found += 1; - goto found; - } - } - - continue; - - found: - if (message_flags_found && message_type_found && stream_id_found && operation_name_found) { - return AWS_OP_SUCCESS; - } - } - - return message_flags_found && message_type_found && stream_id_found - ? AWS_OP_SUCCESS - : aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); -} - -static const uint32_t s_bit_scrambling_magic = 0x45d9f3bU; -static const uint32_t s_bit_shift_magic = 16U; - -/* this is a repurposed hash function based on the technique in splitmix64. The magic number was a result of numerical - * analysis on maximum bit entropy. */ -uint64_t aws_event_stream_rpc_hash_streamid(const void *to_hash) { - uint32_t int_to_hash = *(const uint32_t *)to_hash; - uint32_t hash = ((int_to_hash >> s_bit_shift_magic) ^ int_to_hash) * s_bit_scrambling_magic; - hash = ((hash >> s_bit_shift_magic) ^ hash) * s_bit_scrambling_magic; - hash = (hash >> s_bit_shift_magic) ^ hash; - return (uint64_t)hash; -} - -bool aws_event_stream_rpc_streamid_eq(const void *a, const void *b) { - return *(const uint32_t *)a == *(const uint32_t *)b; -} +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#include <aws/event-stream/event_stream_rpc.h> + +#include <inttypes.h> + +const struct aws_byte_cursor aws_event_stream_rpc_message_type_name = + AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(":message-type"); +const struct aws_byte_cursor aws_event_stream_rpc_message_flags_name = + AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(":message-flags"); +const struct aws_byte_cursor aws_event_stream_rpc_stream_id_name = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(":stream-id"); +const struct aws_byte_cursor aws_event_stream_rpc_operation_name = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("operation"); + +/* just a convenience function for fetching message metadata from the event stream headers on a single iteration. */ +int aws_event_stream_rpc_extract_message_metadata( + const struct aws_array_list *message_headers, + int32_t *stream_id, + int32_t *message_type, + int32_t *message_flags, + struct aws_byte_buf *operation_name) { + size_t length = aws_array_list_length(message_headers); + bool message_type_found = 0; + bool message_flags_found = 0; + bool stream_id_found = 0; + bool operation_name_found = 0; + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_GENERAL, "processing message headers for rpc protocol. %zu headers to process.", length); + + for (size_t i = 0; i < length; ++i) { + struct aws_event_stream_header_value_pair *header = NULL; + aws_array_list_get_at_ptr(message_headers, (void **)&header, i); + struct aws_byte_buf name_buf = aws_event_stream_header_name(header); + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "processing header name " PRInSTR, AWS_BYTE_BUF_PRI(name_buf)); + + /* check type first since that's cheaper than a string compare */ + if (header->header_value_type == AWS_EVENT_STREAM_HEADER_INT32) { + + struct aws_byte_buf stream_id_field = aws_byte_buf_from_array( + aws_event_stream_rpc_stream_id_name.ptr, aws_event_stream_rpc_stream_id_name.len); + if (aws_byte_buf_eq_ignore_case(&name_buf, &stream_id_field)) { + *stream_id = aws_event_stream_header_value_as_int32(header); + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "stream id header value %" PRId32, *stream_id); + stream_id_found += 1; + goto found; + } + + struct aws_byte_buf message_type_field = aws_byte_buf_from_array( + aws_event_stream_rpc_message_type_name.ptr, aws_event_stream_rpc_message_type_name.len); + if (aws_byte_buf_eq_ignore_case(&name_buf, &message_type_field)) { + *message_type = aws_event_stream_header_value_as_int32(header); + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "message type header value %" PRId32, *message_type); + message_type_found += 1; + goto found; + } + + struct aws_byte_buf message_flags_field = aws_byte_buf_from_array( + aws_event_stream_rpc_message_flags_name.ptr, aws_event_stream_rpc_message_flags_name.len); + if (aws_byte_buf_eq_ignore_case(&name_buf, &message_flags_field)) { + *message_flags = aws_event_stream_header_value_as_int32(header); + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_GENERAL, "message flags header value %" PRId32, *message_flags); + message_flags_found += 1; + goto found; + } + } + + if (header->header_value_type == AWS_EVENT_STREAM_HEADER_STRING) { + struct aws_byte_buf operation_field = aws_byte_buf_from_array( + aws_event_stream_rpc_operation_name.ptr, aws_event_stream_rpc_operation_name.len); + + if (aws_byte_buf_eq_ignore_case(&name_buf, &operation_field)) { + *operation_name = aws_event_stream_header_value_as_string(header); + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_GENERAL, + "operation name header value" PRInSTR, + AWS_BYTE_BUF_PRI(*operation_name)); + operation_name_found += 1; + goto found; + } + } + + continue; + + found: + if (message_flags_found && message_type_found && stream_id_found && operation_name_found) { + return AWS_OP_SUCCESS; + } + } + + return message_flags_found && message_type_found && stream_id_found + ? AWS_OP_SUCCESS + : aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); +} + +static const uint32_t s_bit_scrambling_magic = 0x45d9f3bU; +static const uint32_t s_bit_shift_magic = 16U; + +/* this is a repurposed hash function based on the technique in splitmix64. The magic number was a result of numerical + * analysis on maximum bit entropy. */ +uint64_t aws_event_stream_rpc_hash_streamid(const void *to_hash) { + uint32_t int_to_hash = *(const uint32_t *)to_hash; + uint32_t hash = ((int_to_hash >> s_bit_shift_magic) ^ int_to_hash) * s_bit_scrambling_magic; + hash = ((hash >> s_bit_shift_magic) ^ hash) * s_bit_scrambling_magic; + hash = (hash >> s_bit_shift_magic) ^ hash; + return (uint64_t)hash; +} + +bool aws_event_stream_rpc_streamid_eq(const void *a, const void *b) { + return *(const uint32_t *)a == *(const uint32_t *)b; +} diff --git a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_client.c b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_client.c index 8f5de22f04..c2977f58a7 100644 --- a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_client.c +++ b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_client.c @@ -1,1040 +1,1040 @@ -/** - * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. - * SPDX-License-Identifier: Apache-2.0. - */ -#include <aws/event-stream/event_stream_channel_handler.h> -#include <aws/event-stream/event_stream_rpc_client.h> -#include <aws/event-stream/private/event_stream_rpc_priv.h> - -#include <aws/common/atomics.h> -#include <aws/common/hash_table.h> -#include <aws/common/mutex.h> - -#include <aws/io/channel_bootstrap.h> - -#include <inttypes.h> - -#ifdef _MSC_VER -/* allow declared initializer using address of automatic variable */ -# pragma warning(disable : 4221) -/* allow non-constant aggregate initializers */ -# pragma warning(disable : 4204) - -#endif - -static void s_clear_continuation_table(struct aws_event_stream_rpc_client_connection *connection); - -struct aws_event_stream_rpc_client_connection { - struct aws_allocator *allocator; - struct aws_hash_table continuation_table; - struct aws_client_bootstrap *bootstrap_ref; - struct aws_atomic_var ref_count; - struct aws_channel *channel; - struct aws_channel_handler *event_stream_handler; - uint32_t latest_stream_id; - struct aws_mutex stream_lock; - struct aws_atomic_var is_open; - struct aws_atomic_var handshake_state; - size_t initial_window_size; - aws_event_stream_rpc_client_on_connection_setup_fn *on_connection_setup; - aws_event_stream_rpc_client_connection_protocol_message_fn *on_connection_protocol_message; - aws_event_stream_rpc_client_on_connection_shutdown_fn *on_connection_shutdown; - void *user_data; - bool bootstrap_owned; - bool enable_read_back_pressure; -}; - -struct aws_event_stream_rpc_client_continuation_token { - uint32_t stream_id; - struct aws_event_stream_rpc_client_connection *connection; - aws_event_stream_rpc_client_stream_continuation_fn *continuation_fn; - aws_event_stream_rpc_client_stream_continuation_closed_fn *closed_fn; - void *user_data; - struct aws_atomic_var ref_count; - struct aws_atomic_var is_closed; -}; - -static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data); - -static int s_create_connection_on_channel( - struct aws_event_stream_rpc_client_connection *connection, - struct aws_channel *channel) { - struct aws_channel_handler *event_stream_handler = NULL; - struct aws_channel_slot *slot = NULL; - - struct aws_event_stream_channel_handler_options handler_options = { - .on_message_received = s_on_message_received, - .user_data = connection, - .initial_window_size = connection->initial_window_size, - .manual_window_management = connection->enable_read_back_pressure, - }; - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: creating an event-stream handler on channel %p", - (void *)connection, - (void *)channel); - event_stream_handler = aws_event_stream_channel_handler_new(connection->allocator, &handler_options); - - if (!event_stream_handler) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: creating an event-stream handler failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - slot = aws_channel_slot_new(channel); - - if (!slot) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: creating channel slot failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - aws_channel_slot_insert_end(channel, slot); - if (aws_channel_slot_set_handler(slot, event_stream_handler)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: setting handler on channel slot failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - connection->event_stream_handler = event_stream_handler; - connection->channel = channel; - aws_channel_acquire_hold(channel); - - return AWS_OP_SUCCESS; - -error: - if (!slot && event_stream_handler) { - aws_channel_handler_destroy(event_stream_handler); - } - - return AWS_OP_ERR; -} - -static void s_on_channel_setup_fn( - struct aws_client_bootstrap *bootstrap, - int error_code, - struct aws_channel *channel, - void *user_data) { - (void)bootstrap; - - struct aws_event_stream_rpc_client_connection *connection = user_data; - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: on_channel_setup_fn invoked with error_code %d with channel %p", - (void *)connection, - error_code, - (void *)channel); - - if (!error_code) { - connection->bootstrap_owned = true; - if (s_create_connection_on_channel(connection, channel)) { - int last_error = aws_last_error(); - connection->on_connection_setup(NULL, last_error, connection->user_data); - aws_channel_shutdown(channel, last_error); - return; - } - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: successful event-stream channel setup %p", - (void *)connection, - (void *)channel); - aws_event_stream_rpc_client_connection_acquire(connection); - connection->on_connection_setup(connection, AWS_OP_SUCCESS, connection->user_data); - aws_event_stream_rpc_client_connection_release(connection); - } else { - connection->on_connection_setup(NULL, error_code, connection->user_data); - aws_event_stream_rpc_client_connection_release(connection); - } -} - -static void s_on_channel_shutdown_fn( - struct aws_client_bootstrap *bootstrap, - int error_code, - struct aws_channel *channel, - void *user_data) { - (void)bootstrap; - - struct aws_event_stream_rpc_client_connection *connection = user_data; - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: on_channel_shutdown_fn invoked with error_code %d with channel %p", - (void *)connection, - error_code, - (void *)channel); - aws_atomic_store_int(&connection->is_open, 0u); - - if (connection->bootstrap_owned) { - s_clear_continuation_table(connection); - - aws_event_stream_rpc_client_connection_acquire(connection); - connection->on_connection_shutdown(connection, error_code, connection->user_data); - aws_event_stream_rpc_client_connection_release(connection); - } - - aws_channel_release_hold(channel); - aws_event_stream_rpc_client_connection_release(connection); -} - -/* Set each continuation's is_closed=true. - * A lock MUST be held while calling this. - * For use with aws_hash_table_foreach(). */ -static int s_mark_each_continuation_closed(void *context, struct aws_hash_element *p_element) { - (void)context; - struct aws_event_stream_rpc_client_continuation_token *continuation = p_element->value; - - aws_atomic_store_int(&continuation->is_closed, 1U); - - return AWS_COMMON_HASH_TABLE_ITER_CONTINUE; -} - -/* Invoke continuation's on_closed() callback. - * A lock must NOT be hold while calling this */ -static void s_complete_continuation(struct aws_event_stream_rpc_client_continuation_token *token) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "token=%p: token with stream-id %" PRIu32 ", purged from the stream table", - (void *)token, - token->stream_id); - - if (token->stream_id) { - token->closed_fn(token, token->user_data); - } - - aws_event_stream_rpc_client_continuation_release(token); -} - -static int s_complete_and_clear_each_continuation(void *context, struct aws_hash_element *p_element) { - (void)context; - struct aws_event_stream_rpc_client_continuation_token *continuation = p_element->value; - - s_complete_continuation(continuation); - - return AWS_COMMON_HASH_TABLE_ITER_DELETE | AWS_COMMON_HASH_TABLE_ITER_CONTINUE; -} - -/* Remove each continuation from hash-table and invoke its on_closed() callback. - * The connection->is_open must be set false before calling this. */ -static void s_clear_continuation_table(struct aws_event_stream_rpc_client_connection *connection) { - AWS_ASSERT(!aws_event_stream_rpc_client_connection_is_open(connection)); - - /* Use lock to ensure synchronization with code that adds entries to table. - * Since connection was just marked closed, no further entries will be - * added to table once we acquire the lock. */ - aws_mutex_lock(&connection->stream_lock); - aws_hash_table_foreach(&connection->continuation_table, s_mark_each_continuation_closed, NULL); - aws_mutex_unlock(&connection->stream_lock); - - /* Now release lock before invoking callbacks. - * It's safe to alter the table now without a lock, since no further - * entries can be added, and we've gone through the critical section - * above to ensure synchronization */ - aws_hash_table_foreach(&connection->continuation_table, s_complete_and_clear_each_continuation, NULL); -} - -int aws_event_stream_rpc_client_connection_connect( - struct aws_allocator *allocator, - const struct aws_event_stream_rpc_client_connection_options *conn_options) { - AWS_PRECONDITION(allocator); - AWS_PRECONDITION(conn_options); - AWS_PRECONDITION(conn_options->on_connection_protocol_message); - AWS_PRECONDITION(conn_options->on_connection_setup); - AWS_PRECONDITION(conn_options->on_connection_shutdown); - - struct aws_event_stream_rpc_client_connection *connection = - aws_mem_calloc(allocator, 1, sizeof(struct aws_event_stream_rpc_client_connection)); - - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: creating new connection", (void *)connection); - - if (!connection) { - return AWS_OP_ERR; - } - - connection->allocator = allocator; - aws_atomic_init_int(&connection->ref_count, 1); - connection->bootstrap_ref = conn_options->bootstrap; - /* this is released in the connection release which gets called regardless of if this function is successful or - * not*/ - aws_client_bootstrap_acquire(connection->bootstrap_ref); - aws_atomic_init_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_INITIALIZED); - aws_atomic_init_int(&connection->is_open, 1); - aws_mutex_init(&connection->stream_lock); - - connection->on_connection_shutdown = conn_options->on_connection_shutdown; - connection->on_connection_protocol_message = conn_options->on_connection_protocol_message; - connection->on_connection_setup = conn_options->on_connection_setup; - connection->user_data = conn_options->user_data; - - if (aws_hash_table_init( - &connection->continuation_table, - allocator, - 64, - aws_event_stream_rpc_hash_streamid, - aws_event_stream_rpc_streamid_eq, - NULL, - NULL)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: failed initializing continuation table with error %s.", - (void *)connection, - aws_error_debug_str(aws_last_error())); - - goto error; - } - - struct aws_socket_channel_bootstrap_options bootstrap_options = { - .bootstrap = connection->bootstrap_ref, - .tls_options = conn_options->tls_options, - .socket_options = conn_options->socket_options, - .user_data = connection, - .host_name = conn_options->host_name, - .port = conn_options->port, - .enable_read_back_pressure = false, - .setup_callback = s_on_channel_setup_fn, - .shutdown_callback = s_on_channel_shutdown_fn, - }; - - if (aws_client_bootstrap_new_socket_channel(&bootstrap_options)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: failed creating new socket channel with error %s.", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - return AWS_OP_SUCCESS; - -error: - aws_event_stream_rpc_client_connection_release(connection); - return AWS_OP_ERR; -} - -void aws_event_stream_rpc_client_connection_acquire(const struct aws_event_stream_rpc_client_connection *connection) { - AWS_PRECONDITION(connection); - size_t current_count = aws_atomic_fetch_add_explicit( - &((struct aws_event_stream_rpc_client_connection *)connection)->ref_count, 1, aws_memory_order_relaxed); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: connection acquired, new ref count is %zu.", - (void *)connection, - current_count + 1); -} - -static void s_destroy_connection(struct aws_event_stream_rpc_client_connection *connection) { - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: destroying connection.", (void *)connection); - aws_hash_table_clean_up(&connection->continuation_table); - aws_client_bootstrap_release(connection->bootstrap_ref); - aws_mem_release(connection->allocator, connection); -} - -void aws_event_stream_rpc_client_connection_release(const struct aws_event_stream_rpc_client_connection *connection) { - if (!connection) { - return; - } - - struct aws_event_stream_rpc_client_connection *connection_mut = - (struct aws_event_stream_rpc_client_connection *)connection; - size_t ref_count = aws_atomic_fetch_sub_explicit(&connection_mut->ref_count, 1, aws_memory_order_seq_cst); - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: connection released, new ref count is %zu.", - (void *)connection, - ref_count - 1); - - if (ref_count == 1) { - s_destroy_connection(connection_mut); - } -} - -void aws_event_stream_rpc_client_connection_close( - struct aws_event_stream_rpc_client_connection *connection, - int shutdown_error_code) { - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: connection close invoked with reason %s.", - (void *)connection, - aws_error_debug_str(shutdown_error_code)); - - size_t expect_open = 1U; - if (aws_atomic_compare_exchange_int(&connection->is_open, &expect_open, 0U)) { - aws_channel_shutdown(connection->channel, shutdown_error_code); - - if (!connection->bootstrap_owned) { - s_clear_continuation_table(connection); - - aws_event_stream_rpc_client_connection_release(connection); - } - } else { - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: connection already closed.", (void *)connection); - } -} - -bool aws_event_stream_rpc_client_connection_is_open(const struct aws_event_stream_rpc_client_connection *connection) { - return aws_atomic_load_int(&connection->is_open) == 1U; -} - -struct event_stream_connection_send_message_args { - struct aws_allocator *allocator; - struct aws_event_stream_message message; - enum aws_event_stream_rpc_message_type message_type; - struct aws_event_stream_rpc_client_connection *connection; - struct aws_event_stream_rpc_client_continuation_token *continuation; - aws_event_stream_rpc_client_message_flush_fn *flush_fn; - void *user_data; - bool end_stream; - bool terminate_connection; -}; - -static void s_on_protocol_message_written_fn( - struct aws_event_stream_message *message, - int error_code, - void *user_data) { - (void)message; - - struct event_stream_connection_send_message_args *message_args = user_data; - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: message %p flushed to channel.", - (void *)message_args->connection, - (void *)message); - - if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: connect message flushed to the wire, waiting on connect ack.", - (void *)message_args->connection); - aws_atomic_store_int(&message_args->connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED); - } - - if (message_args->end_stream) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: the end stream flag was set, closing continuation %p.", - (void *)message_args->connection, - (void *)message_args->continuation); - AWS_FATAL_ASSERT(message_args->continuation && "end stream flag was set but it wasn't on a continuation"); - aws_atomic_store_int(&message_args->continuation->is_closed, 1U); - - aws_mutex_lock(&message_args->connection->stream_lock); - aws_hash_table_remove( - &message_args->connection->continuation_table, &message_args->continuation->stream_id, NULL, NULL); - aws_mutex_unlock(&message_args->connection->stream_lock); - - /* Lock must NOT be held while invoking callback */ - s_complete_continuation(message_args->continuation); - } - - message_args->flush_fn(error_code, message_args->user_data); - - if (message_args->terminate_connection) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: terminate_connection flag was specified. Shutting down the connection.", - (void *)message_args->connection); - aws_event_stream_rpc_client_connection_close(message_args->connection, AWS_ERROR_SUCCESS); - } - - aws_event_stream_rpc_client_connection_release(message_args->connection); - - if (message_args->continuation) { - aws_event_stream_rpc_client_continuation_release(message_args->continuation); - } - - aws_event_stream_message_clean_up(&message_args->message); - aws_mem_release(message_args->allocator, message_args); -} - -static int s_send_protocol_message( - struct aws_event_stream_rpc_client_connection *connection, - struct aws_event_stream_rpc_client_continuation_token *continuation, - struct aws_byte_cursor *operation_name, - const struct aws_event_stream_rpc_message_args *message_args, - int32_t stream_id, - aws_event_stream_rpc_client_message_flush_fn *flush_fn, - void *user_data) { - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: sending message. continuation: %p, stream id %" PRId32, - (void *)connection, - (void *)continuation, - stream_id); - - size_t connect_handshake_state = aws_atomic_load_int(&connection->handshake_state); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: handshake completion value %zu", - (void *)connection, - connect_handshake_state); - - /* handshake step 1 is a connect message being received. Handshake 2 is the connect ack being sent. - * no messages other than connect and connect ack are allowed until this count reaches 2. */ - if (connect_handshake_state != CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && - message_args->message_type < AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: handshake not completed, only a connect message can be sent.", - (void *)connection); - return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - } - - struct event_stream_connection_send_message_args *args = - aws_mem_calloc(connection->allocator, 1, sizeof(struct event_stream_connection_send_message_args)); - - if (!message_args) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: failed to allocate callback arguments %s.", - (void *)connection, - aws_error_debug_str(aws_last_error())); - return AWS_OP_ERR; - } - - args->allocator = connection->allocator; - args->user_data = user_data; - args->message_type = message_args->message_type; - args->connection = connection; - args->flush_fn = flush_fn; - - if (continuation) { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: sending message on continuation %p", - (void *)connection, - (void *)continuation); - args->continuation = continuation; - aws_event_stream_rpc_client_continuation_acquire(continuation); - - if (message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p:end stream flag was specified on continuation %p", - (void *)connection, - (void *)continuation); - args->end_stream = true; - } - } - - if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK && - !(message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_CONNECTION_ACCEPTED)) { - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: terminating connection", (void *)connection); - args->terminate_connection = true; - } - - if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: sending connect message", (void *)connection); - aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED); - } - - args->flush_fn = flush_fn; - - size_t headers_count = operation_name ? message_args->headers_count + 4 : message_args->headers_count + 3; - struct aws_array_list headers_list; - AWS_ZERO_STRUCT(headers_list); - - if (aws_array_list_init_dynamic( - &headers_list, connection->allocator, headers_count, sizeof(struct aws_event_stream_header_value_pair))) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: an error occurred while initializing the headers list %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto args_allocated_before_failure; - } - - /* since we preallocated the space for the headers, these can't fail, but we'll go ahead an assert on them just in - * case */ - for (size_t i = 0; i < message_args->headers_count; ++i) { - AWS_FATAL_ASSERT(!aws_array_list_push_back(&headers_list, &message_args->headers[i])); - } - - AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( - &headers_list, - (const char *)aws_event_stream_rpc_message_type_name.ptr, - (uint8_t)aws_event_stream_rpc_message_type_name.len, - message_args->message_type)); - AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( - &headers_list, - (const char *)aws_event_stream_rpc_message_flags_name.ptr, - (uint8_t)aws_event_stream_rpc_message_flags_name.len, - message_args->message_flags)); - AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( - &headers_list, - (const char *)aws_event_stream_rpc_stream_id_name.ptr, - (uint8_t)aws_event_stream_rpc_stream_id_name.len, - stream_id)); - - if (operation_name) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: operation name specified " PRInSTR, - (void *)connection, - AWS_BYTE_CURSOR_PRI(*operation_name)); - AWS_FATAL_ASSERT(!aws_event_stream_add_string_header( - &headers_list, - (const char *)aws_event_stream_rpc_operation_name.ptr, - (uint8_t)aws_event_stream_rpc_operation_name.len, - (const char *)operation_name->ptr, - (uint16_t)operation_name->len, - 0)); - } - - int message_init_err_code = - aws_event_stream_message_init(&args->message, connection->allocator, &headers_list, message_args->payload); - aws_array_list_clean_up(&headers_list); - - if (message_init_err_code) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: message init failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto args_allocated_before_failure; - } - - aws_event_stream_rpc_client_connection_acquire(connection); - - if (aws_event_stream_channel_handler_write_message( - connection->event_stream_handler, &args->message, s_on_protocol_message_written_fn, args)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: writing message failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto message_initialized_before_failure; - } - - return AWS_OP_SUCCESS; - -message_initialized_before_failure: - aws_event_stream_message_clean_up(&args->message); - -args_allocated_before_failure: - aws_mem_release(args->allocator, args); - aws_event_stream_rpc_client_connection_release(connection); - - return AWS_OP_ERR; -} - -int aws_event_stream_rpc_client_connection_send_protocol_message( - struct aws_event_stream_rpc_client_connection *connection, - const struct aws_event_stream_rpc_message_args *message_args, - aws_event_stream_rpc_client_message_flush_fn *flush_fn, - void *user_data) { - if (!aws_event_stream_rpc_client_connection_is_open(connection)) { - return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED); - } - - return s_send_protocol_message(connection, NULL, NULL, message_args, 0, flush_fn, user_data); -} - -static void s_connection_error_message_flush_fn(int error_code, void *user_data) { - (void)error_code; - - struct aws_event_stream_rpc_client_connection *connection = user_data; - aws_event_stream_rpc_client_connection_close(connection, AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); -} - -static void s_send_connection_level_error( - struct aws_event_stream_rpc_client_connection *connection, - uint32_t message_type, - uint32_t message_flags, - const struct aws_byte_cursor *message) { - struct aws_byte_buf payload_buf = aws_byte_buf_from_array(message->ptr, message->len); - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: sending connection-level error\n" PRInSTR, - (void *)connection, - AWS_BYTE_BUF_PRI(payload_buf)); - - struct aws_event_stream_header_value_pair content_type_header = - aws_event_stream_create_string_header(s_json_content_type_name, s_json_content_type_value); - - struct aws_event_stream_header_value_pair headers[] = { - content_type_header, - }; - - struct aws_event_stream_rpc_message_args message_args = { - .message_type = message_type, - .message_flags = message_flags, - .payload = &payload_buf, - .headers_count = 1, - .headers = headers, - }; - - aws_event_stream_rpc_client_connection_send_protocol_message( - connection, &message_args, s_connection_error_message_flush_fn, connection); -} - -static void s_route_message_by_type( - struct aws_event_stream_rpc_client_connection *connection, - struct aws_event_stream_message *message, - struct aws_array_list *headers_list, - uint32_t stream_id, - uint32_t message_type, - uint32_t message_flags) { - struct aws_byte_buf payload_buf = aws_byte_buf_from_array( - aws_event_stream_message_payload(message), aws_event_stream_message_payload_len(message)); - - struct aws_event_stream_rpc_message_args message_args = { - .headers = headers_list->data, - .headers_count = aws_array_list_length(headers_list), - .payload = &payload_buf, - .message_flags = message_flags, - .message_type = message_type, - }; - - size_t handshake_complete = aws_atomic_load_int(&connection->handshake_state); - - /* make sure if this is not a CONNECT message being received, the handshake has been completed. */ - if (handshake_complete < CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && - message_type != AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: a message was received on this connection prior to the " - "connect handshake completing", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); - return; - } - - /* stream_id being non zero ALWAYS indicates APPLICATION_DATA or APPLICATION_ERROR. */ - if (stream_id > 0) { - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: stream id %" PRIu32, (void *)connection, stream_id); - struct aws_event_stream_rpc_client_continuation_token *continuation = NULL; - if (message_type > AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: only application messages can be sent on a stream id, " - "but this message is the incorrect type", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_stream_id_error); - return; - } - - aws_mutex_lock(&connection->stream_lock); - struct aws_hash_element *continuation_element = NULL; - if (aws_hash_table_find(&connection->continuation_table, &stream_id, &continuation_element) || - !continuation_element) { - aws_mutex_unlock(&connection->stream_lock); - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: a stream id was received that was not created by this client", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_client_stream_id_error); - return; - } - - aws_mutex_unlock(&connection->stream_lock); - - continuation = continuation_element->value; - aws_event_stream_rpc_client_continuation_acquire(continuation); - continuation->continuation_fn(continuation, &message_args, continuation->user_data); - aws_event_stream_rpc_client_continuation_release(continuation); - - /* if it was a terminal stream message purge it from the hash table. The delete will decref the continuation. */ - if (message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: the terminate stream flag was specified for continuation %p", - (void *)connection, - (void *)continuation); - aws_atomic_store_int(&continuation->is_closed, 1U); - aws_mutex_lock(&connection->stream_lock); - aws_hash_table_remove(&connection->continuation_table, &stream_id, NULL, NULL); - aws_mutex_unlock(&connection->stream_lock); - - /* Note that we do not invoke callback while holding lock */ - s_complete_continuation(continuation); - } - } else { - if (message_type <= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR || - message_type >= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_COUNT) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: a zero stream id was received with an invalid message-type %" PRIu32, - (void *)connection, - message_type); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_type_error); - return; - } - - if (message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { - if (handshake_complete != CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: connect ack received but the handshake is already completed. Only one is allowed.", - (void *)connection); - /* only one connect is allowed. This would be a duplicate. */ - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); - return; - } - aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED); - AWS_LOGF_INFO( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: connect ack received, connection handshake completed", - (void *)connection); - } - - connection->on_connection_protocol_message(connection, &message_args, connection->user_data); - } -} - -/* invoked by the event stream channel handler when a complete message has been read from the channel. */ -static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data) { - - if (!error_code) { - struct aws_event_stream_rpc_client_connection *connection = user_data; - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: message received on connection of length %" PRIu32, - (void *)connection, - aws_event_stream_message_total_length(message)); - - struct aws_array_list headers; - if (aws_array_list_init_dynamic( - &headers, connection->allocator, 8, sizeof(struct aws_event_stream_header_value_pair))) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: error initializing headers %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); - return; - } - - if (aws_event_stream_message_headers(message, &headers)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: error fetching headers %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); - goto clean_up; - } - - int32_t stream_id = -1; - int32_t message_type = -1; - int32_t message_flags = -1; - - struct aws_byte_buf operation_name_buf; - AWS_ZERO_STRUCT(operation_name_buf); - if (aws_event_stream_rpc_extract_message_metadata( - &headers, &stream_id, &message_type, &message_flags, &operation_name_buf)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: invalid protocol message with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_error); - goto clean_up; - } - - (void)operation_name_buf; - - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: routing message", (void *)connection); - s_route_message_by_type(connection, message, &headers, stream_id, message_type, message_flags); - - clean_up: - aws_event_stream_headers_list_cleanup(&headers); - } -} - -struct aws_event_stream_rpc_client_continuation_token *aws_event_stream_rpc_client_connection_new_stream( - struct aws_event_stream_rpc_client_connection *connection, - const struct aws_event_stream_rpc_client_stream_continuation_options *continuation_options) { - AWS_PRECONDITION(continuation_options->on_continuation_closed); - AWS_PRECONDITION(continuation_options->on_continuation); - - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: creating a new stream on connection", (void *)connection); - - struct aws_event_stream_rpc_client_continuation_token *continuation = - aws_mem_calloc(connection->allocator, 1, sizeof(struct aws_event_stream_rpc_client_continuation_token)); - - if (!continuation) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: error while allocating continuation %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - return NULL; - } - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: continuation created %p", (void *)connection, (void *)continuation); - continuation->connection = connection; - aws_event_stream_rpc_client_connection_acquire(continuation->connection); - aws_atomic_init_int(&continuation->ref_count, 1); - aws_atomic_init_int(&continuation->is_closed, 0); - continuation->continuation_fn = continuation_options->on_continuation; - continuation->closed_fn = continuation_options->on_continuation_closed; - continuation->user_data = continuation_options->user_data; - - return continuation; -} - -void *aws_event_stream_rpc_client_continuation_get_user_data( - struct aws_event_stream_rpc_client_continuation_token *continuation) { - return continuation->user_data; -} - -void aws_event_stream_rpc_client_continuation_acquire( - const struct aws_event_stream_rpc_client_continuation_token *continuation) { - size_t current_count = aws_atomic_fetch_add_explicit( - &((struct aws_event_stream_rpc_client_continuation_token *)continuation)->ref_count, - 1u, - aws_memory_order_relaxed); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: continuation acquired, new ref count is %zu.", - (void *)continuation, - current_count + 1); -} - -void aws_event_stream_rpc_client_continuation_release( - const struct aws_event_stream_rpc_client_continuation_token *continuation) { - if (AWS_UNLIKELY(!continuation)) { - return; - } - - struct aws_event_stream_rpc_client_continuation_token *continuation_mut = - (struct aws_event_stream_rpc_client_continuation_token *)continuation; - size_t ref_count = aws_atomic_fetch_sub_explicit(&continuation_mut->ref_count, 1, aws_memory_order_seq_cst); - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: continuation released, new ref count is %zu.", - (void *)continuation, - ref_count - 1); - - if (ref_count == 1) { - struct aws_allocator *allocator = continuation_mut->connection->allocator; - aws_event_stream_rpc_client_connection_release(continuation_mut->connection); - aws_mem_release(allocator, continuation_mut); - } -} - -bool aws_event_stream_rpc_client_continuation_is_closed( - const struct aws_event_stream_rpc_client_continuation_token *continuation) { - return aws_atomic_load_int(&continuation->is_closed) == 1u; -} - -int aws_event_stream_rpc_client_continuation_activate( - struct aws_event_stream_rpc_client_continuation_token *continuation, - struct aws_byte_cursor operation_name, - const struct aws_event_stream_rpc_message_args *message_args, - aws_event_stream_rpc_client_message_flush_fn *flush_fn, - void *user_data) { - - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: activating continuation", (void *)continuation); - int ret_val = AWS_OP_ERR; - - aws_mutex_lock(&continuation->connection->stream_lock); - - if (continuation->stream_id) { - AWS_LOGF_ERROR(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: stream has already been activated", (void *)continuation) - aws_raise_error(AWS_ERROR_INVALID_STATE); - goto clean_up; - } - - /* Even though is_open is atomic, we need to hold a lock while checking it. - * This lets us coordinate with code that sets is_open to false. */ - if (!aws_event_stream_rpc_client_connection_is_open(continuation->connection)) { - AWS_LOGF_ERROR(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: stream's connection is not open", (void *)continuation) - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED); - goto clean_up; - } - - /* we cannot update the connection's stream id until we're certain the message at least made it to the wire, because - * the next stream id must be consecutively increasing by 1. So send the message then update the connection state - * once we've made it to the wire. */ - continuation->stream_id = continuation->connection->latest_stream_id + 1; - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: continuation's new stream id is %" PRIu32, - (void *)continuation, - continuation->stream_id); - - if (aws_hash_table_put( - &continuation->connection->continuation_table, &continuation->stream_id, continuation, NULL)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: storing the new stream failed with %s", - (void *)continuation, - aws_error_debug_str(aws_last_error())); - continuation->stream_id = 0; - goto clean_up; - } - - if (s_send_protocol_message( - continuation->connection, - continuation, - &operation_name, - message_args, - continuation->stream_id, - flush_fn, - user_data)) { - aws_hash_table_remove(&continuation->connection->continuation_table, &continuation->stream_id, NULL, NULL); - continuation->stream_id = 0; - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_CLIENT, - "id=%p: failed to flush the new stream to the channel with error %s", - (void *)continuation, - aws_error_debug_str(aws_last_error())); - goto clean_up; - } - - /* The continuation table gets a ref count on the continuation. Take it here. */ - aws_event_stream_rpc_client_continuation_acquire(continuation); - - continuation->connection->latest_stream_id = continuation->stream_id; - ret_val = AWS_OP_SUCCESS; - -clean_up: - aws_mutex_unlock(&continuation->connection->stream_lock); - return ret_val; -} - -int aws_event_stream_rpc_client_continuation_send_message( - struct aws_event_stream_rpc_client_continuation_token *continuation, - const struct aws_event_stream_rpc_message_args *message_args, - aws_event_stream_rpc_client_message_flush_fn *flush_fn, - void *user_data) { - - if (aws_event_stream_rpc_client_continuation_is_closed(continuation)) { - return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_STREAM_CLOSED); - } - - if (!continuation->stream_id) { - return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_STREAM_NOT_ACTIVATED); - } - - return s_send_protocol_message( - continuation->connection, continuation, NULL, message_args, continuation->stream_id, flush_fn, user_data); -} +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#include <aws/event-stream/event_stream_channel_handler.h> +#include <aws/event-stream/event_stream_rpc_client.h> +#include <aws/event-stream/private/event_stream_rpc_priv.h> + +#include <aws/common/atomics.h> +#include <aws/common/hash_table.h> +#include <aws/common/mutex.h> + +#include <aws/io/channel_bootstrap.h> + +#include <inttypes.h> + +#ifdef _MSC_VER +/* allow declared initializer using address of automatic variable */ +# pragma warning(disable : 4221) +/* allow non-constant aggregate initializers */ +# pragma warning(disable : 4204) + +#endif + +static void s_clear_continuation_table(struct aws_event_stream_rpc_client_connection *connection); + +struct aws_event_stream_rpc_client_connection { + struct aws_allocator *allocator; + struct aws_hash_table continuation_table; + struct aws_client_bootstrap *bootstrap_ref; + struct aws_atomic_var ref_count; + struct aws_channel *channel; + struct aws_channel_handler *event_stream_handler; + uint32_t latest_stream_id; + struct aws_mutex stream_lock; + struct aws_atomic_var is_open; + struct aws_atomic_var handshake_state; + size_t initial_window_size; + aws_event_stream_rpc_client_on_connection_setup_fn *on_connection_setup; + aws_event_stream_rpc_client_connection_protocol_message_fn *on_connection_protocol_message; + aws_event_stream_rpc_client_on_connection_shutdown_fn *on_connection_shutdown; + void *user_data; + bool bootstrap_owned; + bool enable_read_back_pressure; +}; + +struct aws_event_stream_rpc_client_continuation_token { + uint32_t stream_id; + struct aws_event_stream_rpc_client_connection *connection; + aws_event_stream_rpc_client_stream_continuation_fn *continuation_fn; + aws_event_stream_rpc_client_stream_continuation_closed_fn *closed_fn; + void *user_data; + struct aws_atomic_var ref_count; + struct aws_atomic_var is_closed; +}; + +static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data); + +static int s_create_connection_on_channel( + struct aws_event_stream_rpc_client_connection *connection, + struct aws_channel *channel) { + struct aws_channel_handler *event_stream_handler = NULL; + struct aws_channel_slot *slot = NULL; + + struct aws_event_stream_channel_handler_options handler_options = { + .on_message_received = s_on_message_received, + .user_data = connection, + .initial_window_size = connection->initial_window_size, + .manual_window_management = connection->enable_read_back_pressure, + }; + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: creating an event-stream handler on channel %p", + (void *)connection, + (void *)channel); + event_stream_handler = aws_event_stream_channel_handler_new(connection->allocator, &handler_options); + + if (!event_stream_handler) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: creating an event-stream handler failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + slot = aws_channel_slot_new(channel); + + if (!slot) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: creating channel slot failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + aws_channel_slot_insert_end(channel, slot); + if (aws_channel_slot_set_handler(slot, event_stream_handler)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: setting handler on channel slot failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + connection->event_stream_handler = event_stream_handler; + connection->channel = channel; + aws_channel_acquire_hold(channel); + + return AWS_OP_SUCCESS; + +error: + if (!slot && event_stream_handler) { + aws_channel_handler_destroy(event_stream_handler); + } + + return AWS_OP_ERR; +} + +static void s_on_channel_setup_fn( + struct aws_client_bootstrap *bootstrap, + int error_code, + struct aws_channel *channel, + void *user_data) { + (void)bootstrap; + + struct aws_event_stream_rpc_client_connection *connection = user_data; + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: on_channel_setup_fn invoked with error_code %d with channel %p", + (void *)connection, + error_code, + (void *)channel); + + if (!error_code) { + connection->bootstrap_owned = true; + if (s_create_connection_on_channel(connection, channel)) { + int last_error = aws_last_error(); + connection->on_connection_setup(NULL, last_error, connection->user_data); + aws_channel_shutdown(channel, last_error); + return; + } + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: successful event-stream channel setup %p", + (void *)connection, + (void *)channel); + aws_event_stream_rpc_client_connection_acquire(connection); + connection->on_connection_setup(connection, AWS_OP_SUCCESS, connection->user_data); + aws_event_stream_rpc_client_connection_release(connection); + } else { + connection->on_connection_setup(NULL, error_code, connection->user_data); + aws_event_stream_rpc_client_connection_release(connection); + } +} + +static void s_on_channel_shutdown_fn( + struct aws_client_bootstrap *bootstrap, + int error_code, + struct aws_channel *channel, + void *user_data) { + (void)bootstrap; + + struct aws_event_stream_rpc_client_connection *connection = user_data; + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: on_channel_shutdown_fn invoked with error_code %d with channel %p", + (void *)connection, + error_code, + (void *)channel); + aws_atomic_store_int(&connection->is_open, 0u); + + if (connection->bootstrap_owned) { + s_clear_continuation_table(connection); + + aws_event_stream_rpc_client_connection_acquire(connection); + connection->on_connection_shutdown(connection, error_code, connection->user_data); + aws_event_stream_rpc_client_connection_release(connection); + } + + aws_channel_release_hold(channel); + aws_event_stream_rpc_client_connection_release(connection); +} + +/* Set each continuation's is_closed=true. + * A lock MUST be held while calling this. + * For use with aws_hash_table_foreach(). */ +static int s_mark_each_continuation_closed(void *context, struct aws_hash_element *p_element) { + (void)context; + struct aws_event_stream_rpc_client_continuation_token *continuation = p_element->value; + + aws_atomic_store_int(&continuation->is_closed, 1U); + + return AWS_COMMON_HASH_TABLE_ITER_CONTINUE; +} + +/* Invoke continuation's on_closed() callback. + * A lock must NOT be hold while calling this */ +static void s_complete_continuation(struct aws_event_stream_rpc_client_continuation_token *token) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "token=%p: token with stream-id %" PRIu32 ", purged from the stream table", + (void *)token, + token->stream_id); + + if (token->stream_id) { + token->closed_fn(token, token->user_data); + } + + aws_event_stream_rpc_client_continuation_release(token); +} + +static int s_complete_and_clear_each_continuation(void *context, struct aws_hash_element *p_element) { + (void)context; + struct aws_event_stream_rpc_client_continuation_token *continuation = p_element->value; + + s_complete_continuation(continuation); + + return AWS_COMMON_HASH_TABLE_ITER_DELETE | AWS_COMMON_HASH_TABLE_ITER_CONTINUE; +} + +/* Remove each continuation from hash-table and invoke its on_closed() callback. + * The connection->is_open must be set false before calling this. */ +static void s_clear_continuation_table(struct aws_event_stream_rpc_client_connection *connection) { + AWS_ASSERT(!aws_event_stream_rpc_client_connection_is_open(connection)); + + /* Use lock to ensure synchronization with code that adds entries to table. + * Since connection was just marked closed, no further entries will be + * added to table once we acquire the lock. */ + aws_mutex_lock(&connection->stream_lock); + aws_hash_table_foreach(&connection->continuation_table, s_mark_each_continuation_closed, NULL); + aws_mutex_unlock(&connection->stream_lock); + + /* Now release lock before invoking callbacks. + * It's safe to alter the table now without a lock, since no further + * entries can be added, and we've gone through the critical section + * above to ensure synchronization */ + aws_hash_table_foreach(&connection->continuation_table, s_complete_and_clear_each_continuation, NULL); +} + +int aws_event_stream_rpc_client_connection_connect( + struct aws_allocator *allocator, + const struct aws_event_stream_rpc_client_connection_options *conn_options) { + AWS_PRECONDITION(allocator); + AWS_PRECONDITION(conn_options); + AWS_PRECONDITION(conn_options->on_connection_protocol_message); + AWS_PRECONDITION(conn_options->on_connection_setup); + AWS_PRECONDITION(conn_options->on_connection_shutdown); + + struct aws_event_stream_rpc_client_connection *connection = + aws_mem_calloc(allocator, 1, sizeof(struct aws_event_stream_rpc_client_connection)); + + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: creating new connection", (void *)connection); + + if (!connection) { + return AWS_OP_ERR; + } + + connection->allocator = allocator; + aws_atomic_init_int(&connection->ref_count, 1); + connection->bootstrap_ref = conn_options->bootstrap; + /* this is released in the connection release which gets called regardless of if this function is successful or + * not*/ + aws_client_bootstrap_acquire(connection->bootstrap_ref); + aws_atomic_init_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_INITIALIZED); + aws_atomic_init_int(&connection->is_open, 1); + aws_mutex_init(&connection->stream_lock); + + connection->on_connection_shutdown = conn_options->on_connection_shutdown; + connection->on_connection_protocol_message = conn_options->on_connection_protocol_message; + connection->on_connection_setup = conn_options->on_connection_setup; + connection->user_data = conn_options->user_data; + + if (aws_hash_table_init( + &connection->continuation_table, + allocator, + 64, + aws_event_stream_rpc_hash_streamid, + aws_event_stream_rpc_streamid_eq, + NULL, + NULL)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: failed initializing continuation table with error %s.", + (void *)connection, + aws_error_debug_str(aws_last_error())); + + goto error; + } + + struct aws_socket_channel_bootstrap_options bootstrap_options = { + .bootstrap = connection->bootstrap_ref, + .tls_options = conn_options->tls_options, + .socket_options = conn_options->socket_options, + .user_data = connection, + .host_name = conn_options->host_name, + .port = conn_options->port, + .enable_read_back_pressure = false, + .setup_callback = s_on_channel_setup_fn, + .shutdown_callback = s_on_channel_shutdown_fn, + }; + + if (aws_client_bootstrap_new_socket_channel(&bootstrap_options)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: failed creating new socket channel with error %s.", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + return AWS_OP_SUCCESS; + +error: + aws_event_stream_rpc_client_connection_release(connection); + return AWS_OP_ERR; +} + +void aws_event_stream_rpc_client_connection_acquire(const struct aws_event_stream_rpc_client_connection *connection) { + AWS_PRECONDITION(connection); + size_t current_count = aws_atomic_fetch_add_explicit( + &((struct aws_event_stream_rpc_client_connection *)connection)->ref_count, 1, aws_memory_order_relaxed); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: connection acquired, new ref count is %zu.", + (void *)connection, + current_count + 1); +} + +static void s_destroy_connection(struct aws_event_stream_rpc_client_connection *connection) { + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: destroying connection.", (void *)connection); + aws_hash_table_clean_up(&connection->continuation_table); + aws_client_bootstrap_release(connection->bootstrap_ref); + aws_mem_release(connection->allocator, connection); +} + +void aws_event_stream_rpc_client_connection_release(const struct aws_event_stream_rpc_client_connection *connection) { + if (!connection) { + return; + } + + struct aws_event_stream_rpc_client_connection *connection_mut = + (struct aws_event_stream_rpc_client_connection *)connection; + size_t ref_count = aws_atomic_fetch_sub_explicit(&connection_mut->ref_count, 1, aws_memory_order_seq_cst); + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: connection released, new ref count is %zu.", + (void *)connection, + ref_count - 1); + + if (ref_count == 1) { + s_destroy_connection(connection_mut); + } +} + +void aws_event_stream_rpc_client_connection_close( + struct aws_event_stream_rpc_client_connection *connection, + int shutdown_error_code) { + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: connection close invoked with reason %s.", + (void *)connection, + aws_error_debug_str(shutdown_error_code)); + + size_t expect_open = 1U; + if (aws_atomic_compare_exchange_int(&connection->is_open, &expect_open, 0U)) { + aws_channel_shutdown(connection->channel, shutdown_error_code); + + if (!connection->bootstrap_owned) { + s_clear_continuation_table(connection); + + aws_event_stream_rpc_client_connection_release(connection); + } + } else { + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: connection already closed.", (void *)connection); + } +} + +bool aws_event_stream_rpc_client_connection_is_open(const struct aws_event_stream_rpc_client_connection *connection) { + return aws_atomic_load_int(&connection->is_open) == 1U; +} + +struct event_stream_connection_send_message_args { + struct aws_allocator *allocator; + struct aws_event_stream_message message; + enum aws_event_stream_rpc_message_type message_type; + struct aws_event_stream_rpc_client_connection *connection; + struct aws_event_stream_rpc_client_continuation_token *continuation; + aws_event_stream_rpc_client_message_flush_fn *flush_fn; + void *user_data; + bool end_stream; + bool terminate_connection; +}; + +static void s_on_protocol_message_written_fn( + struct aws_event_stream_message *message, + int error_code, + void *user_data) { + (void)message; + + struct event_stream_connection_send_message_args *message_args = user_data; + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: message %p flushed to channel.", + (void *)message_args->connection, + (void *)message); + + if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: connect message flushed to the wire, waiting on connect ack.", + (void *)message_args->connection); + aws_atomic_store_int(&message_args->connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED); + } + + if (message_args->end_stream) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: the end stream flag was set, closing continuation %p.", + (void *)message_args->connection, + (void *)message_args->continuation); + AWS_FATAL_ASSERT(message_args->continuation && "end stream flag was set but it wasn't on a continuation"); + aws_atomic_store_int(&message_args->continuation->is_closed, 1U); + + aws_mutex_lock(&message_args->connection->stream_lock); + aws_hash_table_remove( + &message_args->connection->continuation_table, &message_args->continuation->stream_id, NULL, NULL); + aws_mutex_unlock(&message_args->connection->stream_lock); + + /* Lock must NOT be held while invoking callback */ + s_complete_continuation(message_args->continuation); + } + + message_args->flush_fn(error_code, message_args->user_data); + + if (message_args->terminate_connection) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: terminate_connection flag was specified. Shutting down the connection.", + (void *)message_args->connection); + aws_event_stream_rpc_client_connection_close(message_args->connection, AWS_ERROR_SUCCESS); + } + + aws_event_stream_rpc_client_connection_release(message_args->connection); + + if (message_args->continuation) { + aws_event_stream_rpc_client_continuation_release(message_args->continuation); + } + + aws_event_stream_message_clean_up(&message_args->message); + aws_mem_release(message_args->allocator, message_args); +} + +static int s_send_protocol_message( + struct aws_event_stream_rpc_client_connection *connection, + struct aws_event_stream_rpc_client_continuation_token *continuation, + struct aws_byte_cursor *operation_name, + const struct aws_event_stream_rpc_message_args *message_args, + int32_t stream_id, + aws_event_stream_rpc_client_message_flush_fn *flush_fn, + void *user_data) { + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: sending message. continuation: %p, stream id %" PRId32, + (void *)connection, + (void *)continuation, + stream_id); + + size_t connect_handshake_state = aws_atomic_load_int(&connection->handshake_state); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: handshake completion value %zu", + (void *)connection, + connect_handshake_state); + + /* handshake step 1 is a connect message being received. Handshake 2 is the connect ack being sent. + * no messages other than connect and connect ack are allowed until this count reaches 2. */ + if (connect_handshake_state != CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && + message_args->message_type < AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: handshake not completed, only a connect message can be sent.", + (void *)connection); + return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + } + + struct event_stream_connection_send_message_args *args = + aws_mem_calloc(connection->allocator, 1, sizeof(struct event_stream_connection_send_message_args)); + + if (!message_args) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: failed to allocate callback arguments %s.", + (void *)connection, + aws_error_debug_str(aws_last_error())); + return AWS_OP_ERR; + } + + args->allocator = connection->allocator; + args->user_data = user_data; + args->message_type = message_args->message_type; + args->connection = connection; + args->flush_fn = flush_fn; + + if (continuation) { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: sending message on continuation %p", + (void *)connection, + (void *)continuation); + args->continuation = continuation; + aws_event_stream_rpc_client_continuation_acquire(continuation); + + if (message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p:end stream flag was specified on continuation %p", + (void *)connection, + (void *)continuation); + args->end_stream = true; + } + } + + if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK && + !(message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_CONNECTION_ACCEPTED)) { + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: terminating connection", (void *)connection); + args->terminate_connection = true; + } + + if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: sending connect message", (void *)connection); + aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED); + } + + args->flush_fn = flush_fn; + + size_t headers_count = operation_name ? message_args->headers_count + 4 : message_args->headers_count + 3; + struct aws_array_list headers_list; + AWS_ZERO_STRUCT(headers_list); + + if (aws_array_list_init_dynamic( + &headers_list, connection->allocator, headers_count, sizeof(struct aws_event_stream_header_value_pair))) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: an error occurred while initializing the headers list %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto args_allocated_before_failure; + } + + /* since we preallocated the space for the headers, these can't fail, but we'll go ahead an assert on them just in + * case */ + for (size_t i = 0; i < message_args->headers_count; ++i) { + AWS_FATAL_ASSERT(!aws_array_list_push_back(&headers_list, &message_args->headers[i])); + } + + AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( + &headers_list, + (const char *)aws_event_stream_rpc_message_type_name.ptr, + (uint8_t)aws_event_stream_rpc_message_type_name.len, + message_args->message_type)); + AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( + &headers_list, + (const char *)aws_event_stream_rpc_message_flags_name.ptr, + (uint8_t)aws_event_stream_rpc_message_flags_name.len, + message_args->message_flags)); + AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( + &headers_list, + (const char *)aws_event_stream_rpc_stream_id_name.ptr, + (uint8_t)aws_event_stream_rpc_stream_id_name.len, + stream_id)); + + if (operation_name) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: operation name specified " PRInSTR, + (void *)connection, + AWS_BYTE_CURSOR_PRI(*operation_name)); + AWS_FATAL_ASSERT(!aws_event_stream_add_string_header( + &headers_list, + (const char *)aws_event_stream_rpc_operation_name.ptr, + (uint8_t)aws_event_stream_rpc_operation_name.len, + (const char *)operation_name->ptr, + (uint16_t)operation_name->len, + 0)); + } + + int message_init_err_code = + aws_event_stream_message_init(&args->message, connection->allocator, &headers_list, message_args->payload); + aws_array_list_clean_up(&headers_list); + + if (message_init_err_code) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: message init failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto args_allocated_before_failure; + } + + aws_event_stream_rpc_client_connection_acquire(connection); + + if (aws_event_stream_channel_handler_write_message( + connection->event_stream_handler, &args->message, s_on_protocol_message_written_fn, args)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: writing message failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto message_initialized_before_failure; + } + + return AWS_OP_SUCCESS; + +message_initialized_before_failure: + aws_event_stream_message_clean_up(&args->message); + +args_allocated_before_failure: + aws_mem_release(args->allocator, args); + aws_event_stream_rpc_client_connection_release(connection); + + return AWS_OP_ERR; +} + +int aws_event_stream_rpc_client_connection_send_protocol_message( + struct aws_event_stream_rpc_client_connection *connection, + const struct aws_event_stream_rpc_message_args *message_args, + aws_event_stream_rpc_client_message_flush_fn *flush_fn, + void *user_data) { + if (!aws_event_stream_rpc_client_connection_is_open(connection)) { + return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED); + } + + return s_send_protocol_message(connection, NULL, NULL, message_args, 0, flush_fn, user_data); +} + +static void s_connection_error_message_flush_fn(int error_code, void *user_data) { + (void)error_code; + + struct aws_event_stream_rpc_client_connection *connection = user_data; + aws_event_stream_rpc_client_connection_close(connection, AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); +} + +static void s_send_connection_level_error( + struct aws_event_stream_rpc_client_connection *connection, + uint32_t message_type, + uint32_t message_flags, + const struct aws_byte_cursor *message) { + struct aws_byte_buf payload_buf = aws_byte_buf_from_array(message->ptr, message->len); + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: sending connection-level error\n" PRInSTR, + (void *)connection, + AWS_BYTE_BUF_PRI(payload_buf)); + + struct aws_event_stream_header_value_pair content_type_header = + aws_event_stream_create_string_header(s_json_content_type_name, s_json_content_type_value); + + struct aws_event_stream_header_value_pair headers[] = { + content_type_header, + }; + + struct aws_event_stream_rpc_message_args message_args = { + .message_type = message_type, + .message_flags = message_flags, + .payload = &payload_buf, + .headers_count = 1, + .headers = headers, + }; + + aws_event_stream_rpc_client_connection_send_protocol_message( + connection, &message_args, s_connection_error_message_flush_fn, connection); +} + +static void s_route_message_by_type( + struct aws_event_stream_rpc_client_connection *connection, + struct aws_event_stream_message *message, + struct aws_array_list *headers_list, + uint32_t stream_id, + uint32_t message_type, + uint32_t message_flags) { + struct aws_byte_buf payload_buf = aws_byte_buf_from_array( + aws_event_stream_message_payload(message), aws_event_stream_message_payload_len(message)); + + struct aws_event_stream_rpc_message_args message_args = { + .headers = headers_list->data, + .headers_count = aws_array_list_length(headers_list), + .payload = &payload_buf, + .message_flags = message_flags, + .message_type = message_type, + }; + + size_t handshake_complete = aws_atomic_load_int(&connection->handshake_state); + + /* make sure if this is not a CONNECT message being received, the handshake has been completed. */ + if (handshake_complete < CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && + message_type != AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: a message was received on this connection prior to the " + "connect handshake completing", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); + return; + } + + /* stream_id being non zero ALWAYS indicates APPLICATION_DATA or APPLICATION_ERROR. */ + if (stream_id > 0) { + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: stream id %" PRIu32, (void *)connection, stream_id); + struct aws_event_stream_rpc_client_continuation_token *continuation = NULL; + if (message_type > AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: only application messages can be sent on a stream id, " + "but this message is the incorrect type", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_stream_id_error); + return; + } + + aws_mutex_lock(&connection->stream_lock); + struct aws_hash_element *continuation_element = NULL; + if (aws_hash_table_find(&connection->continuation_table, &stream_id, &continuation_element) || + !continuation_element) { + aws_mutex_unlock(&connection->stream_lock); + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: a stream id was received that was not created by this client", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_client_stream_id_error); + return; + } + + aws_mutex_unlock(&connection->stream_lock); + + continuation = continuation_element->value; + aws_event_stream_rpc_client_continuation_acquire(continuation); + continuation->continuation_fn(continuation, &message_args, continuation->user_data); + aws_event_stream_rpc_client_continuation_release(continuation); + + /* if it was a terminal stream message purge it from the hash table. The delete will decref the continuation. */ + if (message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: the terminate stream flag was specified for continuation %p", + (void *)connection, + (void *)continuation); + aws_atomic_store_int(&continuation->is_closed, 1U); + aws_mutex_lock(&connection->stream_lock); + aws_hash_table_remove(&connection->continuation_table, &stream_id, NULL, NULL); + aws_mutex_unlock(&connection->stream_lock); + + /* Note that we do not invoke callback while holding lock */ + s_complete_continuation(continuation); + } + } else { + if (message_type <= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR || + message_type >= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_COUNT) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: a zero stream id was received with an invalid message-type %" PRIu32, + (void *)connection, + message_type); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_type_error); + return; + } + + if (message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { + if (handshake_complete != CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: connect ack received but the handshake is already completed. Only one is allowed.", + (void *)connection); + /* only one connect is allowed. This would be a duplicate. */ + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); + return; + } + aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED); + AWS_LOGF_INFO( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: connect ack received, connection handshake completed", + (void *)connection); + } + + connection->on_connection_protocol_message(connection, &message_args, connection->user_data); + } +} + +/* invoked by the event stream channel handler when a complete message has been read from the channel. */ +static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data) { + + if (!error_code) { + struct aws_event_stream_rpc_client_connection *connection = user_data; + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: message received on connection of length %" PRIu32, + (void *)connection, + aws_event_stream_message_total_length(message)); + + struct aws_array_list headers; + if (aws_array_list_init_dynamic( + &headers, connection->allocator, 8, sizeof(struct aws_event_stream_header_value_pair))) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: error initializing headers %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); + return; + } + + if (aws_event_stream_message_headers(message, &headers)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: error fetching headers %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); + goto clean_up; + } + + int32_t stream_id = -1; + int32_t message_type = -1; + int32_t message_flags = -1; + + struct aws_byte_buf operation_name_buf; + AWS_ZERO_STRUCT(operation_name_buf); + if (aws_event_stream_rpc_extract_message_metadata( + &headers, &stream_id, &message_type, &message_flags, &operation_name_buf)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: invalid protocol message with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_error); + goto clean_up; + } + + (void)operation_name_buf; + + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: routing message", (void *)connection); + s_route_message_by_type(connection, message, &headers, stream_id, message_type, message_flags); + + clean_up: + aws_event_stream_headers_list_cleanup(&headers); + } +} + +struct aws_event_stream_rpc_client_continuation_token *aws_event_stream_rpc_client_connection_new_stream( + struct aws_event_stream_rpc_client_connection *connection, + const struct aws_event_stream_rpc_client_stream_continuation_options *continuation_options) { + AWS_PRECONDITION(continuation_options->on_continuation_closed); + AWS_PRECONDITION(continuation_options->on_continuation); + + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: creating a new stream on connection", (void *)connection); + + struct aws_event_stream_rpc_client_continuation_token *continuation = + aws_mem_calloc(connection->allocator, 1, sizeof(struct aws_event_stream_rpc_client_continuation_token)); + + if (!continuation) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: error while allocating continuation %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + return NULL; + } + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: continuation created %p", (void *)connection, (void *)continuation); + continuation->connection = connection; + aws_event_stream_rpc_client_connection_acquire(continuation->connection); + aws_atomic_init_int(&continuation->ref_count, 1); + aws_atomic_init_int(&continuation->is_closed, 0); + continuation->continuation_fn = continuation_options->on_continuation; + continuation->closed_fn = continuation_options->on_continuation_closed; + continuation->user_data = continuation_options->user_data; + + return continuation; +} + +void *aws_event_stream_rpc_client_continuation_get_user_data( + struct aws_event_stream_rpc_client_continuation_token *continuation) { + return continuation->user_data; +} + +void aws_event_stream_rpc_client_continuation_acquire( + const struct aws_event_stream_rpc_client_continuation_token *continuation) { + size_t current_count = aws_atomic_fetch_add_explicit( + &((struct aws_event_stream_rpc_client_continuation_token *)continuation)->ref_count, + 1u, + aws_memory_order_relaxed); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: continuation acquired, new ref count is %zu.", + (void *)continuation, + current_count + 1); +} + +void aws_event_stream_rpc_client_continuation_release( + const struct aws_event_stream_rpc_client_continuation_token *continuation) { + if (AWS_UNLIKELY(!continuation)) { + return; + } + + struct aws_event_stream_rpc_client_continuation_token *continuation_mut = + (struct aws_event_stream_rpc_client_continuation_token *)continuation; + size_t ref_count = aws_atomic_fetch_sub_explicit(&continuation_mut->ref_count, 1, aws_memory_order_seq_cst); + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: continuation released, new ref count is %zu.", + (void *)continuation, + ref_count - 1); + + if (ref_count == 1) { + struct aws_allocator *allocator = continuation_mut->connection->allocator; + aws_event_stream_rpc_client_connection_release(continuation_mut->connection); + aws_mem_release(allocator, continuation_mut); + } +} + +bool aws_event_stream_rpc_client_continuation_is_closed( + const struct aws_event_stream_rpc_client_continuation_token *continuation) { + return aws_atomic_load_int(&continuation->is_closed) == 1u; +} + +int aws_event_stream_rpc_client_continuation_activate( + struct aws_event_stream_rpc_client_continuation_token *continuation, + struct aws_byte_cursor operation_name, + const struct aws_event_stream_rpc_message_args *message_args, + aws_event_stream_rpc_client_message_flush_fn *flush_fn, + void *user_data) { + + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: activating continuation", (void *)continuation); + int ret_val = AWS_OP_ERR; + + aws_mutex_lock(&continuation->connection->stream_lock); + + if (continuation->stream_id) { + AWS_LOGF_ERROR(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: stream has already been activated", (void *)continuation) + aws_raise_error(AWS_ERROR_INVALID_STATE); + goto clean_up; + } + + /* Even though is_open is atomic, we need to hold a lock while checking it. + * This lets us coordinate with code that sets is_open to false. */ + if (!aws_event_stream_rpc_client_connection_is_open(continuation->connection)) { + AWS_LOGF_ERROR(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: stream's connection is not open", (void *)continuation) + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED); + goto clean_up; + } + + /* we cannot update the connection's stream id until we're certain the message at least made it to the wire, because + * the next stream id must be consecutively increasing by 1. So send the message then update the connection state + * once we've made it to the wire. */ + continuation->stream_id = continuation->connection->latest_stream_id + 1; + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: continuation's new stream id is %" PRIu32, + (void *)continuation, + continuation->stream_id); + + if (aws_hash_table_put( + &continuation->connection->continuation_table, &continuation->stream_id, continuation, NULL)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: storing the new stream failed with %s", + (void *)continuation, + aws_error_debug_str(aws_last_error())); + continuation->stream_id = 0; + goto clean_up; + } + + if (s_send_protocol_message( + continuation->connection, + continuation, + &operation_name, + message_args, + continuation->stream_id, + flush_fn, + user_data)) { + aws_hash_table_remove(&continuation->connection->continuation_table, &continuation->stream_id, NULL, NULL); + continuation->stream_id = 0; + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_CLIENT, + "id=%p: failed to flush the new stream to the channel with error %s", + (void *)continuation, + aws_error_debug_str(aws_last_error())); + goto clean_up; + } + + /* The continuation table gets a ref count on the continuation. Take it here. */ + aws_event_stream_rpc_client_continuation_acquire(continuation); + + continuation->connection->latest_stream_id = continuation->stream_id; + ret_val = AWS_OP_SUCCESS; + +clean_up: + aws_mutex_unlock(&continuation->connection->stream_lock); + return ret_val; +} + +int aws_event_stream_rpc_client_continuation_send_message( + struct aws_event_stream_rpc_client_continuation_token *continuation, + const struct aws_event_stream_rpc_message_args *message_args, + aws_event_stream_rpc_client_message_flush_fn *flush_fn, + void *user_data) { + + if (aws_event_stream_rpc_client_continuation_is_closed(continuation)) { + return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_STREAM_CLOSED); + } + + if (!continuation->stream_id) { + return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_STREAM_NOT_ACTIVATED); + } + + return s_send_protocol_message( + continuation->connection, continuation, NULL, message_args, continuation->stream_id, flush_fn, user_data); +} diff --git a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_server.c b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_server.c index 6e06a7842a..1678308664 100644 --- a/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_server.c +++ b/contrib/restricted/aws/aws-c-event-stream/source/event_stream_rpc_server.c @@ -1,1110 +1,1110 @@ -/* - * Copyright 2010-2020 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. - */ - -#include <aws/event-stream/event_stream_channel_handler.h> -#include <aws/event-stream/event_stream_rpc_server.h> -#include <aws/event-stream/private/event_stream_rpc_priv.h> - -#include <aws/common/atomics.h> -#include <aws/common/hash_table.h> - -#include <aws/io/channel.h> -#include <aws/io/channel_bootstrap.h> - -#include <inttypes.h> - -#if defined(_MSC_VER) -/* allow non-constant aggregate initializer */ -# pragma warning(disable : 4204) -/* allow passing a pointer to an automatically allocated variable around, cause I'm smarter than the compiler. */ -# pragma warning(disable : 4221) -#endif - -static const struct aws_byte_cursor s_missing_operation_name_error = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL( - "{ \"message\": \"The first message for on a non-zero :stream-id must contain an operation header value.\"; }"); - -struct aws_event_stream_rpc_server_listener { - struct aws_allocator *allocator; - struct aws_socket *listener; - struct aws_server_bootstrap *bootstrap; - struct aws_atomic_var ref_count; - aws_event_stream_rpc_server_on_new_connection_fn *on_new_connection; - aws_event_stream_rpc_server_on_connection_shutdown_fn *on_connection_shutdown; - aws_event_stream_rpc_server_on_listener_destroy_fn *on_destroy_callback; - size_t initial_window_size; - bool enable_read_backpressure; - bool initialized; - void *user_data; -}; - -struct aws_event_stream_rpc_server_connection { - struct aws_allocator *allocator; - struct aws_hash_table continuation_table; - struct aws_event_stream_rpc_server_listener *server; - struct aws_atomic_var ref_count; - aws_event_stream_rpc_server_on_incoming_stream_fn *on_incoming_stream; - aws_event_stream_rpc_server_connection_protocol_message_fn *on_connection_protocol_message; - struct aws_channel *channel; - struct aws_channel_handler *event_stream_handler; - uint32_t latest_stream_id; - void *user_data; - struct aws_atomic_var is_open; - struct aws_atomic_var handshake_state; - bool bootstrap_owned; -}; - -struct aws_event_stream_rpc_server_continuation_token { - uint32_t stream_id; - struct aws_event_stream_rpc_server_connection *connection; - aws_event_stream_rpc_server_stream_continuation_fn *continuation_fn; - aws_event_stream_rpc_server_stream_continuation_closed_fn *closed_fn; - void *user_data; - struct aws_atomic_var ref_count; - struct aws_atomic_var is_closed; -}; - -/** This is the destructor callback invoked by the connections continuation table when a continuation is removed - * from the hash table. - */ -void s_continuation_destroy(void *value) { - struct aws_event_stream_rpc_server_continuation_token *continuation = value; - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying continuation", (void *)continuation); - continuation->closed_fn(continuation, continuation->user_data); - aws_event_stream_rpc_server_continuation_release(continuation); -} - -static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data); - -/* We have two paths for creating a connection on a channel. The first is an incoming connection on the server listener. - * The second is adding a connection to an already existing channel. This is the code common to both cases. */ -static struct aws_event_stream_rpc_server_connection *s_create_connection_on_channel( - struct aws_event_stream_rpc_server_listener *server, - struct aws_channel *channel) { - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: creating connection on channel %p", (void *)server, (void *)channel); - struct aws_event_stream_rpc_server_connection *connection = - aws_mem_calloc(server->allocator, 1, sizeof(struct aws_event_stream_rpc_server_connection)); - struct aws_channel_handler *event_stream_handler = NULL; - struct aws_channel_slot *slot = NULL; - - if (!connection) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: allocation failed for connection with error %s", - (void *)server, - aws_error_debug_str(aws_last_error())); - return NULL; - } - - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: new connection is %p", (void *)server, (void *)connection); - aws_atomic_init_int(&connection->ref_count, 1u); - aws_atomic_init_int(&connection->is_open, 1u); - /* handshake step 1 is a connect message being received. Handshake 2 is the connect ack being sent. - * no messages other than connect and connect ack are allowed until this count reaches 2. */ - aws_atomic_init_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_INITIALIZED); - connection->allocator = server->allocator; - - if (aws_hash_table_init( - &connection->continuation_table, - server->allocator, - 64, - aws_event_stream_rpc_hash_streamid, - aws_event_stream_rpc_streamid_eq, - NULL, - s_continuation_destroy)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: initialization of connection stream table failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - struct aws_event_stream_channel_handler_options handler_options = { - .on_message_received = s_on_message_received, - .user_data = connection, - .initial_window_size = server->initial_window_size, - .manual_window_management = server->enable_read_backpressure, - }; - - event_stream_handler = aws_event_stream_channel_handler_new(server->allocator, &handler_options); - - if (!event_stream_handler) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: initialization of event-stream handler failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - slot = aws_channel_slot_new(channel); - - if (!slot) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: initialization of channel slot failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - aws_channel_slot_insert_end(channel, slot); - if (aws_channel_slot_set_handler(slot, event_stream_handler)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: setting the handler on the slot failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto error; - } - - aws_event_stream_rpc_server_listener_acquire(server); - connection->server = server; - - connection->event_stream_handler = event_stream_handler; - connection->channel = channel; - aws_channel_acquire_hold(channel); - - return connection; - -error: - if (!slot && event_stream_handler) { - aws_channel_handler_destroy(event_stream_handler); - } - - if (connection) { - aws_event_stream_rpc_server_connection_release(connection); - } - - return NULL; -} - -struct aws_event_stream_rpc_server_connection *aws_event_stream_rpc_server_connection_from_existing_channel( - struct aws_event_stream_rpc_server_listener *server, - struct aws_channel *channel, - const struct aws_event_stream_rpc_connection_options *connection_options) { - AWS_FATAL_ASSERT( - connection_options->on_connection_protocol_message && "on_connection_protocol_message must be specified!"); - AWS_FATAL_ASSERT(connection_options->on_incoming_stream && "on_connection_protocol_message must be specified"); - - struct aws_event_stream_rpc_server_connection *connection = s_create_connection_on_channel(server, channel); - - if (!connection) { - return NULL; - } - - connection->on_incoming_stream = connection_options->on_incoming_stream; - connection->on_connection_protocol_message = connection_options->on_connection_protocol_message; - connection->user_data = connection_options->user_data; - aws_event_stream_rpc_server_connection_acquire(connection); - - return connection; -} - -void aws_event_stream_rpc_server_connection_acquire(struct aws_event_stream_rpc_server_connection *connection) { - size_t current_count = aws_atomic_fetch_add_explicit(&connection->ref_count, 1, aws_memory_order_relaxed); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: connection acquired, new ref count is %zu.", - (void *)connection, - current_count + 1); -} - -void aws_event_stream_rpc_server_connection_release(struct aws_event_stream_rpc_server_connection *connection) { - if (!connection) { - return; - } - - size_t value = aws_atomic_fetch_sub_explicit(&connection->ref_count, 1, aws_memory_order_seq_cst); - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: connection released, new ref count is %zu.", - (void *)connection, - value - 1); - if (value == 1) { - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying connection.", (void *)connection); - aws_channel_release_hold(connection->channel); - aws_hash_table_clean_up(&connection->continuation_table); - aws_event_stream_rpc_server_listener_release(connection->server); - aws_mem_release(connection->allocator, connection); - } -} - -/* incoming from a socket on this listener. */ -static void s_on_accept_channel_setup( - struct aws_server_bootstrap *bootstrap, - int error_code, - struct aws_channel *channel, - void *user_data) { - (void)bootstrap; - - struct aws_event_stream_rpc_server_listener *server = user_data; - - if (!error_code) { - AWS_LOGF_INFO( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: incoming connection with channel %p.", - (void *)server, - (void *)channel); - AWS_FATAL_ASSERT(channel && "Channel should never be null with a 0 error code."); - - struct aws_event_stream_rpc_server_connection *connection = s_create_connection_on_channel(server, channel); - - if (!connection) { - int error = aws_last_error(); - server->on_new_connection(NULL, error, NULL, server->user_data); - aws_channel_shutdown(channel, error); - } - - struct aws_event_stream_rpc_connection_options connection_options; - AWS_ZERO_STRUCT(connection_options); - - aws_event_stream_rpc_server_connection_acquire(connection); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: invoking on_new_connection with connection %p.", - (void *)server, - (void *)connection); - - if (server->on_new_connection(connection, AWS_ERROR_SUCCESS, &connection_options, server->user_data)) { - aws_channel_shutdown(channel, aws_last_error()); - aws_event_stream_rpc_server_connection_release(connection); - return; - } - - AWS_FATAL_ASSERT( - connection_options.on_connection_protocol_message && "on_connection_protocol_message must be specified!"); - AWS_FATAL_ASSERT(connection_options.on_incoming_stream && "on_connection_protocol_message must be specified"); - connection->on_incoming_stream = connection_options.on_incoming_stream; - connection->on_connection_protocol_message = connection_options.on_connection_protocol_message; - connection->user_data = connection_options.user_data; - connection->bootstrap_owned = true; - aws_event_stream_rpc_server_connection_release(connection); - - } else { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: invoking on_new_connection with error %s", - (void *)server, - aws_error_debug_str(error_code)); - server->on_new_connection(NULL, error_code, NULL, server->user_data); - } -} - -/* this is just to get the connection object off of the channel. */ -static inline struct aws_event_stream_rpc_server_connection *s_rpc_connection_from_channel( - struct aws_channel *channel) { - struct aws_channel_slot *our_slot = NULL; - struct aws_channel_slot *current_slot = aws_channel_get_first_slot(channel); - AWS_FATAL_ASSERT( - current_slot && - "It should be logically impossible to have a channel in this callback that doesn't have a slot in it"); - while (current_slot->adj_right) { - current_slot = current_slot->adj_right; - } - our_slot = current_slot; - struct aws_channel_handler *our_handler = our_slot->handler; - return aws_event_stream_channel_handler_get_user_data(our_handler); -} - -static void s_on_accept_channel_shutdown( - struct aws_server_bootstrap *bootstrap, - int error_code, - struct aws_channel *channel, - void *user_data) { - (void)bootstrap; - - struct aws_event_stream_rpc_server_listener *server = user_data; - struct aws_event_stream_rpc_server_connection *connection = s_rpc_connection_from_channel(channel); - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: channel %p and connection %p shutdown occurred with error %s", - (void *)server, - (void *)channel, - (void *)connection, - aws_error_debug_str(error_code)); - - aws_atomic_store_int(&connection->is_open, 0U); - aws_hash_table_clear(&connection->continuation_table); - aws_event_stream_rpc_server_connection_acquire(connection); - server->on_connection_shutdown(connection, error_code, server->user_data); - aws_event_stream_rpc_server_connection_release(connection); - aws_event_stream_rpc_server_connection_release(connection); -} - -static void s_on_server_listener_destroy(struct aws_server_bootstrap *bootstrap, void *user_data) { - (void)bootstrap; - struct aws_event_stream_rpc_server_listener *listener = user_data; - - AWS_LOGF_INFO(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying server", (void *)listener); - - /* server bootstrap invokes this callback regardless of if the listener was successfully created, so - * just check that we successfully set it up before freeing anything. When that's fixed in aws-c-io, this - * code will still be correct, so just leave it here for now. */ - if (listener->initialized) { - if (listener->on_destroy_callback) { - listener->on_destroy_callback(listener, listener->user_data); - } - - aws_mem_release(listener->allocator, listener); - } -} - -struct aws_event_stream_rpc_server_listener *aws_event_stream_rpc_server_new_listener( - struct aws_allocator *allocator, - struct aws_event_stream_rpc_server_listener_options *options) { - struct aws_event_stream_rpc_server_listener *server = - aws_mem_calloc(allocator, 1, sizeof(struct aws_event_stream_rpc_server_listener)); - - if (!server) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "static: failed to allocate new server with error %s", - aws_error_debug_str(aws_last_error())); - return NULL; - } - - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "static: new server is %p", (void *)server); - aws_atomic_init_int(&server->ref_count, 1); - - struct aws_server_socket_channel_bootstrap_options bootstrap_options = { - .bootstrap = options->bootstrap, - .socket_options = options->socket_options, - .tls_options = options->tls_options, - .enable_read_back_pressure = false, - .host_name = options->host_name, - .port = options->port, - .incoming_callback = s_on_accept_channel_setup, - .shutdown_callback = s_on_accept_channel_shutdown, - .destroy_callback = s_on_server_listener_destroy, - .user_data = server, - }; - - server->bootstrap = options->bootstrap; - server->allocator = allocator; - server->on_destroy_callback = options->on_destroy_callback; - server->on_new_connection = options->on_new_connection; - server->on_connection_shutdown = options->on_connection_shutdown; - server->user_data = options->user_data; - - server->listener = aws_server_bootstrap_new_socket_listener(&bootstrap_options); - - if (!server->listener) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "static: failed to allocate new socket listener with error %s", - aws_error_debug_str(aws_last_error())); - goto error; - } - - server->initialized = true; - return server; - -error: - if (server->listener) { - aws_server_bootstrap_destroy_socket_listener(options->bootstrap, server->listener); - } - - aws_mem_release(server->allocator, server); - return NULL; -} - -void aws_event_stream_rpc_server_listener_acquire(struct aws_event_stream_rpc_server_listener *server) { - size_t current_count = aws_atomic_fetch_add_explicit(&server->ref_count, 1, aws_memory_order_relaxed); - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: server acquired, new ref count is %zu.", - (void *)server, - current_count + 1); -} - -static void s_destroy_server(struct aws_event_stream_rpc_server_listener *server) { - if (server) { - AWS_LOGF_INFO(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying server", (void *)server); - /* the memory for this is cleaned up in the listener shutdown complete callback. */ - aws_server_bootstrap_destroy_socket_listener(server->bootstrap, server->listener); - } -} - -void aws_event_stream_rpc_server_listener_release(struct aws_event_stream_rpc_server_listener *server) { - if (!server) { - return; - } - - size_t ref_count = aws_atomic_fetch_sub_explicit(&server->ref_count, 1, aws_memory_order_seq_cst); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: server released, new ref count is %zu.", (void *)server, ref_count - 1); - - if (ref_count == 1) { - s_destroy_server(server); - } -} - -struct event_stream_connection_send_message_args { - struct aws_allocator *allocator; - struct aws_event_stream_message message; - enum aws_event_stream_rpc_message_type message_type; - struct aws_event_stream_rpc_server_connection *connection; - struct aws_event_stream_rpc_server_continuation_token *continuation; - aws_event_stream_rpc_server_message_flush_fn *flush_fn; - void *user_data; - bool end_stream; - bool terminate_connection; -}; - -static void s_on_protocol_message_written_fn( - struct aws_event_stream_message *message, - int error_code, - void *user_data) { - (void)message; - - struct event_stream_connection_send_message_args *message_args = user_data; - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: message flushed to channel with error %s", - (void *)message_args->connection, - aws_error_debug_str(error_code)); - - if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { - AWS_LOGF_INFO( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: connect ack message sent, the connect handshake is completed", - (void *)message_args->connection); - aws_atomic_store_int( - &message_args->connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED); - } - - if (message_args->end_stream) { - AWS_FATAL_ASSERT(message_args->continuation && "end stream flag was set but it wasn't on a continuation"); - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: end_stream flag for continuation %p was set, closing", - (void *)message_args->connection, - (void *)message_args->continuation); - aws_atomic_store_int(&message_args->continuation->is_closed, 1U); - aws_hash_table_remove( - &message_args->connection->continuation_table, &message_args->continuation->stream_id, NULL, NULL); - } - - message_args->flush_fn(error_code, message_args->user_data); - - if (message_args->terminate_connection) { - AWS_LOGF_INFO( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: terminate connection flag was set. closing", - (void *)message_args->connection); - aws_event_stream_rpc_server_connection_close(message_args->connection, AWS_ERROR_SUCCESS); - } - - aws_event_stream_rpc_server_connection_release(message_args->connection); - - if (message_args->continuation) { - aws_event_stream_rpc_server_continuation_release(message_args->continuation); - } - - aws_event_stream_message_clean_up(&message_args->message); - aws_mem_release(message_args->allocator, message_args); -} - -static int s_send_protocol_message( - struct aws_event_stream_rpc_server_connection *connection, - struct aws_event_stream_rpc_server_continuation_token *continuation, - const struct aws_event_stream_rpc_message_args *message_args, - int32_t stream_id, - aws_event_stream_rpc_server_message_flush_fn *flush_fn, - void *user_data) { - - size_t connect_handshake_state = aws_atomic_load_int(&connection->handshake_state); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: connect handshake state %zu", - (void *)connection, - connect_handshake_state); - /* handshake step 1 is a connect message being received. Handshake 2 is the connect ack being sent. - * no messages other than connect and connect ack are allowed until this count reaches 2. */ - if (connect_handshake_state != CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && - message_args->message_type < AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: invalid state, a message was received prior to connect handshake completion", - (void *)connection); - return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - } - - struct event_stream_connection_send_message_args *args = - aws_mem_calloc(connection->allocator, 1, sizeof(struct event_stream_connection_send_message_args)); - - if (!message_args) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: allocation of callback args failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - return AWS_OP_ERR; - } - - args->allocator = connection->allocator; - args->user_data = user_data; - args->message_type = message_args->message_type; - args->connection = connection; - args->flush_fn = flush_fn; - - if (continuation) { - args->continuation = continuation; - aws_event_stream_rpc_server_continuation_acquire(continuation); - - if (message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: continuation with terminate stream flag was specified closing", - (void *)continuation); - args->end_stream = true; - } - } - - if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK && - !(message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_CONNECTION_ACCEPTED)) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: connection ack was rejected closing connection", - (void *)connection); - args->terminate_connection = true; - } - - args->flush_fn = flush_fn; - - size_t headers_count = message_args->headers_count + 3; - struct aws_array_list headers_list; - AWS_ZERO_STRUCT(headers_list); - - if (aws_array_list_init_dynamic( - &headers_list, connection->allocator, headers_count, sizeof(struct aws_event_stream_header_value_pair))) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: allocation of headers failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto args_allocated_before_failure; - } - - /* since we preallocated the space for the headers, these can't fail, but we'll go ahead an assert on them just in - * case */ - for (size_t i = 0; i < message_args->headers_count; ++i) { - AWS_FATAL_ASSERT(!aws_array_list_push_back(&headers_list, &message_args->headers[i])); - } - - AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( - &headers_list, - (const char *)aws_event_stream_rpc_message_type_name.ptr, - (uint8_t)aws_event_stream_rpc_message_type_name.len, - message_args->message_type)); - AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( - &headers_list, - (const char *)aws_event_stream_rpc_message_flags_name.ptr, - (uint8_t)aws_event_stream_rpc_message_flags_name.len, - message_args->message_flags)); - AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( - &headers_list, - (const char *)aws_event_stream_rpc_stream_id_name.ptr, - (uint8_t)aws_event_stream_rpc_stream_id_name.len, - stream_id)); - - int message_init_err_code = - aws_event_stream_message_init(&args->message, connection->allocator, &headers_list, message_args->payload); - aws_array_list_clean_up(&headers_list); - - if (message_init_err_code) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: initialization of message failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto args_allocated_before_failure; - } - - aws_event_stream_rpc_server_connection_acquire(connection); - - if (aws_event_stream_channel_handler_write_message( - connection->event_stream_handler, &args->message, s_on_protocol_message_written_fn, args)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: message send failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - goto message_initialized_before_failure; - } - - return AWS_OP_SUCCESS; - -message_initialized_before_failure: - aws_event_stream_message_clean_up(&args->message); - -args_allocated_before_failure: - aws_mem_release(args->allocator, args); - aws_event_stream_rpc_server_connection_release(connection); - - return AWS_OP_ERR; -} - -int aws_event_stream_rpc_server_connection_send_protocol_message( - struct aws_event_stream_rpc_server_connection *connection, - const struct aws_event_stream_rpc_message_args *message_args, - aws_event_stream_rpc_server_message_flush_fn *flush_fn, - void *user_data) { - if (!aws_event_stream_rpc_server_connection_is_open(connection)) { - return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED); - } - - return s_send_protocol_message(connection, NULL, message_args, 0, flush_fn, user_data); -} - -void *aws_event_stream_rpc_server_connection_get_user_data(struct aws_event_stream_rpc_server_connection *connection) { - return connection->user_data; -} - -AWS_EVENT_STREAM_API void aws_event_stream_rpc_server_override_last_stream_id( - struct aws_event_stream_rpc_server_connection *connection, - int32_t value) { - connection->latest_stream_id = value; -} - -void aws_event_stream_rpc_server_connection_close( - struct aws_event_stream_rpc_server_connection *connection, - int shutdown_error_code) { - - if (aws_event_stream_rpc_server_connection_is_open(connection)) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: closing connection with error %s", - (void *)connection, - aws_error_debug_str(shutdown_error_code)); - aws_atomic_store_int(&connection->is_open, 0U); - aws_channel_shutdown(connection->channel, shutdown_error_code); - - if (!connection->bootstrap_owned) { - aws_hash_table_clear(&connection->continuation_table); - aws_event_stream_rpc_server_connection_release(connection); - } - } -} - -bool aws_event_stream_rpc_server_continuation_is_closed( - struct aws_event_stream_rpc_server_continuation_token *continuation) { - return aws_atomic_load_int(&continuation->is_closed) == 1U; -} - -bool aws_event_stream_rpc_server_connection_is_open(struct aws_event_stream_rpc_server_connection *connection) { - return aws_atomic_load_int(&connection->is_open) == 1U; -} - -void aws_event_stream_rpc_server_continuation_acquire( - struct aws_event_stream_rpc_server_continuation_token *continuation) { - size_t current_count = aws_atomic_fetch_add_explicit(&continuation->ref_count, 1, aws_memory_order_relaxed); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: continuation acquired, new ref count is %zu.", - (void *)continuation, - current_count + 1); -} - -void aws_event_stream_rpc_server_continuation_release( - struct aws_event_stream_rpc_server_continuation_token *continuation) { - size_t value = aws_atomic_fetch_sub_explicit(&continuation->ref_count, 1, aws_memory_order_seq_cst); - - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: continuation released, new ref count is %zu.", - (void *)continuation, - value - 1); - - if (value == 1) { - AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying continuation.", (void *)continuation); - struct aws_allocator *allocator = continuation->connection->allocator; - aws_event_stream_rpc_server_connection_release(continuation->connection); - aws_mem_release(allocator, continuation); - } -} - -int aws_event_stream_rpc_server_continuation_send_message( - struct aws_event_stream_rpc_server_continuation_token *continuation, - const struct aws_event_stream_rpc_message_args *message_args, - aws_event_stream_rpc_server_message_flush_fn *flush_fn, - void *user_data) { - AWS_FATAL_PRECONDITION(continuation->continuation_fn); - AWS_FATAL_PRECONDITION(continuation->closed_fn); - - if (aws_event_stream_rpc_server_continuation_is_closed(continuation)) { - return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_STREAM_CLOSED); - } - - return s_send_protocol_message( - continuation->connection, continuation, message_args, continuation->stream_id, flush_fn, user_data); -} - -static void s_connection_error_message_flush_fn(int error_code, void *user_data) { - (void)error_code; - - struct aws_event_stream_rpc_server_connection *connection = user_data; - aws_event_stream_rpc_server_connection_close(connection, AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); -} - -static void s_send_connection_level_error( - struct aws_event_stream_rpc_server_connection *connection, - uint32_t message_type, - uint32_t message_flags, - const struct aws_byte_cursor *message) { - struct aws_byte_buf payload_buf = aws_byte_buf_from_array(message->ptr, message->len); - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: sending connection-level error\n" PRInSTR, - (void *)connection, - AWS_BYTE_BUF_PRI(payload_buf)); - - struct aws_event_stream_header_value_pair content_type_header = - aws_event_stream_create_string_header(s_json_content_type_name, s_json_content_type_value); - - struct aws_event_stream_header_value_pair headers[] = { - content_type_header, - }; - - struct aws_event_stream_rpc_message_args message_args = { - .message_type = message_type, - .message_flags = message_flags, - .payload = &payload_buf, - .headers_count = 1, - .headers = headers, - }; - - aws_event_stream_rpc_server_connection_send_protocol_message( - connection, &message_args, s_connection_error_message_flush_fn, connection); -} - -/* TODO: come back and make this a proper state pattern. For now it's branches all over the place until we nail - * down the spec. */ -static void s_route_message_by_type( - struct aws_event_stream_rpc_server_connection *connection, - struct aws_event_stream_message *message, - struct aws_array_list *headers_list, - uint32_t stream_id, - uint32_t message_type, - uint32_t message_flags, - struct aws_byte_cursor operation_name) { - struct aws_byte_buf payload_buf = aws_byte_buf_from_array( - aws_event_stream_message_payload(message), aws_event_stream_message_payload_len(message)); - - struct aws_event_stream_rpc_message_args message_args = { - .headers = headers_list->data, - .headers_count = aws_array_list_length(headers_list), - .payload = &payload_buf, - .message_flags = message_flags, - .message_type = message_type, - }; - - size_t handshake_state = aws_atomic_load_int(&connection->handshake_state); - - /* make sure if this is not a CONNECT message being received, the handshake has been completed. */ - if (handshake_state < CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && - message_type != AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: a message was received on this connection prior to the " - "connect handshake completing", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); - return; - } - - /* stream_id being non zero ALWAYS indicates APPLICATION_DATA or APPLICATION_ERROR. */ - if (stream_id > 0) { - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: stream id %" PRIu32, (void *)connection, stream_id); - - struct aws_event_stream_rpc_server_continuation_token *continuation = NULL; - if (message_type > AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: only application messages can be sent on a stream id, " - "but this message is the incorrect type", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_stream_id_error); - return; - } - - /* INT32_MAX is the max stream id. */ - if (stream_id > INT32_MAX) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: stream_id is larger than the max acceptable value", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_stream_id_error); - return; - } - - /* if the stream is is in the past, look it up from the continuation table. If it's not there, that's an error. - * if it is, find it and notify the user a message arrived */ - if (stream_id <= connection->latest_stream_id) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: stream_id is an already seen stream_id, looking for existing continuation", - (void *)connection); - - struct aws_hash_element *continuation_element = NULL; - if (aws_hash_table_find(&connection->continuation_table, &stream_id, &continuation_element) || - !continuation_element) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: stream_id does not have a corresponding continuation", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_client_stream_id_error); - return; - } - - continuation = continuation_element->value; - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: stream_id corresponds to continuation %p", - (void *)connection, - (void *)continuation); - - aws_event_stream_rpc_server_continuation_acquire(continuation); - continuation->continuation_fn(continuation, &message_args, continuation->user_data); - aws_event_stream_rpc_server_continuation_release(continuation); - /* now these are potentially new streams. Make sure they're in bounds, create a new continuation - * and notify the user the stream has been created, then send them the message. */ - } else { - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: stream_id is unknown, attempting to create a continuation for it", - (void *)connection); - if (stream_id != connection->latest_stream_id + 1) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: stream_id is invalid because it's not sequentially increasing", - (void *)connection); - - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, - AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, - 0, - &s_invalid_new_client_stream_id_error); - return; - } - - /* new streams must always have an operation name. */ - if (operation_name.len == 0) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: new stream_id encountered, but an operation name was not received", - (void *)connection); - aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_missing_operation_name_error); - return; - } - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: stream_id is a valid new stream. Creating continuation", - (void *)connection); - continuation = - aws_mem_calloc(connection->allocator, 1, sizeof(struct aws_event_stream_rpc_server_continuation_token)); - if (!continuation) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: continuation allocation failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); - return; - } - - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: new continuation is %p", - (void *)connection, - (void *)continuation); - - continuation->stream_id = stream_id; - continuation->connection = connection; - aws_event_stream_rpc_server_connection_acquire(continuation->connection); - aws_atomic_init_int(&continuation->ref_count, 1); - - if (aws_hash_table_put(&connection->continuation_table, &continuation->stream_id, continuation, NULL)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: continuation table update failed with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - /* continuation release will drop the connection reference as well */ - aws_event_stream_rpc_server_continuation_release(continuation); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); - return; - } - - struct aws_event_stream_rpc_server_stream_continuation_options options; - AWS_ZERO_STRUCT(options); - - aws_event_stream_rpc_server_continuation_acquire(continuation); - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: invoking on_incoming_stream callback", (void *)connection); - if (connection->on_incoming_stream( - continuation->connection, continuation, operation_name, &options, connection->user_data)) { - aws_event_stream_rpc_server_continuation_release(continuation); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); - return; - } - AWS_FATAL_ASSERT(options.on_continuation); - AWS_FATAL_ASSERT(options.on_continuation_closed); - - continuation->continuation_fn = options.on_continuation; - continuation->closed_fn = options.on_continuation_closed; - continuation->user_data = options.user_data; - - connection->latest_stream_id = stream_id; - continuation->continuation_fn(continuation, &message_args, continuation->user_data); - aws_event_stream_rpc_server_continuation_release(continuation); - } - - /* if it was a terminal stream message purge it from the hash table. The delete will decref the continuation. */ - if (message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { - AWS_LOGF_DEBUG( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: the terminate_stream flag was received for continuation %p, closing", - (void *)connection, - (void *)continuation); - aws_atomic_store_int(&continuation->is_closed, 1U); - aws_hash_table_remove(&connection->continuation_table, &stream_id, NULL, NULL); - } - } else { - if (message_type <= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR || - message_type >= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_COUNT) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: a zero stream id was received with an invalid message-type %" PRIu32, - (void *)connection, - message_type); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_type_error); - return; - } - - if (message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { - if (handshake_state) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: connect received but the handshake is already completed. Only one is allowed.", - (void *)connection); - /* only one connect is allowed. This would be a duplicate. */ - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); - return; - } - aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED); - AWS_LOGF_INFO( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: connect received, connection handshake completion pending the server sending an ack.", - (void *)connection); - } - - connection->on_connection_protocol_message(connection, &message_args, connection->user_data); - } -} - -/* invoked by the event stream channel handler when a complete message has been read from the channel. */ -static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data) { - - if (!error_code) { - struct aws_event_stream_rpc_server_connection *connection = user_data; - AWS_LOGF_TRACE( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: message received on connection of length %" PRIu32, - (void *)connection, - aws_event_stream_message_total_length(message)); - - struct aws_array_list headers; - if (aws_array_list_init_dynamic( - &headers, connection->allocator, 8, sizeof(struct aws_event_stream_header_value_pair))) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: error initializing headers %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); - return; - } - - if (aws_event_stream_message_headers(message, &headers)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: error fetching headers %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); - goto clean_up; - } - - int32_t stream_id = -1; - int32_t message_type = -1; - int32_t message_flags = -1; - - struct aws_byte_buf operation_name_buf; - AWS_ZERO_STRUCT(operation_name_buf); - if (aws_event_stream_rpc_extract_message_metadata( - &headers, &stream_id, &message_type, &message_flags, &operation_name_buf)) { - AWS_LOGF_ERROR( - AWS_LS_EVENT_STREAM_RPC_SERVER, - "id=%p: invalid protocol message with error %s", - (void *)connection, - aws_error_debug_str(aws_last_error())); - s_send_connection_level_error( - connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_error); - goto clean_up; - } - - AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: routing message", (void *)connection); - - s_route_message_by_type( - connection, - message, - &headers, - stream_id, - message_type, - message_flags, - aws_byte_cursor_from_buf(&operation_name_buf)); - - clean_up: - aws_event_stream_headers_list_cleanup(&headers); - } -} +/* + * Copyright 2010-2020 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. + */ + +#include <aws/event-stream/event_stream_channel_handler.h> +#include <aws/event-stream/event_stream_rpc_server.h> +#include <aws/event-stream/private/event_stream_rpc_priv.h> + +#include <aws/common/atomics.h> +#include <aws/common/hash_table.h> + +#include <aws/io/channel.h> +#include <aws/io/channel_bootstrap.h> + +#include <inttypes.h> + +#if defined(_MSC_VER) +/* allow non-constant aggregate initializer */ +# pragma warning(disable : 4204) +/* allow passing a pointer to an automatically allocated variable around, cause I'm smarter than the compiler. */ +# pragma warning(disable : 4221) +#endif + +static const struct aws_byte_cursor s_missing_operation_name_error = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL( + "{ \"message\": \"The first message for on a non-zero :stream-id must contain an operation header value.\"; }"); + +struct aws_event_stream_rpc_server_listener { + struct aws_allocator *allocator; + struct aws_socket *listener; + struct aws_server_bootstrap *bootstrap; + struct aws_atomic_var ref_count; + aws_event_stream_rpc_server_on_new_connection_fn *on_new_connection; + aws_event_stream_rpc_server_on_connection_shutdown_fn *on_connection_shutdown; + aws_event_stream_rpc_server_on_listener_destroy_fn *on_destroy_callback; + size_t initial_window_size; + bool enable_read_backpressure; + bool initialized; + void *user_data; +}; + +struct aws_event_stream_rpc_server_connection { + struct aws_allocator *allocator; + struct aws_hash_table continuation_table; + struct aws_event_stream_rpc_server_listener *server; + struct aws_atomic_var ref_count; + aws_event_stream_rpc_server_on_incoming_stream_fn *on_incoming_stream; + aws_event_stream_rpc_server_connection_protocol_message_fn *on_connection_protocol_message; + struct aws_channel *channel; + struct aws_channel_handler *event_stream_handler; + uint32_t latest_stream_id; + void *user_data; + struct aws_atomic_var is_open; + struct aws_atomic_var handshake_state; + bool bootstrap_owned; +}; + +struct aws_event_stream_rpc_server_continuation_token { + uint32_t stream_id; + struct aws_event_stream_rpc_server_connection *connection; + aws_event_stream_rpc_server_stream_continuation_fn *continuation_fn; + aws_event_stream_rpc_server_stream_continuation_closed_fn *closed_fn; + void *user_data; + struct aws_atomic_var ref_count; + struct aws_atomic_var is_closed; +}; + +/** This is the destructor callback invoked by the connections continuation table when a continuation is removed + * from the hash table. + */ +void s_continuation_destroy(void *value) { + struct aws_event_stream_rpc_server_continuation_token *continuation = value; + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying continuation", (void *)continuation); + continuation->closed_fn(continuation, continuation->user_data); + aws_event_stream_rpc_server_continuation_release(continuation); +} + +static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data); + +/* We have two paths for creating a connection on a channel. The first is an incoming connection on the server listener. + * The second is adding a connection to an already existing channel. This is the code common to both cases. */ +static struct aws_event_stream_rpc_server_connection *s_create_connection_on_channel( + struct aws_event_stream_rpc_server_listener *server, + struct aws_channel *channel) { + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: creating connection on channel %p", (void *)server, (void *)channel); + struct aws_event_stream_rpc_server_connection *connection = + aws_mem_calloc(server->allocator, 1, sizeof(struct aws_event_stream_rpc_server_connection)); + struct aws_channel_handler *event_stream_handler = NULL; + struct aws_channel_slot *slot = NULL; + + if (!connection) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: allocation failed for connection with error %s", + (void *)server, + aws_error_debug_str(aws_last_error())); + return NULL; + } + + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: new connection is %p", (void *)server, (void *)connection); + aws_atomic_init_int(&connection->ref_count, 1u); + aws_atomic_init_int(&connection->is_open, 1u); + /* handshake step 1 is a connect message being received. Handshake 2 is the connect ack being sent. + * no messages other than connect and connect ack are allowed until this count reaches 2. */ + aws_atomic_init_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_INITIALIZED); + connection->allocator = server->allocator; + + if (aws_hash_table_init( + &connection->continuation_table, + server->allocator, + 64, + aws_event_stream_rpc_hash_streamid, + aws_event_stream_rpc_streamid_eq, + NULL, + s_continuation_destroy)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: initialization of connection stream table failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + struct aws_event_stream_channel_handler_options handler_options = { + .on_message_received = s_on_message_received, + .user_data = connection, + .initial_window_size = server->initial_window_size, + .manual_window_management = server->enable_read_backpressure, + }; + + event_stream_handler = aws_event_stream_channel_handler_new(server->allocator, &handler_options); + + if (!event_stream_handler) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: initialization of event-stream handler failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + slot = aws_channel_slot_new(channel); + + if (!slot) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: initialization of channel slot failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + aws_channel_slot_insert_end(channel, slot); + if (aws_channel_slot_set_handler(slot, event_stream_handler)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: setting the handler on the slot failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto error; + } + + aws_event_stream_rpc_server_listener_acquire(server); + connection->server = server; + + connection->event_stream_handler = event_stream_handler; + connection->channel = channel; + aws_channel_acquire_hold(channel); + + return connection; + +error: + if (!slot && event_stream_handler) { + aws_channel_handler_destroy(event_stream_handler); + } + + if (connection) { + aws_event_stream_rpc_server_connection_release(connection); + } + + return NULL; +} + +struct aws_event_stream_rpc_server_connection *aws_event_stream_rpc_server_connection_from_existing_channel( + struct aws_event_stream_rpc_server_listener *server, + struct aws_channel *channel, + const struct aws_event_stream_rpc_connection_options *connection_options) { + AWS_FATAL_ASSERT( + connection_options->on_connection_protocol_message && "on_connection_protocol_message must be specified!"); + AWS_FATAL_ASSERT(connection_options->on_incoming_stream && "on_connection_protocol_message must be specified"); + + struct aws_event_stream_rpc_server_connection *connection = s_create_connection_on_channel(server, channel); + + if (!connection) { + return NULL; + } + + connection->on_incoming_stream = connection_options->on_incoming_stream; + connection->on_connection_protocol_message = connection_options->on_connection_protocol_message; + connection->user_data = connection_options->user_data; + aws_event_stream_rpc_server_connection_acquire(connection); + + return connection; +} + +void aws_event_stream_rpc_server_connection_acquire(struct aws_event_stream_rpc_server_connection *connection) { + size_t current_count = aws_atomic_fetch_add_explicit(&connection->ref_count, 1, aws_memory_order_relaxed); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: connection acquired, new ref count is %zu.", + (void *)connection, + current_count + 1); +} + +void aws_event_stream_rpc_server_connection_release(struct aws_event_stream_rpc_server_connection *connection) { + if (!connection) { + return; + } + + size_t value = aws_atomic_fetch_sub_explicit(&connection->ref_count, 1, aws_memory_order_seq_cst); + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: connection released, new ref count is %zu.", + (void *)connection, + value - 1); + if (value == 1) { + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying connection.", (void *)connection); + aws_channel_release_hold(connection->channel); + aws_hash_table_clean_up(&connection->continuation_table); + aws_event_stream_rpc_server_listener_release(connection->server); + aws_mem_release(connection->allocator, connection); + } +} + +/* incoming from a socket on this listener. */ +static void s_on_accept_channel_setup( + struct aws_server_bootstrap *bootstrap, + int error_code, + struct aws_channel *channel, + void *user_data) { + (void)bootstrap; + + struct aws_event_stream_rpc_server_listener *server = user_data; + + if (!error_code) { + AWS_LOGF_INFO( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: incoming connection with channel %p.", + (void *)server, + (void *)channel); + AWS_FATAL_ASSERT(channel && "Channel should never be null with a 0 error code."); + + struct aws_event_stream_rpc_server_connection *connection = s_create_connection_on_channel(server, channel); + + if (!connection) { + int error = aws_last_error(); + server->on_new_connection(NULL, error, NULL, server->user_data); + aws_channel_shutdown(channel, error); + } + + struct aws_event_stream_rpc_connection_options connection_options; + AWS_ZERO_STRUCT(connection_options); + + aws_event_stream_rpc_server_connection_acquire(connection); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: invoking on_new_connection with connection %p.", + (void *)server, + (void *)connection); + + if (server->on_new_connection(connection, AWS_ERROR_SUCCESS, &connection_options, server->user_data)) { + aws_channel_shutdown(channel, aws_last_error()); + aws_event_stream_rpc_server_connection_release(connection); + return; + } + + AWS_FATAL_ASSERT( + connection_options.on_connection_protocol_message && "on_connection_protocol_message must be specified!"); + AWS_FATAL_ASSERT(connection_options.on_incoming_stream && "on_connection_protocol_message must be specified"); + connection->on_incoming_stream = connection_options.on_incoming_stream; + connection->on_connection_protocol_message = connection_options.on_connection_protocol_message; + connection->user_data = connection_options.user_data; + connection->bootstrap_owned = true; + aws_event_stream_rpc_server_connection_release(connection); + + } else { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: invoking on_new_connection with error %s", + (void *)server, + aws_error_debug_str(error_code)); + server->on_new_connection(NULL, error_code, NULL, server->user_data); + } +} + +/* this is just to get the connection object off of the channel. */ +static inline struct aws_event_stream_rpc_server_connection *s_rpc_connection_from_channel( + struct aws_channel *channel) { + struct aws_channel_slot *our_slot = NULL; + struct aws_channel_slot *current_slot = aws_channel_get_first_slot(channel); + AWS_FATAL_ASSERT( + current_slot && + "It should be logically impossible to have a channel in this callback that doesn't have a slot in it"); + while (current_slot->adj_right) { + current_slot = current_slot->adj_right; + } + our_slot = current_slot; + struct aws_channel_handler *our_handler = our_slot->handler; + return aws_event_stream_channel_handler_get_user_data(our_handler); +} + +static void s_on_accept_channel_shutdown( + struct aws_server_bootstrap *bootstrap, + int error_code, + struct aws_channel *channel, + void *user_data) { + (void)bootstrap; + + struct aws_event_stream_rpc_server_listener *server = user_data; + struct aws_event_stream_rpc_server_connection *connection = s_rpc_connection_from_channel(channel); + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: channel %p and connection %p shutdown occurred with error %s", + (void *)server, + (void *)channel, + (void *)connection, + aws_error_debug_str(error_code)); + + aws_atomic_store_int(&connection->is_open, 0U); + aws_hash_table_clear(&connection->continuation_table); + aws_event_stream_rpc_server_connection_acquire(connection); + server->on_connection_shutdown(connection, error_code, server->user_data); + aws_event_stream_rpc_server_connection_release(connection); + aws_event_stream_rpc_server_connection_release(connection); +} + +static void s_on_server_listener_destroy(struct aws_server_bootstrap *bootstrap, void *user_data) { + (void)bootstrap; + struct aws_event_stream_rpc_server_listener *listener = user_data; + + AWS_LOGF_INFO(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying server", (void *)listener); + + /* server bootstrap invokes this callback regardless of if the listener was successfully created, so + * just check that we successfully set it up before freeing anything. When that's fixed in aws-c-io, this + * code will still be correct, so just leave it here for now. */ + if (listener->initialized) { + if (listener->on_destroy_callback) { + listener->on_destroy_callback(listener, listener->user_data); + } + + aws_mem_release(listener->allocator, listener); + } +} + +struct aws_event_stream_rpc_server_listener *aws_event_stream_rpc_server_new_listener( + struct aws_allocator *allocator, + struct aws_event_stream_rpc_server_listener_options *options) { + struct aws_event_stream_rpc_server_listener *server = + aws_mem_calloc(allocator, 1, sizeof(struct aws_event_stream_rpc_server_listener)); + + if (!server) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "static: failed to allocate new server with error %s", + aws_error_debug_str(aws_last_error())); + return NULL; + } + + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "static: new server is %p", (void *)server); + aws_atomic_init_int(&server->ref_count, 1); + + struct aws_server_socket_channel_bootstrap_options bootstrap_options = { + .bootstrap = options->bootstrap, + .socket_options = options->socket_options, + .tls_options = options->tls_options, + .enable_read_back_pressure = false, + .host_name = options->host_name, + .port = options->port, + .incoming_callback = s_on_accept_channel_setup, + .shutdown_callback = s_on_accept_channel_shutdown, + .destroy_callback = s_on_server_listener_destroy, + .user_data = server, + }; + + server->bootstrap = options->bootstrap; + server->allocator = allocator; + server->on_destroy_callback = options->on_destroy_callback; + server->on_new_connection = options->on_new_connection; + server->on_connection_shutdown = options->on_connection_shutdown; + server->user_data = options->user_data; + + server->listener = aws_server_bootstrap_new_socket_listener(&bootstrap_options); + + if (!server->listener) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "static: failed to allocate new socket listener with error %s", + aws_error_debug_str(aws_last_error())); + goto error; + } + + server->initialized = true; + return server; + +error: + if (server->listener) { + aws_server_bootstrap_destroy_socket_listener(options->bootstrap, server->listener); + } + + aws_mem_release(server->allocator, server); + return NULL; +} + +void aws_event_stream_rpc_server_listener_acquire(struct aws_event_stream_rpc_server_listener *server) { + size_t current_count = aws_atomic_fetch_add_explicit(&server->ref_count, 1, aws_memory_order_relaxed); + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: server acquired, new ref count is %zu.", + (void *)server, + current_count + 1); +} + +static void s_destroy_server(struct aws_event_stream_rpc_server_listener *server) { + if (server) { + AWS_LOGF_INFO(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying server", (void *)server); + /* the memory for this is cleaned up in the listener shutdown complete callback. */ + aws_server_bootstrap_destroy_socket_listener(server->bootstrap, server->listener); + } +} + +void aws_event_stream_rpc_server_listener_release(struct aws_event_stream_rpc_server_listener *server) { + if (!server) { + return; + } + + size_t ref_count = aws_atomic_fetch_sub_explicit(&server->ref_count, 1, aws_memory_order_seq_cst); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: server released, new ref count is %zu.", (void *)server, ref_count - 1); + + if (ref_count == 1) { + s_destroy_server(server); + } +} + +struct event_stream_connection_send_message_args { + struct aws_allocator *allocator; + struct aws_event_stream_message message; + enum aws_event_stream_rpc_message_type message_type; + struct aws_event_stream_rpc_server_connection *connection; + struct aws_event_stream_rpc_server_continuation_token *continuation; + aws_event_stream_rpc_server_message_flush_fn *flush_fn; + void *user_data; + bool end_stream; + bool terminate_connection; +}; + +static void s_on_protocol_message_written_fn( + struct aws_event_stream_message *message, + int error_code, + void *user_data) { + (void)message; + + struct event_stream_connection_send_message_args *message_args = user_data; + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: message flushed to channel with error %s", + (void *)message_args->connection, + aws_error_debug_str(error_code)); + + if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { + AWS_LOGF_INFO( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: connect ack message sent, the connect handshake is completed", + (void *)message_args->connection); + aws_atomic_store_int( + &message_args->connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED); + } + + if (message_args->end_stream) { + AWS_FATAL_ASSERT(message_args->continuation && "end stream flag was set but it wasn't on a continuation"); + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: end_stream flag for continuation %p was set, closing", + (void *)message_args->connection, + (void *)message_args->continuation); + aws_atomic_store_int(&message_args->continuation->is_closed, 1U); + aws_hash_table_remove( + &message_args->connection->continuation_table, &message_args->continuation->stream_id, NULL, NULL); + } + + message_args->flush_fn(error_code, message_args->user_data); + + if (message_args->terminate_connection) { + AWS_LOGF_INFO( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: terminate connection flag was set. closing", + (void *)message_args->connection); + aws_event_stream_rpc_server_connection_close(message_args->connection, AWS_ERROR_SUCCESS); + } + + aws_event_stream_rpc_server_connection_release(message_args->connection); + + if (message_args->continuation) { + aws_event_stream_rpc_server_continuation_release(message_args->continuation); + } + + aws_event_stream_message_clean_up(&message_args->message); + aws_mem_release(message_args->allocator, message_args); +} + +static int s_send_protocol_message( + struct aws_event_stream_rpc_server_connection *connection, + struct aws_event_stream_rpc_server_continuation_token *continuation, + const struct aws_event_stream_rpc_message_args *message_args, + int32_t stream_id, + aws_event_stream_rpc_server_message_flush_fn *flush_fn, + void *user_data) { + + size_t connect_handshake_state = aws_atomic_load_int(&connection->handshake_state); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: connect handshake state %zu", + (void *)connection, + connect_handshake_state); + /* handshake step 1 is a connect message being received. Handshake 2 is the connect ack being sent. + * no messages other than connect and connect ack are allowed until this count reaches 2. */ + if (connect_handshake_state != CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && + message_args->message_type < AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: invalid state, a message was received prior to connect handshake completion", + (void *)connection); + return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + } + + struct event_stream_connection_send_message_args *args = + aws_mem_calloc(connection->allocator, 1, sizeof(struct event_stream_connection_send_message_args)); + + if (!message_args) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: allocation of callback args failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + return AWS_OP_ERR; + } + + args->allocator = connection->allocator; + args->user_data = user_data; + args->message_type = message_args->message_type; + args->connection = connection; + args->flush_fn = flush_fn; + + if (continuation) { + args->continuation = continuation; + aws_event_stream_rpc_server_continuation_acquire(continuation); + + if (message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: continuation with terminate stream flag was specified closing", + (void *)continuation); + args->end_stream = true; + } + } + + if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK && + !(message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_CONNECTION_ACCEPTED)) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: connection ack was rejected closing connection", + (void *)connection); + args->terminate_connection = true; + } + + args->flush_fn = flush_fn; + + size_t headers_count = message_args->headers_count + 3; + struct aws_array_list headers_list; + AWS_ZERO_STRUCT(headers_list); + + if (aws_array_list_init_dynamic( + &headers_list, connection->allocator, headers_count, sizeof(struct aws_event_stream_header_value_pair))) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: allocation of headers failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto args_allocated_before_failure; + } + + /* since we preallocated the space for the headers, these can't fail, but we'll go ahead an assert on them just in + * case */ + for (size_t i = 0; i < message_args->headers_count; ++i) { + AWS_FATAL_ASSERT(!aws_array_list_push_back(&headers_list, &message_args->headers[i])); + } + + AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( + &headers_list, + (const char *)aws_event_stream_rpc_message_type_name.ptr, + (uint8_t)aws_event_stream_rpc_message_type_name.len, + message_args->message_type)); + AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( + &headers_list, + (const char *)aws_event_stream_rpc_message_flags_name.ptr, + (uint8_t)aws_event_stream_rpc_message_flags_name.len, + message_args->message_flags)); + AWS_FATAL_ASSERT(!aws_event_stream_add_int32_header( + &headers_list, + (const char *)aws_event_stream_rpc_stream_id_name.ptr, + (uint8_t)aws_event_stream_rpc_stream_id_name.len, + stream_id)); + + int message_init_err_code = + aws_event_stream_message_init(&args->message, connection->allocator, &headers_list, message_args->payload); + aws_array_list_clean_up(&headers_list); + + if (message_init_err_code) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: initialization of message failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto args_allocated_before_failure; + } + + aws_event_stream_rpc_server_connection_acquire(connection); + + if (aws_event_stream_channel_handler_write_message( + connection->event_stream_handler, &args->message, s_on_protocol_message_written_fn, args)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: message send failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + goto message_initialized_before_failure; + } + + return AWS_OP_SUCCESS; + +message_initialized_before_failure: + aws_event_stream_message_clean_up(&args->message); + +args_allocated_before_failure: + aws_mem_release(args->allocator, args); + aws_event_stream_rpc_server_connection_release(connection); + + return AWS_OP_ERR; +} + +int aws_event_stream_rpc_server_connection_send_protocol_message( + struct aws_event_stream_rpc_server_connection *connection, + const struct aws_event_stream_rpc_message_args *message_args, + aws_event_stream_rpc_server_message_flush_fn *flush_fn, + void *user_data) { + if (!aws_event_stream_rpc_server_connection_is_open(connection)) { + return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_CONNECTION_CLOSED); + } + + return s_send_protocol_message(connection, NULL, message_args, 0, flush_fn, user_data); +} + +void *aws_event_stream_rpc_server_connection_get_user_data(struct aws_event_stream_rpc_server_connection *connection) { + return connection->user_data; +} + +AWS_EVENT_STREAM_API void aws_event_stream_rpc_server_override_last_stream_id( + struct aws_event_stream_rpc_server_connection *connection, + int32_t value) { + connection->latest_stream_id = value; +} + +void aws_event_stream_rpc_server_connection_close( + struct aws_event_stream_rpc_server_connection *connection, + int shutdown_error_code) { + + if (aws_event_stream_rpc_server_connection_is_open(connection)) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: closing connection with error %s", + (void *)connection, + aws_error_debug_str(shutdown_error_code)); + aws_atomic_store_int(&connection->is_open, 0U); + aws_channel_shutdown(connection->channel, shutdown_error_code); + + if (!connection->bootstrap_owned) { + aws_hash_table_clear(&connection->continuation_table); + aws_event_stream_rpc_server_connection_release(connection); + } + } +} + +bool aws_event_stream_rpc_server_continuation_is_closed( + struct aws_event_stream_rpc_server_continuation_token *continuation) { + return aws_atomic_load_int(&continuation->is_closed) == 1U; +} + +bool aws_event_stream_rpc_server_connection_is_open(struct aws_event_stream_rpc_server_connection *connection) { + return aws_atomic_load_int(&connection->is_open) == 1U; +} + +void aws_event_stream_rpc_server_continuation_acquire( + struct aws_event_stream_rpc_server_continuation_token *continuation) { + size_t current_count = aws_atomic_fetch_add_explicit(&continuation->ref_count, 1, aws_memory_order_relaxed); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: continuation acquired, new ref count is %zu.", + (void *)continuation, + current_count + 1); +} + +void aws_event_stream_rpc_server_continuation_release( + struct aws_event_stream_rpc_server_continuation_token *continuation) { + size_t value = aws_atomic_fetch_sub_explicit(&continuation->ref_count, 1, aws_memory_order_seq_cst); + + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: continuation released, new ref count is %zu.", + (void *)continuation, + value - 1); + + if (value == 1) { + AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: destroying continuation.", (void *)continuation); + struct aws_allocator *allocator = continuation->connection->allocator; + aws_event_stream_rpc_server_connection_release(continuation->connection); + aws_mem_release(allocator, continuation); + } +} + +int aws_event_stream_rpc_server_continuation_send_message( + struct aws_event_stream_rpc_server_continuation_token *continuation, + const struct aws_event_stream_rpc_message_args *message_args, + aws_event_stream_rpc_server_message_flush_fn *flush_fn, + void *user_data) { + AWS_FATAL_PRECONDITION(continuation->continuation_fn); + AWS_FATAL_PRECONDITION(continuation->closed_fn); + + if (aws_event_stream_rpc_server_continuation_is_closed(continuation)) { + return aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_STREAM_CLOSED); + } + + return s_send_protocol_message( + continuation->connection, continuation, message_args, continuation->stream_id, flush_fn, user_data); +} + +static void s_connection_error_message_flush_fn(int error_code, void *user_data) { + (void)error_code; + + struct aws_event_stream_rpc_server_connection *connection = user_data; + aws_event_stream_rpc_server_connection_close(connection, AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); +} + +static void s_send_connection_level_error( + struct aws_event_stream_rpc_server_connection *connection, + uint32_t message_type, + uint32_t message_flags, + const struct aws_byte_cursor *message) { + struct aws_byte_buf payload_buf = aws_byte_buf_from_array(message->ptr, message->len); + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: sending connection-level error\n" PRInSTR, + (void *)connection, + AWS_BYTE_BUF_PRI(payload_buf)); + + struct aws_event_stream_header_value_pair content_type_header = + aws_event_stream_create_string_header(s_json_content_type_name, s_json_content_type_value); + + struct aws_event_stream_header_value_pair headers[] = { + content_type_header, + }; + + struct aws_event_stream_rpc_message_args message_args = { + .message_type = message_type, + .message_flags = message_flags, + .payload = &payload_buf, + .headers_count = 1, + .headers = headers, + }; + + aws_event_stream_rpc_server_connection_send_protocol_message( + connection, &message_args, s_connection_error_message_flush_fn, connection); +} + +/* TODO: come back and make this a proper state pattern. For now it's branches all over the place until we nail + * down the spec. */ +static void s_route_message_by_type( + struct aws_event_stream_rpc_server_connection *connection, + struct aws_event_stream_message *message, + struct aws_array_list *headers_list, + uint32_t stream_id, + uint32_t message_type, + uint32_t message_flags, + struct aws_byte_cursor operation_name) { + struct aws_byte_buf payload_buf = aws_byte_buf_from_array( + aws_event_stream_message_payload(message), aws_event_stream_message_payload_len(message)); + + struct aws_event_stream_rpc_message_args message_args = { + .headers = headers_list->data, + .headers_count = aws_array_list_length(headers_list), + .payload = &payload_buf, + .message_flags = message_flags, + .message_type = message_type, + }; + + size_t handshake_state = aws_atomic_load_int(&connection->handshake_state); + + /* make sure if this is not a CONNECT message being received, the handshake has been completed. */ + if (handshake_state < CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED && + message_type != AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: a message was received on this connection prior to the " + "connect handshake completing", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); + return; + } + + /* stream_id being non zero ALWAYS indicates APPLICATION_DATA or APPLICATION_ERROR. */ + if (stream_id > 0) { + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: stream id %" PRIu32, (void *)connection, stream_id); + + struct aws_event_stream_rpc_server_continuation_token *continuation = NULL; + if (message_type > AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: only application messages can be sent on a stream id, " + "but this message is the incorrect type", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_stream_id_error); + return; + } + + /* INT32_MAX is the max stream id. */ + if (stream_id > INT32_MAX) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: stream_id is larger than the max acceptable value", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_stream_id_error); + return; + } + + /* if the stream is is in the past, look it up from the continuation table. If it's not there, that's an error. + * if it is, find it and notify the user a message arrived */ + if (stream_id <= connection->latest_stream_id) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: stream_id is an already seen stream_id, looking for existing continuation", + (void *)connection); + + struct aws_hash_element *continuation_element = NULL; + if (aws_hash_table_find(&connection->continuation_table, &stream_id, &continuation_element) || + !continuation_element) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: stream_id does not have a corresponding continuation", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_client_stream_id_error); + return; + } + + continuation = continuation_element->value; + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: stream_id corresponds to continuation %p", + (void *)connection, + (void *)continuation); + + aws_event_stream_rpc_server_continuation_acquire(continuation); + continuation->continuation_fn(continuation, &message_args, continuation->user_data); + aws_event_stream_rpc_server_continuation_release(continuation); + /* now these are potentially new streams. Make sure they're in bounds, create a new continuation + * and notify the user the stream has been created, then send them the message. */ + } else { + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: stream_id is unknown, attempting to create a continuation for it", + (void *)connection); + if (stream_id != connection->latest_stream_id + 1) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: stream_id is invalid because it's not sequentially increasing", + (void *)connection); + + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, + AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, + 0, + &s_invalid_new_client_stream_id_error); + return; + } + + /* new streams must always have an operation name. */ + if (operation_name.len == 0) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: new stream_id encountered, but an operation name was not received", + (void *)connection); + aws_raise_error(AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_missing_operation_name_error); + return; + } + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: stream_id is a valid new stream. Creating continuation", + (void *)connection); + continuation = + aws_mem_calloc(connection->allocator, 1, sizeof(struct aws_event_stream_rpc_server_continuation_token)); + if (!continuation) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: continuation allocation failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); + return; + } + + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: new continuation is %p", + (void *)connection, + (void *)continuation); + + continuation->stream_id = stream_id; + continuation->connection = connection; + aws_event_stream_rpc_server_connection_acquire(continuation->connection); + aws_atomic_init_int(&continuation->ref_count, 1); + + if (aws_hash_table_put(&connection->continuation_table, &continuation->stream_id, continuation, NULL)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: continuation table update failed with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + /* continuation release will drop the connection reference as well */ + aws_event_stream_rpc_server_continuation_release(continuation); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); + return; + } + + struct aws_event_stream_rpc_server_stream_continuation_options options; + AWS_ZERO_STRUCT(options); + + aws_event_stream_rpc_server_continuation_acquire(continuation); + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: invoking on_incoming_stream callback", (void *)connection); + if (connection->on_incoming_stream( + continuation->connection, continuation, operation_name, &options, connection->user_data)) { + aws_event_stream_rpc_server_continuation_release(continuation); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); + return; + } + AWS_FATAL_ASSERT(options.on_continuation); + AWS_FATAL_ASSERT(options.on_continuation_closed); + + continuation->continuation_fn = options.on_continuation; + continuation->closed_fn = options.on_continuation_closed; + continuation->user_data = options.user_data; + + connection->latest_stream_id = stream_id; + continuation->continuation_fn(continuation, &message_args, continuation->user_data); + aws_event_stream_rpc_server_continuation_release(continuation); + } + + /* if it was a terminal stream message purge it from the hash table. The delete will decref the continuation. */ + if (message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM) { + AWS_LOGF_DEBUG( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: the terminate_stream flag was received for continuation %p, closing", + (void *)connection, + (void *)continuation); + aws_atomic_store_int(&continuation->is_closed, 1U); + aws_hash_table_remove(&connection->continuation_table, &stream_id, NULL, NULL); + } + } else { + if (message_type <= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_ERROR || + message_type >= AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_COUNT) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: a zero stream id was received with an invalid message-type %" PRIu32, + (void *)connection, + message_type); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_type_error); + return; + } + + if (message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) { + if (handshake_state) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: connect received but the handshake is already completed. Only one is allowed.", + (void *)connection); + /* only one connect is allowed. This would be a duplicate. */ + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_connect_not_completed_error); + return; + } + aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED); + AWS_LOGF_INFO( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: connect received, connection handshake completion pending the server sending an ack.", + (void *)connection); + } + + connection->on_connection_protocol_message(connection, &message_args, connection->user_data); + } +} + +/* invoked by the event stream channel handler when a complete message has been read from the channel. */ +static void s_on_message_received(struct aws_event_stream_message *message, int error_code, void *user_data) { + + if (!error_code) { + struct aws_event_stream_rpc_server_connection *connection = user_data; + AWS_LOGF_TRACE( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: message received on connection of length %" PRIu32, + (void *)connection, + aws_event_stream_message_total_length(message)); + + struct aws_array_list headers; + if (aws_array_list_init_dynamic( + &headers, connection->allocator, 8, sizeof(struct aws_event_stream_header_value_pair))) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: error initializing headers %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); + return; + } + + if (aws_event_stream_message_headers(message, &headers)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: error fetching headers %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_INTERNAL_ERROR, 0, &s_internal_error); + goto clean_up; + } + + int32_t stream_id = -1; + int32_t message_type = -1; + int32_t message_flags = -1; + + struct aws_byte_buf operation_name_buf; + AWS_ZERO_STRUCT(operation_name_buf); + if (aws_event_stream_rpc_extract_message_metadata( + &headers, &stream_id, &message_type, &message_flags, &operation_name_buf)) { + AWS_LOGF_ERROR( + AWS_LS_EVENT_STREAM_RPC_SERVER, + "id=%p: invalid protocol message with error %s", + (void *)connection, + aws_error_debug_str(aws_last_error())); + s_send_connection_level_error( + connection, AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_PROTOCOL_ERROR, 0, &s_invalid_message_error); + goto clean_up; + } + + AWS_LOGF_TRACE(AWS_LS_EVENT_STREAM_RPC_SERVER, "id=%p: routing message", (void *)connection); + + s_route_message_by_type( + connection, + message, + &headers, + stream_id, + message_type, + message_flags, + aws_byte_cursor_from_buf(&operation_name_buf)); + + clean_up: + aws_event_stream_headers_list_cleanup(&headers); + } +} |