napi_value aws_napi_mqtt_client_connection_connect()

in source/mqtt_client_connection.c [637:790]

• 128 lines of code
• 12 McCabe index (conditional complexity)

napi_value aws_napi_mqtt_client_connection_connect(napi_env env, napi_callback_info cb_info) {

    bool success = false;

    struct aws_socket_options *socket_options = NULL;
    struct mqtt_connection_binding *binding = NULL;

    struct aws_byte_buf client_id;
    AWS_ZERO_STRUCT(client_id);
    struct aws_byte_buf server_name;
    AWS_ZERO_STRUCT(server_name);
    struct connect_args *on_connect_args = NULL;

    napi_value node_args[10];
    size_t num_args = AWS_ARRAY_SIZE(node_args);
    napi_value *arg = &node_args[0];
    AWS_NAPI_CALL(env, napi_get_cb_info(env, cb_info, &num_args, node_args, NULL, NULL), {
        napi_throw_error(env, NULL, "Failed to retreive callback information");
        goto cleanup;
    });
    if (num_args != AWS_ARRAY_SIZE(node_args)) {
        napi_throw_error(env, NULL, "mqtt_client_connection_connect needs exactly 10 arguments");
        goto cleanup;
    }

    napi_value node_binding = *arg++;
    AWS_NAPI_CALL(env, napi_get_value_external(env, node_binding, (void **)&binding), {
        napi_throw_error(env, NULL, "Failed to extract connection from first argument");
        goto cleanup;
    });

    struct aws_allocator *allocator = binding->allocator;

    napi_value node_client_id = *arg++;
    AWS_NAPI_CALL(env, aws_byte_buf_init_from_napi(&client_id, env, node_client_id), {
        napi_throw_type_error(env, NULL, "client_id must be a String");
        goto cleanup;
    });

    napi_value node_server_name = *arg++;
    AWS_NAPI_CALL(env, aws_byte_buf_init_from_napi(&server_name, env, node_server_name), {
        napi_throw_type_error(env, NULL, "server_name must be a String");
        goto cleanup;
    });

    napi_value node_port = *arg++;
    uint32_t port_number = 0;
    AWS_NAPI_CALL(env, napi_get_value_uint32(env, node_port, &port_number), {
        napi_throw_type_error(env, NULL, "port must be a Number");
        goto cleanup;
    });

    napi_value node_socket_options = *arg++;
    if (!aws_napi_is_null_or_undefined(env, node_socket_options)) {
        AWS_NAPI_CALL(env, napi_get_value_external(env, node_socket_options, (void **)&socket_options), {
            napi_throw_type_error(env, NULL, "connect_timeout must be a Number");
            goto cleanup;
        });
    }

    napi_value node_keep_alive_time = *arg++;
    uint32_t keep_alive_time = 0;
    if (!aws_napi_is_null_or_undefined(env, node_keep_alive_time)) {
        AWS_NAPI_CALL(env, napi_get_value_uint32(env, node_keep_alive_time, &keep_alive_time), {
            napi_throw_type_error(env, NULL, "keep_alive must be a Number");
            goto cleanup;
        });
    }

    napi_value node_ping_timeout = *arg++;
    uint32_t ping_timeout = 0;
    if (!aws_napi_is_null_or_undefined(env, node_ping_timeout)) {
        AWS_NAPI_CALL(env, napi_get_value_uint32(env, node_ping_timeout, &ping_timeout), {
            napi_throw_type_error(env, NULL, "ping_timeout must be a Number");
            goto cleanup;
        });
    }

    napi_value node_protocol_operation_timeout = *arg++;
    uint32_t protocol_operation_timeout = 0;
    if (!aws_napi_is_null_or_undefined(env, node_protocol_operation_timeout)) {
        AWS_NAPI_CALL(env, napi_get_value_uint32(env, node_protocol_operation_timeout, &protocol_operation_timeout), {
            napi_throw_type_error(env, NULL, "protocol_operation_timeout must be a Number");
            goto cleanup;
        });
    }

    napi_value node_clean_session = *arg++;
    bool clean_session = false;
    if (!aws_napi_is_null_or_undefined(env, node_clean_session)) {
        AWS_NAPI_CALL(env, napi_get_value_bool(env, node_clean_session, &clean_session), {
            napi_throw_type_error(env, NULL, "clean_session must be a boolean");
            goto cleanup;
        });
    }

    napi_value node_on_connect = *arg++;
    if (!aws_napi_is_null_or_undefined(env, node_on_connect)) {

        on_connect_args = aws_mem_calloc(allocator, 1, sizeof(struct connect_args));
        AWS_FATAL_ASSERT(on_connect_args);
        on_connect_args->allocator = allocator;
        on_connect_args->binding = binding;

        AWS_NAPI_CALL(
            env,
            aws_napi_create_threadsafe_function(
                env,
                node_on_connect,
                "aws_mqtt_client_connection_on_connect",
                s_on_connect_call,
                binding,
                &on_connect_args->on_connect),
            {
                napi_throw_error(env, NULL, "Failed to bind on_connect callback");
                goto cleanup;
            });
    }

    struct aws_byte_cursor client_id_cur = aws_byte_cursor_from_buf(&client_id);
    struct aws_byte_cursor server_name_cur = aws_byte_cursor_from_buf(&server_name);

    struct aws_mqtt_connection_options options;
    options.clean_session = clean_session;
    options.client_id = client_id_cur;
    options.host_name = server_name_cur;
    options.keep_alive_time_secs = (uint16_t)keep_alive_time;
    options.on_connection_complete = s_on_connected;
    options.ping_timeout_ms = ping_timeout;
    options.protocol_operation_timeout_ms = protocol_operation_timeout;
    options.port = (uint16_t)port_number;

    options.socket_options = socket_options;
    options.tls_options = binding->use_tls_options ? &binding->tls_options : NULL;
    options.user_data = on_connect_args; /* on_connect user_data */

    if (aws_mqtt_client_connection_connect(binding->connection, &options)) {
        aws_napi_throw_last_error(env);
        goto cleanup;
    }

    success = true;

cleanup:

    aws_byte_buf_clean_up(&client_id);
    aws_byte_buf_clean_up(&server_name);

    if (!success && on_connect_args) {
        s_destroy_connect_args(on_connect_args);
    }

    return NULL;
}