MQTT C

/**
 * 消息处理函数
 */
int messageArrived(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
	size_t delimlen = 0;

	if (opts.verbose)
		printf("%d %s\t", message->payloadlen, topicName);
	if (opts.delimiter)
		delimlen = strlen(opts.delimiter);
	if (opts.delimiter == NULL || (message->payloadlen > delimlen &&
		strncmp(opts.delimiter, &((char*)message->payload)[message->payloadlen - delimlen], delimlen) == 0))
		printf("%.*s", message->payloadlen, (char*)message->payload);
	else
		printf("%.*s%s", message->payloadlen, (char*)message->payload, opts.delimiter);
	if (message->struct_version == 1 && opts.verbose)
		logProperties(&message->properties);
	fflush(stdout);
	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);
	return 1;
}

void reConnect() {
    printf("reConnect now\n");
    if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
    {
        if (!opts.quiet)
            fprintf(stderr, "reConnect---Failed to start connect, return code %s\n", MQTTAsync_strerror(rc));
    }
}

/**
 * 断开重连
 */
void connectionLost(void* context, char* cause) {
    printf("connectionLost: %s\n", cause);
    reConnect()
}

void deliveryComplete(void* context, MQTTAsync_token token) {
    printf("deliveryComplete\n");
}

/**
 * 订阅失败
 */
void onSubscribeFailure(void* context, MQTTAsync_failureData* response)
{
	if (!opts.quiet)
		fprintf(stderr, "Subscribe failed, rc %s\n",
			MQTTAsync_strerror(response->code));
    finished = 1;
}

/**
 * 连接失败后重试
 */
void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
	if (!opts.quiet)
		fprintf(stderr, "Connect failed, rc %s\n", response ? MQTTAsync_strerror(response->code) : "none");
    //finished = 1;
    reConnect();
}


/**
 * 连接成功后订阅主题
 */
void onConnect5(void* context, MQTTAsync_successData5* response)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_callOptions copts = MQTTAsync_callOptions_initializer;
	int rc;

        // QoS 0: 至多一次（At most once）
        // QoS 1: 至少一次（At least once）
        // QoS 2: 只有一次（Exactly once）
	if (opts.verbose)
		printf("Subscribing to topic %s with client %s at QoS %d\n", opts.topic, opts.clientid, opts.qos);

	copts.onSuccess5 = onSubscribe5;
	copts.onFailure5 = onSubscribeFailure5;
	copts.context = client;
	if ((rc = MQTTAsync_subscribe(client, opts.topic, opts.qos, &copts)) != MQTTASYNC_SUCCESS)
	{
		if (!opts.quiet)
			fprintf(stderr, "Failed to start subscribe, return code %s\n", MQTTAsync_strerror(rc));
		finished = 1;
	}
}


int main(int argc, char** argv)
{
	MQTTAsync client;
	MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
	MQTTAsync_createOptions create_opts = MQTTAsync_createOptions_initializer;
	MQTTAsync_willOptions will_opts = MQTTAsync_willOptions_initializer;
	MQTTAsync_SSLOptions ssl_opts = MQTTAsync_SSLOptions_initializer;
	int rc = 0;
	
	//-------------------------------------------参数解析----------------------------------------
	if (getopts(argc, argv, &opts) != 0) {
            usage(&opts, (pubsub_opts_nameValue*)infos, program_name);
        }
        //--------------------------------------创建MQTTAsync对象------------------------------------
	rc = MQTTAsync_createWithOptions(&client, url, opts.clientid, MQTTCLIENT_PERSISTENCE_NONE,
			NULL, &create_opts);

        //------------------------------------设置消息回调-------------------------------------------
	rc = MQTTAsync_setCallbacks(client, client, connectionLost, messageArrived, deliveryComplete);

        //--------------------------------------连接状态回调----------------------------------------
	if (opts.MQTTVersion == MQTTVERSION_5)
	{
		MQTTAsync_connectOptions conn_opts5 = MQTTAsync_connectOptions_initializer5;
		conn_opts = conn_opts5;
		conn_opts.onSuccess5 = onConnect5;
		conn_opts.onFailure5 = onConnectFailure5;
		conn_opts.cleanstart = 1;
	}
	else
	{
		conn_opts.onSuccess = onConnect;
		conn_opts.onFailure = onConnectFailure;
		conn_opts.cleansession = 1;
	}
	
	// keepAliveInterval（保持连接间隔）是一个重要的参数，它定义了客户端与服务器之间在无任何数据传输的情况下，为了维持TCP连接的活跃状态，客户端需要多长时间发送一次心跳（PINGREQ）消息
	conn_opts.keepAliveInterval = opts.keepalive;
	conn_opts.username = opts.username;
	conn_opts.password = opts.password;
	conn_opts.MQTTVersion = opts.MQTTVersion;
	conn_opts.context = client;
	
	// 用于控制客户端在与服务器断开连接后是否自动尝试重新建立连接。当设置为 true 时，如果客户端检测到与服务器的连接中断，它会按照一定的重连策略（如线性退避、指数退避等）自动尝试重新连接到服务器
	conn_opts.automaticReconnect = 1;
	conn_opts.httpProxy = opts.http_proxy;
	conn_opts.httpsProxy = opts.https_proxy;

        // 通过设置 will_topic 和 will_message，即使设备突然离线，其他 MQTT 客户端也能通过监听此 will_topic 获知该设备的离线状态或者接收到设备预先设定好的离线信息。
        // 这对于监控系统中的设备状态、确保数据完整性以及执行相应的故障恢复操作非常有用。
	if (opts.will_topic) 	/* will options */
	{
		will_opts.message = opts.will_payload;
		will_opts.topicName = opts.will_topic;
		will_opts.qos = opts.will_qos;
		will_opts.retained = opts.will_retain;
		conn_opts.will = &will_opts;
	}

        // 如果你关心数据安全性，应当优先考虑使用 wss:// 或 ssl://（或等同于 tls:// 的方案），它们能为MQTT连接提供加密保护。
        // 而 tcp:// 方案适用于不需要加密或者已通过其他手段保证网络环境安全的情况。
	if (opts.connection && (strncmp(opts.connection, "ssl://", 6) == 0 ||
			strncmp(opts.connection, "wss://", 6) == 0))
	{
		ssl_opts.verify = (opts.insecure) ? 0 : 1;
		ssl_opts.CApath = opts.capath;
		ssl_opts.keyStore = opts.cert;
		ssl_opts.trustStore = opts.cafile;
		ssl_opts.privateKey = opts.key;
		ssl_opts.privateKeyPassword = opts.keypass;
		ssl_opts.enabledCipherSuites = opts.ciphers;
		conn_opts.ssl = &ssl_opts;
	}

        // 连接MQTT
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		if (!opts.quiet)
			fprintf(stderr, "Failed to start connect, return code %s\n", MQTTAsync_strerror(rc));
		exit(EXIT_FAILURE);
	}

	while (!subscribed)
		mysleep(100);

	if (finished)
		goto exit;

	while (!finished)
		mysleep(100);

	disc_opts.onSuccess = onDisconnect;
	if ((rc = MQTTAsync_disconnect(client, &disc_opts)) != MQTTASYNC_SUCCESS)
	{
		if (!opts.quiet)
			fprintf(stderr, "Failed to start disconnect, return code: %s\n", MQTTAsync_strerror(rc));
		exit(EXIT_FAILURE);
	}

	while (!disconnected)
		mysleep(100);

