int MQTTUnsubscribe(Client *c,const char* topicFilter, unsigned int msgId)
{
int len = 0;
int rc = 0;
Timer timer;
MessageHandlers handler = {topicFilter,NULL};
list_node_t *node = NULL;
MQTTString topic = MQTTString_initializer;
topic.cstring = (char *)topicFilter;
InitTimer(&timer);
countdown_ms(&timer, c->command_timeout_ms);
aliyun_iot_mutex_lock(&c->writebufLock);
if ((len = MQTTSerialize_unsubscribe(c->buf, c->buf_size, 0, (unsigned short)msgId, 1, &topic)) <= 0)
{
aliyun_iot_mutex_unlock(&c->writebufLock);
return MQTT_UNSUBSCRIBE_PACKET_ERROR;
}
//将sub信息保存到subInfoList中
if (SUCCESS_RETURN != push_subInfo_to(c,len,msgId,UNSUBSCRIBE,&handler,&node))
{
WRITE_IOT_ERROR_LOG("push publish into to pubInfolist failed!");
aliyun_iot_mutex_unlock(&c->writebufLock);
return MQTT_PUSH_TO_LIST_ERROR;
}
if ((rc = sendPacket(c, c->buf,len, &timer)) != SUCCESS_RETURN) // send the subscribe packet
{
//发送失败则删除之前放入subInfoList链表中的节点
aliyun_iot_mutex_lock(&c->subInfoLock);
list_remove(c->subInfoList, node);
aliyun_iot_mutex_unlock(&c->subInfoLock);
aliyun_iot_mutex_unlock(&c->writebufLock);
return MQTT_NETWORK_ERROR;
}
aliyun_iot_mutex_unlock(&c->writebufLock);
return SUCCESS_RETURN;
}
|