# Mqtt 插件


## 相关链接

参考示例：
- {{ '[mqtt_plugin]({}/src/examples/plugins/mqtt_plugin)'.format(code_site_root_path_url) }}


## 插件概述

**mqtt_plugin**是一个基于 mqtt 协议实现的网络传输插件，此插件提供了以下组件：
- `mqtt`类型 RPC 后端
- `mqtt`类型 Channel 后端


插件的配置项如下：

| 节点           | 类型   | 是否可选 | 默认值 | 作用                    |
| -------------- | ------ | -------- | ------ | ----------------------- |
| broker_addr    | string | 必选     | ""     | mqtt broker 的地址      |
| client_id      | string | 必选     | ""     | 本节点的 mqtt client id |
| max_pkg_size_k | int    | 可选     | 1024   | 最大包尺寸，单位：KB    |
| truststore     | string | 可选     | ""     | broker的CA证书路径      |


关于**mqtt_plugin**的配置，使用注意点如下：
- `broker_addr`表示 mqtt broker 的地址，使用者必须保证有 mqtt 的 broker 运行在该地址，否则启动会失败。
- `client_id`表示本节点连接 mqtt broker 时的 client id。
- `max_pkg_size_k`表示传输数据时的最大包尺寸，默认 1 MB。注意，必须 broker 也要支持该尺寸才行。
- `truststore`表示 broker 的 CA 证书路径，例如`/etc/emqx/certs/cacert.pem` 。当`broker_addr`的协议被配置为`ssl`或者`mqtts`时，该选项生效，用于指定 CA 证书路径，否则自动忽略该选项。