exit:
	MQTTAsync_destroy(&client);

	return EXIT_SUCCESS;

}

/**
 * 客户端连接
 */
void myconnect(MQTTAsync client)
{
        //-------------------------------------------------------------------

	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		fprintf(stderr, "Failed to start connect, return code %s\n", MQTTAsync_strerror(rc));
		exit(EXIT_FAILURE);
	}
}

/**
 * 消息发布
 */
int mypublish(MQTTAsync client, int datalen, char* data)
{
	int rc;

    //if (opts.verbose)
		printf("Publishing data of length %d\n", datalen);

	rc = MQTTAsync_send(client, opts.topic, datalen, data, opts.qos, opts.retained, &pub_opts);
	if (opts.verbose && rc != MQTTASYNC_SUCCESS && !opts.quiet)
		fprintf(stderr, "Error from MQTTAsync_send: %s\n", MQTTAsync_strerror(rc));

	return rc;
}


/**
 * 连接回调
 */
void onConnect(void* context, MQTTAsync_successData* response)
{
	MQTTAsync client = (MQTTAsync)context;
	int rc = 0;

    //if (opts.verbose)
		printf("Connected\n");

	if (opts.null_message == 1)
		rc = mypublish(client, 0, "");
	else if (opts.message)
		rc = mypublish(client, (int)strlen(opts.message), opts.message);
	else if (opts.filename)
	{
		int data_len = 0;
		char* buffer = readfile(&data_len, &opts);

		if (buffer == NULL)
			toStop = 1;
		else
		{
			rc = mypublish(client, data_len, buffer);
			free(buffer);
		}
	}

	connected = 1;
}

void onPublish(void* context, MQTTAsync_successData* response)
{
    //if (opts.verbose)
		printf("Publish succeeded\n");

	if (opts.null_message || opts.message || opts.filename)
		toStop = 1;

	published = 1;
}

int main(int argc, char** argv)
{

	MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
	MQTTAsync_createOptions create_opts = MQTTAsync_createOptions_initializer;
	MQTTAsync client;
	char* buffer = NULL;
	char* url = NULL;
	int url_allocated = 0;
	int rc = 0;

	//-----------------------------------创建mqtt客户端-----------------------------------

        rc = MQTTAsync_createWithOptions(&client, url, opts.clientid, MQTTCLIENT_PERSISTENCE_NONE, NULL, &create_opts);

	//-----------------------------------设置MQTT 回调------------------------------------

	rc = MQTTAsync_setCallbacks(client, client, NULL, messageArrived, NULL);

	//-----------------------------------建立MQTT 连接------------------------------------

	myconnect(client);

	while (!toStop)
	{
	                //-----------------------------发布消息---------------------------------
			rc = mypublish(client, data_len, buffer);
	}

        //------------------------------断开 MQTT 连接，回收资源--------------------------------

	if ((rc = MQTTAsync_disconnect(client, &disc_opts)) != MQTTASYNC_SUCCESS)
	{
		if (!opts.quiet)
			fprintf(stderr, "Failed to start disconnect, return code: %s\n", MQTTAsync_strerror(rc));
		exit(EXIT_FAILURE);
	}

	while (!disconnected)
		mysleep(100);

	MQTTAsync_destroy(&client);

	if (url_allocated)
		free(url);

}