MQTT GetID Failure. cURL Code 28

[ngiustini-invn]

Hello,

Using release version 1.9.105

I am currently facing an issue with MQTT functionality with my embedded system built on C/C++. When trying to get Cognito details to be used for MQTT communication, it seems that the library fails when trying to get the ID returning a cURL code 28. This indicates that it is timing out. In our code we then try to restart the library which only leads to the same cURL code 28 in an infinite loop. This only seems to happen on some networks with higher latency for its upload and download values as I have tested on other networks which see the cURL operation from the library working without issue. I am not aware of any settings we can set to change the timeout settings or other QoS settings that may help to ensure that the network calls are successful. Is anyone able to help point to any settings that may help solve this problem? Alternatively, is there any other solution that can be suggested to avoid seeing the request time out? Does AWS’’s library have any expectation about network conditions or requirements that must be met for the library to properly function?

Please see below some sample code for how we are setting up and using the MQTT connection.

#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/time.h>
#include <unistd.h>
#include <linux/ioctl.h>
#include <math.h>
#include <sys/ioctl.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <math.h>
#include <curl/curl.h>
#include <json-c/json.h>

#include <algorithm>
#include <aws/cognito-identity/CognitoIdentityClient.h>
#include <aws/cognito-identity/CognitoIdentityErrors.h>
#include <aws/cognito-identity/model/CreateIdentityPoolRequest.h>
#include <aws/cognito-identity/model/DeleteIdentityPoolRequest.h>
#include <aws/cognito-identity/model/DescribeIdentityPoolRequest.h>
#include <aws/cognito-identity/model/GetIdentityPoolRolesRequest.h>
#include <aws/cognito-identity/model/UpdateIdentityPoolRequest.h>
#include <aws/cognito-identity/model/ListIdentityPoolsRequest.h>
#include <aws/cognito-identity/model/GetCredentialsForIdentityRequest.h>
#include <aws/cognito-identity/model/GetIdRequest.h>
#include <aws/cognito-identity/model/ListIdentitiesRequest.h>
#include <aws/cognito-identity/model/GetOpenIdTokenRequest.h>
#include <aws/cognito-identity/model/UnlinkIdentityRequest.h>
#include <aws/cognito-identity/model/GetOpenIdTokenForDeveloperIdentityRequest.h>
#include <aws/cognito-identity/model/LookupDeveloperIdentityRequest.h>
#include <aws/cognito-identity/CognitoIdentityErrors.h>
#include <aws/access-management/AccessManagementClient.h>
#include <aws/iam/IAMClient.h>
#include <aws/core/client/CoreErrors.h>
#include <aws/core/utils/json/JsonSerializer.h>
#include <aws/core/utils/Outcome.h>
#include <aws/crt/auth/Credentials.h>
#include <aws/crt/Types.h>
#include <aws/iot/MqttClient.h>
#include <aws/crt/Api.h>
#include <aws/core/Aws.h>

#define MAX_MQTT_PAYLOAD_LEN (1024)

struct aws_cog_mqtt_msg_t{
	int type;
	int payload_len;
	char payload[MAX_MQTT_PAYLOAD_LEN];
};

int run_flag = 1;