**mqtt_plugin**插件基于[paho.mqtt.c](https://github.com/eclipse/paho.mqtt.c)封装，在使用时，Channel 订阅回调、RPC Server 处理方法、RPC Client 返回时，使用的都是**paho.mqtt.c**提供的线程，当使用者在回调中阻塞了线程时，有可能导致无法继续接收/发送消息。正如 Module 接口文档中所述，一般来说，如果回调中的任务非常轻量，那就可以直接在回调里处理；但如果回调中的任务比较重，那最好调度到其他专门执行任务的执行器里处理。


以下是一个简单的示例：
```yaml
aimrt:
  plugin:
    plugins:
      - name: mqtt_plugin
        path: ./libaimrt_mqtt_plugin.so
        options:
          broker_addr: tcp://127.0.0.1:1883
          client_id: example_mqtt_client
          max_pkg_size_k: 1024
```


## mqtt 类型 RPC 后端


`mqtt`类型的 RPC 后端是**mqtt_plugin**中提供的一种 RPC 后端，用于通过 mqtt 的方式来调用和处理 AimRT RPC 请求。其所有的配置项如下：


| 节点                              | 类型   | 是否可选 | 默认值 | 作用                                                                         |
| --------------------------------- | ------ | -------- | ------ | ---------------------------------------------------------------------------- |
| timeout_executor                  | string | 可选     | ""     | Client 端发起 RPC 超时情况下的执行器                                         |
| clients_options                   | array  | 可选     | []     | Client 端发起 RPC 请求时的规则                                               |
| clients_options[i].func_name      | string | 必选     | ""     | RPC Func 名称，支持正则表达式                                                |
| clients_options[i].server_mqtt_id | string | 可选     | ""     | RPC Func 发起调用时请求的 mqtt 服务端 id                                     |
| clients_options[i].qos            | int    | 可选     | 2      | RPC Client 端 mqtt qos，取值范围：0/1/2                                      |
| servers_options                   | array  | 可选     | []     | 服务端处理 RPC 请求时的规则                                                  |
| servers_options[i].func_name      | string | 必选     | ""     | RPC Func 名称，支持正则表达式                                                |
| servers_options[i].allow_share    | bool   | 可选     | true   | 该 RPC 服务是否允许共享订阅，不允许的话该服务只能通过指定 server id 进行调用 |
| servers_options[i].qos            | int    | 可选     | 2      | RPC Server 端 mqtt qos，取值范围：0/1/2                                      |

以下是一个简单的客户端的示例：
```yaml
aimrt:
  plugin:
    plugins:
      - name: mqtt_plugin
        path: ./libaimrt_mqtt_plugin.so
        options:
          broker_addr: tcp://127.0.0.1:1883
          client_id: example_client
          max_pkg_size_k: 1024
  executor:
    executors:
      - name: timeout_handle
        type: time_wheel
  rpc:
    backends:
      - type: mqtt
        options:
          timeout_executor: timeout_handle
          clients_options:
            - func_name: "(.*)"
              qos: 0
    clients_options:
      - func_name: "(.*)"
        enable_backends: [mqtt]
```

以下则是一个简单的服务端的示例：
```yaml
aimrt:
  plugin:
    plugins:
      - name: mqtt_plugin
        path: ./libaimrt_mqtt_plugin.so
        options:
          broker_addr: tcp://127.0.0.1:1883
          client_id: example_server
          max_pkg_size_k: 1024
  rpc:
    backends:
      - type: mqtt
        options:
          servers_options:
            - func_name: "(.*)"
              allow_share: true
              qos: 0
    servers_options:
      - func_name: "(.*)"
        enable_backends: [mqtt]
```

以上示例中，Client 端和 Server 端都连上了`tcp://127.0.0.1:1883`这个地址的一个 Mqtt broker，Client 端也配置了所有的 RPC 请求都通过 mqtt 后端进行处理，从而完成 RPC 的调用闭环。

如果有多个 server 端同时注册了某个 RPC 服务，那么 client 端会随机的挑选一个 server 端发送请求。如果想指定某个 server 端处理，可以在 client 端的 ctx 中按照如下方法设置 ToAddr：
```cpp
auto ctx_ptr = proxy->NewContextSharedPtr();
// mqtt://{{target server mqtt id}}
ctx_ptr->SetToAddr("mqtt://target_server_mqtt_id");

auto status = proxy->Foo(ctx_ptr, req, rsp);
```


在整个 RPC 过程中，底层使用的 Mqtt Topic 名称格式如下：
- Server 端
  - 订阅 Req 使用的 topic（两个都会订阅）：
    - `$share/aimrt/aimrt_rpc_req/${func_name}`
    - `aimrt_rpc_req/${server_id}/${func_name}`
  - 发布 Rsp 使用的 topic：`aimrt_rpc_rsp/${client_id}/${func_name}`
- Client 端
  - 发布 Req 使用的 topic（二选一）：
    - `aimrt_rpc_req/${func_name}`
    - `aimrt_rpc_req/${server_id}/${func_name}`
  - 订阅 Rsp 使用的 topic：`aimrt_rpc_rsp/${client_id}/${func_name}`

其中`${client_id}`、`${server_id}`是 Client 端和 Server 端需要保证在同一个 Mqtt broker 环境下全局唯一的一个值，一般使用在 Mqtt broker 处注册的 id。`${func_name}`是 url 编码后的 AimRT RPC 方法名称。Server 端订阅使用共享订阅，保证只有一个服务端处理请求。此项特性需要支持 Mqtt5.0 协议的 Broker。


例如，client 端向 Mqtt broker 注册的 id 为`example_client`，func 名称为`/aimrt.protocols.example.ExampleService/GetBarData`，则`${client_id}`值为`example_client`，`${func_name}`值为`%2Faimrt.protocols.example.ExampleService%2FGetBarData`。



Client -> Server 的 Mqtt 数据包格式整体分 5 段:
- 序列化类型，一般是`pb`或`json`
- client 端想要 server 端回复 rsp 的 mqtt topic 名称。client 端自己需要订阅这个 mqtt topic
- msg id，4 字节，server 端会原封不动的封装到 rsp 包里，供 client 端定位 rsp 对应哪个 req
- context 区
  - context 数量，1 字节，最大 255 个 context
  - context_1 key, 2 字节长度 + 数据区
  - context_2 key, 2 字节长度 + 数据区
  - ...
- msg 数据

```
| n(0~255) [1 byte] | content type [n byte]
| m(0~255) [1 byte] | rsp topic name [m byte]
| msg id [4 byte]
| context num [1 byte]
| context_1 key size [2 byte] | context_1 key data [key_1_size byte]
| context_1 val size [2 byte] | context_1 val data [val_1_size byte]
| context_2 key size [2 byte] | context_2 key data [key_2_size byte]
| context_2 val size [2 byte] | context_2 val data [val_2_size byte]
| ...
| msg data [remaining byte]
```

Server -> Client 的 Mqtt 数据包格式整体分 4 段:
- 序列化类型，一般是`pb`或`json`
- msg id，4 字节，req 中的 msg id
- status code，4 字节，框架错误码，如果这个部分不为零，则代表服务端发生了错误，数据段将没有内容
- msg 数据

```
| n(0~255) [1 byte] | content type [n byte]
| msg id [4 byte]
| status code [4 byte]
| msg data [remaining byte]
```


## mqtt 类型 Channel 后端


`mqtt`类型的 Channel 后端是**mqtt_plugin**中提供的一种 Channel 后端，用于通过 mqtt 的方式来发布和订阅消息。其所有的配置项如下：


| 节点                             | 类型   | 是否可选 | 默认值 | 作用                                   |
| -------------------------------- | ------ | -------- | ------ | -------------------------------------- |
| pub_topics_options               | array  | 可选     | []     | 发布 Topic 时的规则                    |
| pub_topics_options[i].topic_name | string | 必选     | ""     | Topic 名称，支持正则表达式             |
| pub_topics_options[i].qos        | int    | 必选     | 2      | Publish 端 mqtt qos，取值范围：0/1/2   |
| sub_topics_options               | array  | 可选     | []     | 发布 Topic 时的规则                    |
| sub_topics_options[i].topic_name | string | 必选     | ""     | Topic 名称，支持正则表达式             |
| sub_topics_options[i].qos        | int    | 必选     | 2      | Subscribe 端 mqtt qos，取值范围：0/1/2 |


以下是一个简单的发布端的示例：
```yaml
aimrt:
  plugin:
    plugins:
      - name: mqtt_plugin
        path: ./libaimrt_mqtt_plugin.so
        options:
          broker_addr: tcp://127.0.0.1:1883
          client_id: example_publisher
          max_pkg_size_k: 1024
  channel:
    backends:
      - type: mqtt
        options:
          pub_topics_options:
            - topic_name: "(.*)"
              qos: 2
    pub_topics_options:
      - topic_name: "(.*)"
        enable_backends: [mqtt]
```

以下则是一个简单的订阅端的示例：
```yaml
aimrt:
  plugin:
    plugins:
      - name: mqtt_plugin
        path: ./libaimrt_mqtt_plugin.so
        options:
          broker_addr: tcp://127.0.0.1:1883
          client_id: example_subscriber
          max_pkg_size_k: 1024
  channel:
    backends:
      - type: mqtt
    sub_topics_options:
      - topic_name: "(.*)"
        enable_backends: [mqtt]
```

以上示例中，发布端和订阅端都连上了`tcp://127.0.0.1:1883`这个地址的一个 Mqtt broker，发布端也配置了所有的消息都通过 mqtt 后端进行处理，订阅端也配置了所有消息都可以从 mqtt 后端触发回调，从而打通消息发布订阅的链路。


在这个过程中，底层使用的 Mqtt Topic 名称格式为：`/channel/${topic_name}/${message_type}`。其中，`${topic_name}`为 AimRT 的 Topic 名称，`${message_type}`为 url 编码后的 AimRT 消息名称。

例如，AimRT Topic 名称为`test_topic`，消息类型为`pb:aimrt.protocols.example.ExampleEventMsg`，则最终 Mqtt 的 topic 名称为：`/channel/test_topic/pb%3Aaimrt.protocols.example.ExampleEventMsg`。


在 AimRT 发布端发布数据到订阅端这个链路上，Mqtt 数据包格式整体分 3 段：
- 序列化类型，一般是`pb`或`json`
- context 区
  - context 数量，1 字节，最大 255 个 context
  - context_1 key, 2 字节长度 + 数据区
  - context_2 key, 2 字节长度 + 数据区
  - ...
- 数据

```
| n(0~255) [1 byte] | content type [n byte]
| context num [1 byte]
| context_1 key size [2 byte] | context_1 key data [key_1_size byte]
| context_1 val size [2 byte] | context_1 val data [val_1_size byte]
| context_2 key size [2 byte] | context_2 key data [key_2_size byte]
| context_2 val size [2 byte] | context_2 val data [val_2_size byte]
| ...
| msg data [len - 1 - n byte] |
```