MQTT

5 mymqtt软件包是否支持多主题订阅及发布？

发布于 2022-05-07 15:36:31 浏览：1391 订阅该版

crystal266 2022-05-07

嵌入式

crystal266 2022-05-07

嵌入式

请登录后再发布评论，点击登录

追加回复

4 个回答

lchnu 2022-05-07

Witness, Understand, Skill

我没有用过umqtt，但是看了一下源码，`umqtt是支持多主题订阅的。`

umqtt_subscribe的过程可以简化为：

1.通过语句取得client的sub_rec_list，若链表不为空，则判断topic是否已存在；
```c
_cnt = rt_list_isempty(&client->sub_recv_list);

rt_list_for_each(node, &client->sub_recv_list)
     ....
```

2.发送sub数据，若收到suback，则将新节点插入到client的`sub_rec_list`链表中

```c
rt_list_insert_after(&client->sub_recv_list, &p_subtop->next_list);
```

完整的umqtt_subscrbie的代码如下:

```c
int umqtt_subscribe(struct umqtt_client *client, const char *topic, enum umqtt_qos qos, umqtt_subscribe_cb callback)
{
    int _ret = 0;
    int _length = 0;
    int _cnt = 0;
    struct subtop_recv_handler *p_subtop = RT_NULL;
    rt_list_t *node = RT_NULL;
    struct umqtt_msg encode_msg = { 0 };
    struct umqtt_msg_ack msg_ack = { 0 };

RT_ASSERT(client);
    RT_ASSERT(topic);

_cnt = rt_list_isempty(&client->sub_recv_list);

if (_cnt == 0)
    {
        _cnt = 0;
        rt_list_for_each(node, &client->sub_recv_list)
        {
            p_subtop = rt_list_entry(node, struct subtop_recv_handler, next_list);
            if (p_subtop->topicfilter
            && (rt_strcmp(p_subtop->topicfilter, topic) == 0))
            {
                LOG_D(" subscribe topic(%s) is already subscribed.", topic);
                goto exit;
            }
        }
        _length = rt_list_len(&client->sub_recv_list);
    }

if (_length > client->sub_recv_list_len)
    {
        _ret = UMQTT_MEM_FULL;
        LOG_E(" subscribe size(%d) is not enough! now length(%d)!", client->sub_recv_list_len, _length);
        goto exit;
    }
    else
    {
        rt_memset(&encode_msg, 0, sizeof(encode_msg));
        encode_msg.header.bits.qos = UMQTT_QOS1;
        encode_msg.msg.subscribe.packet_id = get_next_packetID(client);
        encode_msg.msg.subscribe.topic_filter[0].topic_filter = topic;
        encode_msg.msg.subscribe.topic_filter[0].filter_len = strlen(topic);
        encode_msg.msg.subscribe.topic_filter[0].req_qos.request_qos = qos;
        encode_msg.msg.subscribe.topic_count = 1;
        rt_memset(client->send_buf, 0, sizeof(rt_uint8_t) * client->mqtt_info.send_size);
        _length = umqtt_encode(UMQTT_TYPE_SUBSCRIBE, client->send_buf, client->mqtt_info.send_size, &encode_msg);
        if (_length <= 0)
        {
            _ret = UMQTT_ENCODE_ERROR;
            LOG_E(" subscribe encode failed! topic: %s", topic);
            goto exit;
        }
        client->send_len = _length;

_ret = umqtt_trans_send(client->sock, client->send_buf, client->send_len, client->mqtt_info.send_timeout);
        if (_ret < 0)
        {
            _ret = UMQTT_SEND_FAILED;
            LOG_E(" subscribe trans send failed!");
            goto exit;
        }

set_uplink_recon_tick(client, UPLINK_LAST_TICK);

rt_memset(&msg_ack, 0, sizeof(msg_ack));
        if (RT_EOK == rt_mq_recv(client->msg_queue,
                                &msg_ack, sizeof(struct umqtt_msg_ack),
                                rt_tick_from_millisecond(client->mqtt_info.send_timeout * 1000)))
        {
            if (msg_ack.msg_type == UMQTT_TYPE_SUBACK)
            {
                p_subtop = RT_NULL;
                p_subtop = (struct subtop_recv_handler *)rt_calloc(1, sizeof(struct subtop_recv_handler));
                RT_ASSERT(p_subtop);
                LOG_D(" start assign datas !");
                p_subtop->topicfilter = rt_strdup(topic);
                p_subtop->qos = qos;
                if (callback)
                {
                    p_subtop->callback = callback;
                }
                rt_list_insert_after(&client->sub_recv_list, &p_subtop->next_list);
                set_uplink_recon_tick(client, UPLINK_NEXT_TICK);

_ret = UMQTT_OK;
                LOG_I("subscribe ack ok! ");
                goto exit;
            }
            else
            {
                _ret = UMQTT_READ_ERROR;
                LOG_E("subscribe ack error!");
                goto exit;
            }
        }
        else
        {
            _ret = UMQTT_READ_FAILED;
            LOG_E(" subscribe recv message timeout! topic: %s", topic);
            goto exit;
        }
    }

exit:
    return _ret;
}
```

此外，内部的`umqtt_deliver_message`函数中，也是遍历链表，然后执行回调的。

因此，umqtt支持同一client的多主题订阅。

补充说明，查看umqtt实现的内部函数`topicname_is_matched`，其中涉及到对`+`, `#`的处理，说明它也是支持通配符的。

仓库：https://github.com/RT-Thread-packages/umqtt

umqtt和rt-thread绑定太深了。我个人习惯用kawaii-mqtt。

再补充一点，lz的标题中提到了`发布`，对于MQTT来说，publish 爱咋咋，发送主题数量随意，用户自行调用`umqtt_publish`函数向不同topic发送就好。QOS1和QOS0情况下虽然要交互几次，但是与Topic无关。

祝顺利！

崔晓光打赏了 1.00 元 2022-05-07 16:09:19

崔晓光 2022-05-07

这家伙很懒，什么也没写！

lchnu 2022-05-07

Witness, Understand, Skill

@崔晓光

mymqtt应该是支持的呀。

证据：内部`deliver_message`函数通过循环来判断topic，并且执行不同的callback。你的工程中不支持多topic，是不是`MAX_MESSAGE_HANDLERS`被配置成了1？

```c
static int deliver_message(mqtt_client *c, MQTTString *topic_name, mqtt_message *message)
{
  int i;
  int rc = PAHO_FAILURE;
  
  // we have to find the right message handler - indexed by topic
  for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
  {
    if (c->message_handlers[i].topicFilter != 0 && (MQTTPacket_equals(topic_name, (char *)c->message_handlers[i].topicFilter) ||
                                                    is_topic_matched((char *)c->message_handlers[i].topicFilter, topic_name)))
    {
      if (c->message_handlers[i].callback != NULL)
      {
        message_data md;
        new_message_data(&md, topic_name, message);
        c->message_handlers[i].callback(c, &md);
        rc = PAHO_SUCCESS;
      }
    }
  }
  
  if (rc == PAHO_FAILURE && c->default_message_handlers != NULL)
  {
    message_data md;
    new_message_data(&md, topic_name, message);
    c->default_message_handlers(c, &md);
    rc = PAHO_SUCCESS;
  }
  
  return rc;
}
```