static int aws_cog_mqtt_start(const char* cognitoPoolId, const char* region, const char* iotEndPoint, const char* client_id, const char* current_topic, void* hook_func)
{
	int ret;

	//start mqtt request
	if ((cognitoPoolId == NULL) || (region == NULL) || (iotEndPoint == NULL)) {
		printf("either cognitoPoolId, region or iotEndPoint is NULL! return...\n");
		return -1;
	}

	//get cognito credentials
	GetIdRequest getIdRequest;
	getIdRequest.WithIdentityPoolId(cognitoPoolId);

	ClientConfiguration clientConfig;
	clientConfig.region = Aws::Region::US_EAST_1;

	auto client = Aws::MakeShared<Aws::CognitoIdentity::CognitoIdentityClient>("cognito_client", clientConfig);

	Aws::String access_key;
	Aws::String secret_key;
	Aws::String session_token;

	GetIdOutcome getIdOutcome = client->GetId(getIdRequest);
	if (getIdOutcome.IsSuccess()) {
		GetCredentialsForIdentityRequest getCredentialsRequest;
		getCredentialsRequest.WithIdentityId(getIdOutcome.GetResult().GetIdentityId());
		GetCredentialsForIdentityOutcome getCredentialsOutcome = client->GetCredentialsForIdentity(getCredentialsRequest);
		access_key = getCredentialsOutcome.GetResult().GetCredentials().GetAccessKeyId();
		secret_key = getCredentialsOutcome.GetResult().GetCredentials().GetSecretKey();
		session_token = getCredentialsOutcome.GetResult().GetCredentials().GetSessionToken();
		if (getCredentialsOutcome.IsSuccess()) {
		} else {
			printf("GetCredentials failed : %s\n", getCredentialsOutcome.GetError().GetMessage().c_str());
			return -1;
		}
	} else {
		printf(" GetId failed : %s\n", getIdOutcome.GetError().GetMessage().c_str());
		return -1;
	}

	//build credential handler
	Aws::Crt::ApiHandle apiHandle;
	Aws::Crt::Auth::CredentialsProviderStaticConfig defaultConfig;
	std::shared_ptr<Aws::Crt::Auth::ICredentialsProvider> provider = nullptr;

	defaultConfig.AccessKeyId = Aws::Crt::ByteCursorFromCString(access_key.c_str());
	defaultConfig.SecretAccessKey = Aws::Crt::ByteCursorFromCString(secret_key.c_str());
	defaultConfig.SessionToken = Aws::Crt::ByteCursorFromCString(session_token.c_str());

	provider = Aws::Crt::Auth::CredentialsProvider::CreateCredentialsProviderStatic(defaultConfig);

	Aws::Iot::MqttClientConnectionConfigBuilder builder;
	Aws::Iot::WebsocketConfig config(region, provider);
	builder = Aws::Iot::MqttClientConnectionConfigBuilder(config);

	//set up mqtt client
	/********************** Now Setup an Mqtt Client ******************/
	/*
	* You need an event loop group to process IO events.
	* If you only have a few connections, 1 thread is ideal
	*/
	Aws::Crt::Io::EventLoopGroup eventLoopGroup(1);
	if (!eventLoopGroup) {
		printf("Event Loop Group Creation failed with error %s\n", ErrorDebugString(eventLoopGroup.LastError()));
		return -1;
	}

	Aws::Crt::Io::DefaultHostResolver defaultHostResolver(eventLoopGroup, 1, 5);
	Aws::Crt::Io::ClientBootstrap bootstrap(eventLoopGroup, defaultHostResolver);

	if (!bootstrap) {
		printf("ClientBootstrap failed with error %s\n", ErrorDebugString(bootstrap.LastError()));
		return -1;
	}

	Aws::Crt::String endpoint = iotEndPoint;
	builder.WithEndpoint(endpoint);

	auto clientConfig1 = builder.Build();
	if (!clientConfig1) {
		printf("Client Configuration initialization failed with error %s\n", ErrorDebugString(clientConfig1.LastError()));
		return -1;
	}

	Aws::Iot::MqttClient mqttClient(bootstrap);
	/*
	* Since no exceptions are used, always check the bool operator
	* when an error could have occurred.
	*/
	if (!mqttClient) {
		debug_print("MQTT Client Creation failed with error %s\n", ErrorDebugString(mqttClient.LastError()));
		return -1;
	}

	/*
	* Now create a connection object. Note: This type is move only
	* and its underlying memory is managed by the client.
	*/
	auto connection = mqttClient.NewConnection(clientConfig1);
	if (!connection) {
		printf("MQTT Connection Creation failed with error %s\n", ErrorDebugString(mqttClient.LastError()));
		return -1;
	}

	//connect to server
	/*
	* TODO: Not to enforce synchronous behavior and no to do with a condition variable.
	*/
	std::promise<bool> connectionCompletedPromise;
	std::promise<void> connectionClosedPromise;

	/*
	* This will execute when an mqtt connect has completed or failed.
	*/
	auto onConnectionCompleted = [&](Mqtt::MqttConnection &, int errorCode, Mqtt::ReturnCode returnCode, bool) {
		if (errorCode) {
			printf("Connection failed with error %s\n", ErrorDebugString(errorCode));
			connectionCompletedPromise.set_value(false);
        	} else {
			printf("errorCode 0 good!\n");
			if (returnCode != AWS_MQTT_CONNECT_ACCEPTED) {
				perror("Connection failed with mqtt return code %d\n", (int)returnCode);
				connectionCompletedPromise.set_value(false);
			} else {
				printf("Connection completed successfully.\n");
				connectionCompletedPromise.set_value(true);
			}
        	}
	};

	auto onInterrupted = [&](Mqtt::MqttConnection &, int error) {
		printf("Connection interrupted with error %s\n", ErrorDebugString(error));
		//return -1;	//Don't retry with interrupt, will recover/resume
	};

	auto onResumed = [&](Mqtt::MqttConnection &, Mqtt::ReturnCode, bool) { printf("Connection resumed\n"); };

	/*
	* Invoked when a disconnect message has completed.
	*/
	auto onDisconnect = [&](Mqtt::MqttConnection &) {
		printf("Disconnect completed\n");
		connectionClosedPromise.set_value();
	};

	connection->OnConnectionCompleted = std::move(onConnectionCompleted);
	connection->OnDisconnect = std::move(onDisconnect);
	connection->OnConnectionInterrupted = std::move(onInterrupted);
	connection->OnConnectionResumed = std::move(onResumed);

	/*
	* Actually perform the connect dance.
	* This will use default ping behavior of 1 hour and 3 second timeouts.
	* If you want different behavior, those arguments go into slots 3 & 4.
	*/
	printf("Connecting...\n");
	if (!connection->Connect(client_id, false, 1000)) {
		printf("MQTT Connection failed with error %s\n", ErrorDebugString(connection->LastError()));
		return -1;
	}

	//attach publish/subscribe cb
	String topic(current_topic);
	if (connectionCompletedPromise.get_future().get()) {
		std::mutex receiveMutex;
		std::condition_variable receiveSignal;
		uint32_t receivedCount = 0;

        	/*
         	* This is invoked upon the receipt of a Publish on a subscribed topic.
         	*/
        	auto onMessage = [&](Mqtt::MqttConnection &,
                             const String &topic,
                             const ByteBuf &byteBuf,
                             bool /*dup*/,
                             Mqtt::QOS /*qos*/,
                             bool /*retain*/) {
			{
				std::lock_guard<std::mutex> lock(receiveMutex);
				++receivedCount;
				printf("Message: %s\n", byteBuf.buffer);
			}
			receiveSignal.notify_all();
		};

		/*
		* Subscribe for incoming publish messages on topic.
		*/
		std::promise<void> subscribeFinishedPromise;
		auto onSubAck = [&](Mqtt::MqttConnection &, uint16_t packetId, const String &topic, Mqtt::QOS QoS, int errorCode) {
			if (errorCode) {
				printf("Subscribe failed with error %s\n", aws_error_debug_str(errorCode));
			} else {
				if (!packetId || QoS == AWS_MQTT_QOS_FAILURE) {
					printf("Subscribe rejected by the broker.\n");
				} else {
                    //do nothing
				}
			}
			subscribeFinishedPromise.set_value();
		};

		connection->Subscribe(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, onMessage, onSubAck);
		subscribeFinishedPromise.get_future().wait();
		init_status = 1;

		if (hook_func != NULL) {
			printf("Executing mqtt hook func...\n")
			((void(*)(void))hook_func)();
		}

		//publish and subscribe loop
		{
			struct aws_cog_mqtt_msg_t msg;
			int cnt = 0;
			int ret;

			while (run_flag) {
				memset(&msg, 0, sizeof(msg));
                //recieve data from the msg queue so it can be sent to the mqtt server
				ret = msgrcv(aws_cog_mqtt_msgid, &msg, sizeof(msg), 0, 0);
				if(-1 == ret) {
					printf("######## mqtt msgrcv error\n");
				}
                printf("cnt=%d, payload_len=%d, payload=%s\n", cnt++, msg.payload_len, msg.payload);
                    ByteBuf payload = ByteBufFromArray((const uint8_t *)msg.payload, msg.payload_len);
                    printf("Start publishing %s\n", payload.buffer);
                    auto onPublishComplete = [](Mqtt::MqttConnection &, uint16_t v1, int v2) {printf("Publishing msg complete, v1 = %u, v2 = %d\n", v1, v2);};
                    connection->Publish(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, false, payload, onPublishComplete);
			}

			init_status = 0;
			printf("start mqtt exit wait...\n");
            std::unique_lock<std::mutex> receivedLock(receiveMutex);
            receiveSignal.wait(receivedLock);
		}

		printf("start unsubscribe...\n");
		/*
		* Unsubscribe from the topic.
		*/
		std::promise<void> unsubscribeFinishedPromise;
		connection->Unsubscribe(topic.c_str(), [&](Mqtt::MqttConnection &, uint16_t, int) { unsubscribeFinishedPromise.set_value(); });
		unsubscribeFinishedPromise.get_future().wait();

		printf("start disconnecting...\n");
		/* Disconnect */
		if (connection->Disconnect()) {
			connectionClosedPromise.get_future().wait();
			printf("inside disconnect return\n");
		}

		ret = 0;
	} else {
		init_status = -1;
	}
	//exit mqtt work

 
	return ret;
}

The error message we are experiencing is the one printed at line 95 of this example code.

If there is anything I can help clear up about our issue or any other details I can provide please let me know.

Thank you,
Noah