refactor(bridges): remote the old emqx_bridge_mqtt app
This commit is contained in:
Shawn
parent
1b5c08adeb
commit
037b75a276
|
|
@ -90,7 +90,7 @@ make eunit ct
|
||||||
### 执行部分应用的 common tests
|
### 执行部分应用的 common tests
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
make apps/emqx_bridge_mqtt-ct
|
make apps/emqx_bridge-ct
|
||||||
```
|
```
|
||||||
|
|
||||||
### 静态分析(Dialyzer)
|
### 静态分析(Dialyzer)
|
||||||
|
|
|
||||||
|
|
@ -84,7 +84,7 @@ make eunit ct
|
||||||
### common test の一部を実行する
|
### common test の一部を実行する
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
make apps/emqx_bridge_mqtt-ct
|
make apps/emqx_bridge-ct
|
||||||
```
|
```
|
||||||
|
|
||||||
### Dialyzer
|
### Dialyzer
|
||||||
|
|
|
||||||
|
|
@ -93,7 +93,7 @@ make eunit ct
|
||||||
Пример:
|
Пример:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
make apps/emqx_bridge_mqtt-ct
|
make apps/emqx_bridge-ct
|
||||||
```
|
```
|
||||||
|
|
||||||
### Dialyzer
|
### Dialyzer
|
||||||
|
|
|
||||||
|
|
@ -92,7 +92,7 @@ make eunit ct
|
||||||
Examples
|
Examples
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
make apps/emqx_bridge_mqtt-ct
|
make apps/emqx_bridge-ct
|
||||||
```
|
```
|
||||||
|
|
||||||
### Dialyzer
|
### Dialyzer
|
||||||
|
|
|
||||||
|
|
@ -1,21 +0,0 @@
|
||||||
.eunit
|
|
||||||
deps
|
|
||||||
*.o
|
|
||||||
*.beam
|
|
||||||
*.plt
|
|
||||||
erl_crash.dump
|
|
||||||
ebin/*.beam
|
|
||||||
rel
|
|
||||||
_build
|
|
||||||
.concrete/DEV_MODE
|
|
||||||
.rebar
|
|
||||||
.erlang.mk
|
|
||||||
data
|
|
||||||
ebin
|
|
||||||
emqx_bridge_mqtt.d
|
|
||||||
*.rendered
|
|
||||||
.rebar3/
|
|
||||||
*.coverdata
|
|
||||||
rebar.lock
|
|
||||||
.DS_Store
|
|
||||||
Mnesia.*/
|
|
||||||
|
|
@ -1,265 +0,0 @@
|
||||||
# EMQ Bridge MQTT
|
|
||||||
|
|
||||||
The concept of **Bridge** means that EMQ X supports forwarding messages
|
|
||||||
of one of its own topics to another MQTT Broker in some way.
|
|
||||||
|
|
||||||
**Bridge** differs from **Cluster** in that the bridge does not
|
|
||||||
replicate the topic trie and routing tables and only forwards MQTT
|
|
||||||
messages based on bridging rules.
|
|
||||||
|
|
||||||
At present, the bridging methods supported by EMQ X are as follows:
|
|
||||||
|
|
||||||
- RPC bridge: RPC Bridge only supports message forwarding and does not
|
|
||||||
support subscribing to the topic of remote nodes to synchronize
|
|
||||||
data;
|
|
||||||
- MQTT Bridge: MQTT Bridge supports both forwarding and data
|
|
||||||
synchronization through subscription topic.
|
|
||||||
|
|
||||||
These concepts are shown below:
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
In addition, the EMQ X message broker supports multi-node bridge mode interconnection
|
|
||||||
|
|
||||||
```
|
|
||||||
--------- --------- ---------
|
|
||||||
Publisher --> | Node1 | --Bridge Forward--> | Node2 | --Bridge Forward--> | Node3 | --> Subscriber
|
|
||||||
--------- --------- ---------
|
|
||||||
```
|
|
||||||
|
|
||||||
In EMQ X, bridge is configured by modifying `etc/emqx.conf`. EMQ X distinguishes between different bridges based on different names. E.g
|
|
||||||
|
|
||||||
```
|
|
||||||
## Bridge address: node name for local bridge, host:port for remote.
|
|
||||||
bridge.mqtt.aws.address = 127.0.0.1:1883
|
|
||||||
```
|
|
||||||
|
|
||||||
This configuration declares a bridge named `aws` and specifies that it is bridged to the MQTT broker of 127.0.0.1:1883 by MQTT mode.
|
|
||||||
|
|
||||||
In case of creating multiple bridges, it is convenient to replicate all configuration items of the first bridge, and modify the bridge name and other configuration items if necessary (such as bridge.$name.address, where $name refers to the name of bridge)
|
|
||||||
|
|
||||||
The next two sections describe how to create a bridge in RPC and MQTT mode respectively and create a forwarding rule that forwards the messages from sensors. Assuming that two EMQ X nodes are running on two hosts:
|
|
||||||
|
|
||||||
|
|
||||||
| Name | Node | MQTT Port |
|
|
||||||
|------|-------------------|-----------|
|
|
||||||
| emqx1| emqx1@192.168.1.1.| 1883 |
|
|
||||||
| emqx2| emqx2@192.168.1.2 | 1883 |
|
|
||||||
|
|
||||||
|
|
||||||
## EMQ X RPC Bridge Configuration
|
|
||||||
|
|
||||||
The following is the basic configuration of RPC bridging. A simplest RPC bridging only requires the following three items
|
|
||||||
|
|
||||||
```
|
|
||||||
## Bridge Address: Use node name (nodename@host) for rpc bridging, and host:port for mqtt connection
|
|
||||||
bridge.mqtt.emqx2.address = "emqx2@192.168.1.2"
|
|
||||||
|
|
||||||
## Forwarding topics of the message
|
|
||||||
bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
|
|
||||||
|
|
||||||
## bridged mountpoint
|
|
||||||
bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
|
|
||||||
```
|
|
||||||
|
|
||||||
If the messages received by the local node emqx1 matches the topic `sersor1/#` or `sensor2/#`, these messages will be forwarded to the `sensor1/#` or `sensor2/#` topic of the remote node emqx2.
|
|
||||||
|
|
||||||
`forwards` is used to specify topics. Messages of the in `forwards` specified topics on local node are forwarded to the remote node.
|
|
||||||
|
|
||||||
`mountpoint` is used to add a topic prefix when forwarding a message. To use `mountpoint`, the `forwards` directive must be set. In the above example, a message with the topic `sensor1/hello` received by the local node will be forwarded to the remote node with the topic `bridge/emqx2/emqx1@192.168.1.1/sensor1/hello`.
|
|
||||||
|
|
||||||
Limitations of RPC bridging:
|
|
||||||
|
|
||||||
1. The RPC bridge of emqx can only forward local messages to the remote node, and cannot synchronize the messages of the remote node to the local node;
|
|
||||||
|
|
||||||
2. RPC bridge can only bridge two EMQ X broker together and cannot bridge EMQ X broker to other MQTT brokers.
|
|
||||||
|
|
||||||
## EMQ X MQTT Bridge Configuration
|
|
||||||
|
|
||||||
EMQ X 3.0 officially introduced MQTT bridge, so that EMQ X can bridge any MQTT broker. Because of the characteristics of the MQTT protocol, EMQ X can subscribe to the remote mqtt broker's topic through MQTT bridge, and then synchronize the remote MQTT broker's message to the local.
|
|
||||||
|
|
||||||
EMQ X MQTT bridging principle: Create an MQTT client on the EMQ X broker, and connect this MQTT client to the remote MQTT broker. Therefore, in the MQTT bridge configuration, following fields may be set for the EMQ X to connect to the remote broker as an mqtt client
|
|
||||||
|
|
||||||
```
|
|
||||||
## Bridge Address: Use node name for rpc bridging, use host:port for mqtt connection
|
|
||||||
bridge.mqtt.emqx2.address = "192.168.1.2:1883"
|
|
||||||
|
|
||||||
## Bridged Protocol Version
|
|
||||||
## Enumeration value: mqttv3 | mqttv4 | mqttv5
|
|
||||||
bridge.mqtt.emqx2.proto_ver = "mqttv4"
|
|
||||||
|
|
||||||
## mqtt client's clientid
|
|
||||||
bridge.mqtt.emqx2.clientid = "bridge_emq"
|
|
||||||
|
|
||||||
## mqtt client's clean_start field
|
|
||||||
## Note: Some MQTT Brokers need to set the clean_start value as `true`
|
|
||||||
bridge.mqtt.emqx2.clean_start = true
|
|
||||||
|
|
||||||
## mqtt client's username field
|
|
||||||
bridge.mqtt.emqx2.username = "user"
|
|
||||||
|
|
||||||
## mqtt client's password field
|
|
||||||
bridge.mqtt.emqx2.password = "passwd"
|
|
||||||
|
|
||||||
## Whether the mqtt client uses ssl to connect to a remote serve or not
|
|
||||||
bridge.mqtt.emqx2.ssl = off
|
|
||||||
|
|
||||||
## CA Certificate of Client SSL Connection (PEM format)
|
|
||||||
bridge.mqtt.emqx2.cacertfile = "etc/certs/cacert.pem"
|
|
||||||
|
|
||||||
## SSL certificate of Client SSL connection
|
|
||||||
bridge.mqtt.emqx2.certfile = "etc/certs/client-cert.pem"
|
|
||||||
|
|
||||||
## Key file of Client SSL connection
|
|
||||||
bridge.mqtt.emqx2.keyfile = "etc/certs/client-key.pem"
|
|
||||||
|
|
||||||
## SSL encryption
|
|
||||||
bridge.mqtt.emqx2.ciphers = "ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384"
|
|
||||||
|
|
||||||
## TTLS PSK password
|
|
||||||
## Note 'listener.ssl.external.ciphers' and 'listener.ssl.external.psk_ciphers' cannot be configured at the same time
|
|
||||||
##
|
|
||||||
## See 'https://tools.ietf.org/html/rfc4279#section-2'.
|
|
||||||
## bridge.mqtt.emqx2.psk_ciphers = "PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA"
|
|
||||||
|
|
||||||
## Client's heartbeat interval
|
|
||||||
bridge.mqtt.emqx2.keepalive = 60s
|
|
||||||
|
|
||||||
## Supported TLS version
|
|
||||||
bridge.mqtt.emqx2.tls_versions = "tlsv1.3,tlsv1.2,tlsv1.1,tlsv1"
|
|
||||||
|
|
||||||
## Forwarding topics of the message
|
|
||||||
bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
|
|
||||||
|
|
||||||
## Bridged mountpoint
|
|
||||||
bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
|
|
||||||
|
|
||||||
## Subscription topic for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.1.topic = "cmd/topic1"
|
|
||||||
|
|
||||||
## Subscription qos for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.1.qos = 1
|
|
||||||
|
|
||||||
## Subscription topic for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.2.topic = "cmd/topic2"
|
|
||||||
|
|
||||||
## Subscription qos for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.2.qos = 1
|
|
||||||
|
|
||||||
## Bridging reconnection interval
|
|
||||||
## Default: 30s
|
|
||||||
bridge.mqtt.emqx2.reconnect_interval = 30s
|
|
||||||
|
|
||||||
## QoS1 message retransmission interval
|
|
||||||
bridge.mqtt.emqx2.retry_interval = 20s
|
|
||||||
|
|
||||||
## Inflight Size.
|
|
||||||
bridge.mqtt.emqx2.max_inflight_batches = 32
|
|
||||||
```
|
|
||||||
|
|
||||||
## Bridge Cache Configuration
|
|
||||||
|
|
||||||
The bridge of EMQ X has a message caching mechanism. The caching mechanism is applicable to both RPC bridging and MQTT bridging. When the bridge is disconnected (such as when the network connection is unstable), the messages with a topic specified in `forwards` can be cached to the local message queue. Until the bridge is restored, these messages are re-forwarded to the remote node. The configuration of the cache queue is as follows
|
|
||||||
|
|
||||||
```
|
|
||||||
## emqx_bridge internal number of messages used for batch
|
|
||||||
bridge.mqtt.emqx2.queue.batch_count_limit = 32
|
|
||||||
|
|
||||||
## emqx_bridge internal number of message bytes used for batch
|
|
||||||
bridge.mqtt.emqx2.queue.batch_bytes_limit = 1000MB
|
|
||||||
|
|
||||||
## The path for placing replayq queue. If it is not specified, then replayq will run in `mem-only` mode and messages will not be cached on disk.
|
|
||||||
bridge.mqtt.emqx2.queue.replayq_dir = data/emqx_emqx2_bridge/
|
|
||||||
|
|
||||||
## Replayq data segment size
|
|
||||||
bridge.mqtt.emqx2.queue.replayq_seg_bytes = 10MB
|
|
||||||
```
|
|
||||||
|
|
||||||
`bridge.mqtt.emqx2.queue.replayq_dir` is a configuration parameter for specifying the path of the bridge storage queue.
|
|
||||||
|
|
||||||
`bridge.mqtt.emqx2.queue.replayq_seg_bytes` is used to specify the size of the largest single file of the message queue that is cached on disk. If the message queue size exceeds the specified value, a new file is created to store the message queue.
|
|
||||||
|
|
||||||
## CLI for EMQ X Bridge MQTT
|
|
||||||
|
|
||||||
CLI for EMQ X Bridge MQTT:
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ cd emqx1/ && ./bin/emqx_ctl bridges
|
|
||||||
bridges list # List bridges
|
|
||||||
bridges start <Name> # Start a bridge
|
|
||||||
bridges stop <Name> # Stop a bridge
|
|
||||||
bridges forwards <Name> # Show a bridge forward topic
|
|
||||||
bridges add-forward <Name> <Topic> # Add bridge forward topic
|
|
||||||
bridges del-forward <Name> <Topic> # Delete bridge forward topic
|
|
||||||
bridges subscriptions <Name> # Show a bridge subscriptions topic
|
|
||||||
bridges add-subscription <Name> <Topic> <Qos> # Add bridge subscriptions topic
|
|
||||||
```
|
|
||||||
|
|
||||||
List all bridge states
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges list
|
|
||||||
name: emqx status: Stopped $ ./bin/emqx_ctl bridges list
|
|
||||||
name: emqx status: Stopped
|
|
||||||
```
|
|
||||||
|
|
||||||
Start the specified bridge
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges start emqx
|
|
||||||
Start bridge successfully.
|
|
||||||
```
|
|
||||||
|
|
||||||
Stop the specified bridge
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges stop emqx
|
|
||||||
Stop bridge successfully.
|
|
||||||
```
|
|
||||||
List the forwarding topics for the specified bridge
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges forwards emqx
|
|
||||||
topic: topic1/#
|
|
||||||
topic: topic2/#
|
|
||||||
```
|
|
||||||
|
|
||||||
Add a forwarding topic for the specified bridge
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges add-forwards emqx topic3/#
|
|
||||||
Add-forward topic successfully.
|
|
||||||
```
|
|
||||||
|
|
||||||
Delete the forwarding topic for the specified bridge
|
|
||||||
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges del-forwards emqx topic3/#
|
|
||||||
Del-forward topic successfully.
|
|
||||||
```
|
|
||||||
|
|
||||||
List subscriptions for the specified bridge
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges subscriptions emqx
|
|
||||||
topic: cmd/topic1, qos: 1
|
|
||||||
topic: cmd/topic2, qos: 1
|
|
||||||
```
|
|
||||||
|
|
||||||
Add a subscription topic for the specified bridge
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges add-subscription emqx cmd/topic3 1
|
|
||||||
Add-subscription topic successfully.
|
|
||||||
```
|
|
||||||
|
|
||||||
Delete the subscription topic for the specified bridge
|
|
||||||
|
|
||||||
``` bash
|
|
||||||
$ ./bin/emqx_ctl bridges del-subscription emqx cmd/topic3
|
|
||||||
Del-subscription topic successfully.
|
|
||||||
```
|
|
||||||
|
|
||||||
Note: In case of creating multiple bridges, it is convenient to replicate all configuration items of the first bridge, and modify the bridge name and other configuration items if necessary.
|
|
||||||
|
|
||||||
|
|
@ -1,286 +0,0 @@
|
||||||
|
|
||||||
EMQ Bridge MQTT
|
|
||||||
===============
|
|
||||||
|
|
||||||
The concept of **Bridge** means that EMQ X supports forwarding messages
|
|
||||||
of one of its own topics to another MQTT Broker in some way.
|
|
||||||
|
|
||||||
**Bridge** differs from **Cluster** in that the bridge does not
|
|
||||||
replicate the topic trie and routing tables and only forwards MQTT
|
|
||||||
messages based on bridging rules.
|
|
||||||
|
|
||||||
At present, the bridging methods supported by EMQ X are as follows:
|
|
||||||
|
|
||||||
|
|
||||||
* RPC bridge: RPC Bridge only supports message forwarding and does not
|
|
||||||
support subscribing to the topic of remote nodes to synchronize
|
|
||||||
data;
|
|
||||||
* MQTT Bridge: MQTT Bridge supports both forwarding and data
|
|
||||||
synchronization through subscription topic.
|
|
||||||
|
|
||||||
These concepts are shown below:
|
|
||||||
|
|
||||||
|
|
||||||
.. image:: images/bridge.png
|
|
||||||
:target: images/bridge.png
|
|
||||||
:alt: bridge
|
|
||||||
|
|
||||||
|
|
||||||
In addition, the EMQ X message broker supports multi-node bridge mode interconnection
|
|
||||||
|
|
||||||
.. code-block::
|
|
||||||
|
|
||||||
--------- --------- ---------
|
|
||||||
Publisher --> | Node1 | --Bridge Forward--> | Node2 | --Bridge Forward--> | Node3 | --> Subscriber
|
|
||||||
--------- --------- ---------
|
|
||||||
|
|
||||||
In EMQ X, bridge is configured by modifying ``etc/emqx.conf``. EMQ X distinguishes between different bridges based on different names. E.g
|
|
||||||
|
|
||||||
.. code-block::
|
|
||||||
|
|
||||||
## Bridge address: node name for local bridge, host:port for remote.
|
|
||||||
bridge.mqtt.aws.address = "127.0.0.1:1883"
|
|
||||||
|
|
||||||
This configuration declares a bridge named ``aws`` and specifies that it is bridged to the MQTT broker of 127.0.0.1:1883 by MQTT mode.
|
|
||||||
|
|
||||||
In case of creating multiple bridges, it is convenient to replicate all configuration items of the first bridge, and modify the bridge name and other configuration items if necessary (such as bridge.$name.address, where $name refers to the name of bridge)
|
|
||||||
|
|
||||||
The next two sections describe how to create a bridge in RPC and MQTT mode respectively and create a forwarding rule that forwards the messages from sensors. Assuming that two EMQ X nodes are running on two hosts:
|
|
||||||
|
|
||||||
.. list-table::
|
|
||||||
:header-rows: 1
|
|
||||||
|
|
||||||
* - Name
|
|
||||||
- Node
|
|
||||||
- MQTT Port
|
|
||||||
* - emqx1
|
|
||||||
- emqx1@192.168.1.1.
|
|
||||||
- 1883
|
|
||||||
* - emqx2
|
|
||||||
- emqx2@192.168.1.2
|
|
||||||
- 1883
|
|
||||||
|
|
||||||
|
|
||||||
EMQ X RPC Bridge Configuration
|
|
||||||
------------------------------
|
|
||||||
|
|
||||||
The following is the basic configuration of RPC bridging. A simplest RPC bridging only requires the following three items
|
|
||||||
|
|
||||||
.. code-block::
|
|
||||||
|
|
||||||
## Bridge Address: Use node name (nodename@host) for rpc bridging, and host:port for mqtt connection
|
|
||||||
bridge.mqtt.emqx2.address = "emqx2@192.168.1.2"
|
|
||||||
|
|
||||||
## Forwarding topics of the message
|
|
||||||
bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
|
|
||||||
|
|
||||||
## bridged mountpoint
|
|
||||||
bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
|
|
||||||
|
|
||||||
If the messages received by the local node emqx1 matches the topic ``sersor1/#`` or ``sensor2/#``\ , these messages will be forwarded to the ``sensor1/#`` or ``sensor2/#`` topic of the remote node emqx2.
|
|
||||||
|
|
||||||
``forwards`` is used to specify topics. Messages of the in ``forwards`` specified topics on local node are forwarded to the remote node.
|
|
||||||
|
|
||||||
``mountpoint`` is used to add a topic prefix when forwarding a message. To use ``mountpoint``\ , the ``forwards`` directive must be set. In the above example, a message with the topic ``sensor1/hello`` received by the local node will be forwarded to the remote node with the topic ``bridge/emqx2/emqx1@192.168.1.1/sensor1/hello``.
|
|
||||||
|
|
||||||
Limitations of RPC bridging:
|
|
||||||
|
|
||||||
|
|
||||||
#.
|
|
||||||
The RPC bridge of emqx can only forward local messages to the remote node, and cannot synchronize the messages of the remote node to the local node;
|
|
||||||
|
|
||||||
#.
|
|
||||||
RPC bridge can only bridge two EMQ X broker together and cannot bridge EMQ X broker to other MQTT brokers.
|
|
||||||
|
|
||||||
EMQ X MQTT Bridge Configuration
|
|
||||||
-------------------------------
|
|
||||||
|
|
||||||
EMQ X 3.0 officially introduced MQTT bridge, so that EMQ X can bridge any MQTT broker. Because of the characteristics of the MQTT protocol, EMQ X can subscribe to the remote mqtt broker's topic through MQTT bridge, and then synchronize the remote MQTT broker's message to the local.
|
|
||||||
|
|
||||||
EMQ X MQTT bridging principle: Create an MQTT client on the EMQ X broker, and connect this MQTT client to the remote MQTT broker. Therefore, in the MQTT bridge configuration, following fields may be set for the EMQ X to connect to the remote broker as an mqtt client
|
|
||||||
|
|
||||||
.. code-block::
|
|
||||||
|
|
||||||
## Bridge Address: Use node name for rpc bridging, use host:port for mqtt connection
|
|
||||||
bridge.mqtt.emqx2.address = "192.168.1.2:1883"
|
|
||||||
|
|
||||||
## Bridged Protocol Version
|
|
||||||
## Enumeration value: mqttv3 | mqttv4 | mqttv5
|
|
||||||
bridge.mqtt.emqx2.proto_ver = "mqttv4"
|
|
||||||
|
|
||||||
## mqtt client's clientid
|
|
||||||
bridge.mqtt.emqx2.clientid = "bridge_emq"
|
|
||||||
|
|
||||||
## mqtt client's clean_start field
|
|
||||||
## Note: Some MQTT Brokers need to set the clean_start value as `true`
|
|
||||||
bridge.mqtt.emqx2.clean_start = true
|
|
||||||
|
|
||||||
## mqtt client's username field
|
|
||||||
bridge.mqtt.emqx2.username = "user"
|
|
||||||
|
|
||||||
## mqtt client's password field
|
|
||||||
bridge.mqtt.emqx2.password = "passwd"
|
|
||||||
|
|
||||||
## Whether the mqtt client uses ssl to connect to a remote serve or not
|
|
||||||
bridge.mqtt.emqx2.ssl = off
|
|
||||||
|
|
||||||
## CA Certificate of Client SSL Connection (PEM format)
|
|
||||||
bridge.mqtt.emqx2.cacertfile = "etc/certs/cacert.pem"
|
|
||||||
|
|
||||||
## SSL certificate of Client SSL connection
|
|
||||||
bridge.mqtt.emqx2.certfile = "etc/certs/client-cert.pem"
|
|
||||||
|
|
||||||
## Key file of Client SSL connection
|
|
||||||
bridge.mqtt.emqx2.keyfile = "etc/certs/client-key.pem"
|
|
||||||
|
|
||||||
## TTLS PSK password
|
|
||||||
## Note 'listener.ssl.external.ciphers' and 'listener.ssl.external.psk_ciphers' cannot be configured at the same time
|
|
||||||
##
|
|
||||||
## See 'https://tools.ietf.org/html/rfc4279#section-2'.
|
|
||||||
## bridge.mqtt.emqx2.psk_ciphers = "PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA"
|
|
||||||
|
|
||||||
## Client's heartbeat interval
|
|
||||||
bridge.mqtt.emqx2.keepalive = 60s
|
|
||||||
|
|
||||||
## Supported TLS version
|
|
||||||
bridge.mqtt.emqx2.tls_versions = "tlsv1.2"
|
|
||||||
|
|
||||||
## SSL encryption
|
|
||||||
bridge.mqtt.emqx2.ciphers = "ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384"
|
|
||||||
|
|
||||||
## Forwarding topics of the message
|
|
||||||
bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
|
|
||||||
|
|
||||||
## Bridged mountpoint
|
|
||||||
bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
|
|
||||||
|
|
||||||
## Subscription topic for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.1.topic = "cmd/topic1"
|
|
||||||
|
|
||||||
## Subscription qos for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.1.qos = 1
|
|
||||||
|
|
||||||
## Subscription topic for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.2.topic = "cmd/topic2"
|
|
||||||
|
|
||||||
## Subscription qos for bridging
|
|
||||||
bridge.mqtt.emqx2.subscription.2.qos = 1
|
|
||||||
|
|
||||||
## Bridging reconnection interval
|
|
||||||
## Default: 30s
|
|
||||||
bridge.mqtt.emqx2.reconnect_interval = 30s
|
|
||||||
|
|
||||||
## QoS1 message retransmission interval
|
|
||||||
bridge.mqtt.emqx2.retry_interval = 20s
|
|
||||||
|
|
||||||
## Inflight Size.
|
|
||||||
bridge.mqtt.emqx2.max_inflight_batches = 32
|
|
||||||
|
|
||||||
Bridge Cache Configuration
|
|
||||||
--------------------------
|
|
||||||
|
|
||||||
The bridge of EMQ X has a message caching mechanism. The caching mechanism is applicable to both RPC bridging and MQTT bridging. When the bridge is disconnected (such as when the network connection is unstable), the messages with a topic specified in ``forwards`` can be cached to the local message queue. Until the bridge is restored, these messages are re-forwarded to the remote node. The configuration of the cache queue is as follows
|
|
||||||
|
|
||||||
.. code-block::
|
|
||||||
|
|
||||||
## emqx_bridge internal number of messages used for batch
|
|
||||||
bridge.mqtt.emqx2.queue.batch_count_limit = 32
|
|
||||||
|
|
||||||
## emqx_bridge internal number of message bytes used for batch
|
|
||||||
bridge.mqtt.emqx2.queue.batch_bytes_limit = 1000MB
|
|
||||||
|
|
||||||
## The path for placing replayq queue. If it is not specified, then replayq will run in `mem-only` mode and messages will not be cached on disk.
|
|
||||||
bridge.mqtt.emqx2.queue.replayq_dir = "data/emqx_emqx2_bridge/"
|
|
||||||
|
|
||||||
## Replayq data segment size
|
|
||||||
bridge.mqtt.emqx2.queue.replayq_seg_bytes = 10MB
|
|
||||||
|
|
||||||
``bridge.mqtt.emqx2.queue.replayq_dir`` is a configuration parameter for specifying the path of the bridge storage queue.
|
|
||||||
|
|
||||||
``bridge.mqtt.emqx2.queue.replayq_seg_bytes`` is used to specify the size of the largest single file of the message queue that is cached on disk. If the message queue size exceeds the specified value, a new file is created to store the message queue.
|
|
||||||
|
|
||||||
CLI for EMQ X Bridge MQTT
|
|
||||||
-------------------------
|
|
||||||
|
|
||||||
CLI for EMQ X Bridge MQTT:
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ cd emqx1/ && ./bin/emqx_ctl bridges
|
|
||||||
bridges list # List bridges
|
|
||||||
bridges start <Name> # Start a bridge
|
|
||||||
bridges stop <Name> # Stop a bridge
|
|
||||||
bridges forwards <Name> # Show a bridge forward topic
|
|
||||||
bridges add-forward <Name> <Topic> # Add bridge forward topic
|
|
||||||
bridges del-forward <Name> <Topic> # Delete bridge forward topic
|
|
||||||
bridges subscriptions <Name> # Show a bridge subscriptions topic
|
|
||||||
bridges add-subscription <Name> <Topic> <Qos> # Add bridge subscriptions topic
|
|
||||||
|
|
||||||
List all bridge states
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges list
|
|
||||||
name: emqx status: Stopped $ ./bin/emqx_ctl bridges list
|
|
||||||
name: emqx status: Stopped
|
|
||||||
|
|
||||||
Start the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges start emqx
|
|
||||||
Start bridge successfully.
|
|
||||||
|
|
||||||
Stop the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges stop emqx
|
|
||||||
Stop bridge successfully.
|
|
||||||
|
|
||||||
List the forwarding topics for the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges forwards emqx
|
|
||||||
topic: topic1/#
|
|
||||||
topic: topic2/#
|
|
||||||
|
|
||||||
Add a forwarding topic for the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges add-forwards emqx topic3/#
|
|
||||||
Add-forward topic successfully.
|
|
||||||
|
|
||||||
Delete the forwarding topic for the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges del-forwards emqx topic3/#
|
|
||||||
Del-forward topic successfully.
|
|
||||||
|
|
||||||
List subscriptions for the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges subscriptions emqx
|
|
||||||
topic: cmd/topic1, qos: 1
|
|
||||||
topic: cmd/topic2, qos: 1
|
|
||||||
|
|
||||||
Add a subscription topic for the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges add-subscription emqx cmd/topic3 1
|
|
||||||
Add-subscription topic successfully.
|
|
||||||
|
|
||||||
Delete the subscription topic for the specified bridge
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
$ ./bin/emqx_ctl bridges del-subscription emqx cmd/topic3
|
|
||||||
Del-subscription topic successfully.
|
|
||||||
|
|
||||||
Note: In case of creating multiple bridges, it is convenient to replicate all configuration items of the first bridge, and modify the bridge name and other configuration items if necessary.
|
|
||||||
|
|
||||||
Binary file not shown.
|
Before Width: | Height: | Size: 144 KiB |
|
|
@ -1,56 +0,0 @@
|
||||||
##====================================================================
|
|
||||||
## Configuration for EMQ X MQTT Broker Bridge
|
|
||||||
##====================================================================
|
|
||||||
|
|
||||||
bridge_mqtt: [
|
|
||||||
# {
|
|
||||||
# name: "mqtt1"
|
|
||||||
# start_type: auto
|
|
||||||
# forwards: ["test/#"],
|
|
||||||
# forward_mountpoint: ""
|
|
||||||
# reconnect_interval: "30s"
|
|
||||||
# batch_size: 100
|
|
||||||
# queue {
|
|
||||||
# replayq_dir: "{{ platform_data_dir }}/replayq/bridge_mqtt/"
|
|
||||||
# replayq_seg_bytes: "100MB"
|
|
||||||
# replayq_offload_mode: false
|
|
||||||
# replayq_max_total_bytes: "1GB"
|
|
||||||
# },
|
|
||||||
# config {
|
|
||||||
# conn_type: mqtt
|
|
||||||
# address: "127.0.0.1:1883"
|
|
||||||
# proto_ver: v4
|
|
||||||
# bridge_mode: true
|
|
||||||
# clientid: "client1"
|
|
||||||
# clean_start: true
|
|
||||||
# username: "username1"
|
|
||||||
# password: ""
|
|
||||||
# keepalive: 300
|
|
||||||
# subscriptions: [{
|
|
||||||
# topic: "t/#"
|
|
||||||
# qos: 1
|
|
||||||
# }]
|
|
||||||
# receive_mountpoint: ""
|
|
||||||
# retry_interval: "30s"
|
|
||||||
# max_inflight: 32
|
|
||||||
# }
|
|
||||||
# },
|
|
||||||
# {
|
|
||||||
# name: "rpc1"
|
|
||||||
# start_type: auto
|
|
||||||
# forwards: ["test/#"],
|
|
||||||
# forward_mountpoint: ""
|
|
||||||
# reconnect_interval: "30s"
|
|
||||||
# batch_size: 100
|
|
||||||
# queue {
|
|
||||||
# replayq_dir: "{{ platform_data_dir }}/replayq/bridge_mqtt/"
|
|
||||||
# replayq_seg_bytes: "100MB"
|
|
||||||
# replayq_offload_mode: false
|
|
||||||
# replayq_max_total_bytes: "1GB"
|
|
||||||
# },
|
|
||||||
# config {
|
|
||||||
# conn_type: rpc
|
|
||||||
# node: "emqx@127.0.0.1"
|
|
||||||
# }
|
|
||||||
# }
|
|
||||||
]
|
|
||||||
|
|
@ -1,18 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-define(APP, emqx_bridge_mqtt).
|
|
||||||
|
|
||||||
|
|
@ -1,19 +0,0 @@
|
||||||
{deps, []}.
|
|
||||||
{edoc_opts, [{preprocess, true}]}.
|
|
||||||
{erl_opts, [warn_unused_vars,
|
|
||||||
warn_shadow_vars,
|
|
||||||
warn_unused_import,
|
|
||||||
warn_obsolete_guard,
|
|
||||||
debug_info]}.
|
|
||||||
|
|
||||||
{xref_checks, [undefined_function_calls, undefined_functions,
|
|
||||||
locals_not_used, deprecated_function_calls,
|
|
||||||
warnings_as_errors, deprecated_functions]}.
|
|
||||||
{cover_enabled, true}.
|
|
||||||
{cover_opts, [verbose]}.
|
|
||||||
{cover_export_enabled, true}.
|
|
||||||
|
|
||||||
{shell, [
|
|
||||||
% {config, "config/sys.config"},
|
|
||||||
{apps, [emqx, emqx_bridge_mqtt]}
|
|
||||||
]}.
|
|
||||||
|
|
@ -1,14 +0,0 @@
|
||||||
{application, emqx_bridge_mqtt,
|
|
||||||
[{description, "EMQ X Bridge to MQTT Broker"},
|
|
||||||
{vsn, "5.0.0"}, % strict semver, bump manually!
|
|
||||||
{modules, []},
|
|
||||||
{registered, []},
|
|
||||||
{applications, [kernel,stdlib,replayq,emqtt,emqx]},
|
|
||||||
{mod, {emqx_bridge_mqtt_app, []}},
|
|
||||||
{env, []},
|
|
||||||
{licenses, ["Apache-2.0"]},
|
|
||||||
{maintainers, ["EMQ X Team <contact@emqx.io>"]},
|
|
||||||
{links, [{"Homepage", "https://emqx.io/"},
|
|
||||||
{"Github", "https://github.com/emqx/emqx-bridge-mqtt"}
|
|
||||||
]}
|
|
||||||
]}.
|
|
||||||
|
|
@ -1,31 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_mqtt_app).
|
|
||||||
|
|
||||||
-behaviour(application).
|
|
||||||
|
|
||||||
-export([start/2, stop/1]).
|
|
||||||
|
|
||||||
start(_StartType, _StartArgs) ->
|
|
||||||
emqx_ctl:register_command(bridges, {emqx_bridge_mqtt_cli, cli}, []),
|
|
||||||
emqx_bridge_worker:register_metrics(),
|
|
||||||
emqx_bridge_mqtt_sup:start_link().
|
|
||||||
|
|
||||||
stop(_State) ->
|
|
||||||
emqx_ctl:unregister_command(bridges),
|
|
||||||
ok.
|
|
||||||
|
|
||||||
|
|
@ -1,92 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_mqtt_cli).
|
|
||||||
|
|
||||||
-include("emqx_bridge_mqtt.hrl").
|
|
||||||
|
|
||||||
-import(lists, [foreach/2]).
|
|
||||||
|
|
||||||
-export([cli/1]).
|
|
||||||
|
|
||||||
cli(["list"]) ->
|
|
||||||
foreach(fun({Name, State0}) ->
|
|
||||||
State = case State0 of
|
|
||||||
connected -> <<"Running">>;
|
|
||||||
_ -> <<"Stopped">>
|
|
||||||
end,
|
|
||||||
emqx_ctl:print("name: ~s status: ~s~n", [Name, State])
|
|
||||||
end, emqx_bridge_mqtt_sup:bridges());
|
|
||||||
|
|
||||||
cli(["start", Name]) ->
|
|
||||||
emqx_ctl:print("~s.~n", [try emqx_bridge_worker:ensure_started(Name) of
|
|
||||||
ok -> <<"Start bridge successfully">>;
|
|
||||||
connected -> <<"Bridge already started">>;
|
|
||||||
_ -> <<"Start bridge failed">>
|
|
||||||
catch
|
|
||||||
_Error:_Reason ->
|
|
||||||
<<"Start bridge failed">>
|
|
||||||
end]);
|
|
||||||
|
|
||||||
cli(["stop", Name]) ->
|
|
||||||
emqx_ctl:print("~s.~n", [try emqx_bridge_worker:ensure_stopped(Name) of
|
|
||||||
ok -> <<"Stop bridge successfully">>;
|
|
||||||
_ -> <<"Stop bridge failed">>
|
|
||||||
catch
|
|
||||||
_Error:_Reason ->
|
|
||||||
<<"Stop bridge failed">>
|
|
||||||
end]);
|
|
||||||
|
|
||||||
cli(["forwards", Name]) ->
|
|
||||||
foreach(fun(Topic) ->
|
|
||||||
emqx_ctl:print("topic: ~s~n", [Topic])
|
|
||||||
end, emqx_bridge_worker:get_forwards(Name));
|
|
||||||
|
|
||||||
cli(["add-forward", Name, Topic]) ->
|
|
||||||
ok = emqx_bridge_worker:ensure_forward_present(Name, iolist_to_binary(Topic)),
|
|
||||||
emqx_ctl:print("Add-forward topic successfully.~n");
|
|
||||||
|
|
||||||
cli(["del-forward", Name, Topic]) ->
|
|
||||||
ok = emqx_bridge_worker:ensure_forward_absent(Name, iolist_to_binary(Topic)),
|
|
||||||
emqx_ctl:print("Del-forward topic successfully.~n");
|
|
||||||
|
|
||||||
cli(["subscriptions", Name]) ->
|
|
||||||
foreach(fun({Topic, Qos}) ->
|
|
||||||
emqx_ctl:print("topic: ~s, qos: ~p~n", [Topic, Qos])
|
|
||||||
end, emqx_bridge_worker:get_subscriptions(Name));
|
|
||||||
|
|
||||||
cli(["add-subscription", Name, Topic, Qos]) ->
|
|
||||||
case emqx_bridge_worker:ensure_subscription_present(Name, Topic, list_to_integer(Qos)) of
|
|
||||||
ok -> emqx_ctl:print("Add-subscription topic successfully.~n");
|
|
||||||
{error, Reason} -> emqx_ctl:print("Add-subscription failed reason: ~p.~n", [Reason])
|
|
||||||
end;
|
|
||||||
|
|
||||||
cli(["del-subscription", Name, Topic]) ->
|
|
||||||
ok = emqx_bridge_worker:ensure_subscription_absent(Name, Topic),
|
|
||||||
emqx_ctl:print("Del-subscription topic successfully.~n");
|
|
||||||
|
|
||||||
cli(_) ->
|
|
||||||
emqx_ctl:usage([{"bridges list", "List bridges"},
|
|
||||||
{"bridges start <Name>", "Start a bridge"},
|
|
||||||
{"bridges stop <Name>", "Stop a bridge"},
|
|
||||||
{"bridges forwards <Name>", "Show a bridge forward topic"},
|
|
||||||
{"bridges add-forward <Name> <Topic>", "Add bridge forward topic"},
|
|
||||||
{"bridges del-forward <Name> <Topic>", "Delete bridge forward topic"},
|
|
||||||
{"bridges subscriptions <Name>", "Show a bridge subscriptions topic"},
|
|
||||||
{"bridges add-subscription <Name> <Topic> <Qos>", "Add bridge subscriptions topic"},
|
|
||||||
{"bridges del-subscription <Name> <Topic>", "Delete bridge subscriptions topic"}]).
|
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -1,94 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_mqtt_schema).
|
|
||||||
|
|
||||||
-include_lib("typerefl/include/types.hrl").
|
|
||||||
|
|
||||||
-behaviour(hocon_schema).
|
|
||||||
|
|
||||||
-export([ namespace/0
|
|
||||||
, roots/0
|
|
||||||
, fields/1]).
|
|
||||||
|
|
||||||
namespace() -> "bridge_mqtt".
|
|
||||||
|
|
||||||
roots() -> [array("bridge_mqtt")].
|
|
||||||
|
|
||||||
array(Name) -> {Name, hoconsc:array(hoconsc:ref(?MODULE, Name))}.
|
|
||||||
|
|
||||||
fields("bridge_mqtt") ->
|
|
||||||
[ {name, sc(string(), #{default => true})}
|
|
||||||
, {start_type, fun start_type/1}
|
|
||||||
, {forwards, fun forwards/1}
|
|
||||||
, {forward_mountpoint, sc(string(), #{})}
|
|
||||||
, {reconnect_interval, sc(emqx_schema:duration_ms(), #{default => "30s"})}
|
|
||||||
, {batch_size, sc(integer(), #{default => 100})}
|
|
||||||
, {queue, sc(hoconsc:ref(?MODULE, "queue"), #{})}
|
|
||||||
, {config, sc(hoconsc:union([hoconsc:ref(?MODULE, "mqtt"),
|
|
||||||
hoconsc:ref(?MODULE, "rpc")]),
|
|
||||||
#{})}
|
|
||||||
];
|
|
||||||
|
|
||||||
fields("mqtt") ->
|
|
||||||
[ {conn_type, fun conn_type/1}
|
|
||||||
, {address, sc(string(), #{default => "127.0.0.1:1883"})}
|
|
||||||
, {proto_ver, fun proto_ver/1}
|
|
||||||
, {bridge_mode, sc(boolean(), #{default => true})}
|
|
||||||
, {clientid, sc(string(), #{})}
|
|
||||||
, {username, sc(string(), #{})}
|
|
||||||
, {password, sc(string(), #{})}
|
|
||||||
, {clean_start, sc(boolean(), #{default => true})}
|
|
||||||
, {keepalive, sc(integer(), #{default => 300})}
|
|
||||||
, {subscriptions, sc(hoconsc:array(hoconsc:ref(?MODULE, "subscriptions")), #{})}
|
|
||||||
, {receive_mountpoint, sc(string(), #{})}
|
|
||||||
, {retry_interval, sc(emqx_schema:duration_ms(), #{default => "30s"})}
|
|
||||||
, {max_inflight, sc(integer(), #{default => 32})}
|
|
||||||
];
|
|
||||||
|
|
||||||
fields("rpc") ->
|
|
||||||
[ {conn_type, fun conn_type/1}
|
|
||||||
, {node, sc(atom(), #{default => 'emqx@127.0.0.1'})}
|
|
||||||
];
|
|
||||||
|
|
||||||
fields("subscriptions") ->
|
|
||||||
[ {topic, #{type => binary(), nullable => false}}
|
|
||||||
, {qos, sc(integer(), #{default => 1})}
|
|
||||||
];
|
|
||||||
|
|
||||||
fields("queue") ->
|
|
||||||
[ {replayq_dir, hoconsc:union([boolean(), string()])}
|
|
||||||
, {replayq_seg_bytes, sc(emqx_schema:bytesize(), #{default => "100MB"})}
|
|
||||||
, {replayq_offload_mode, sc(boolean(), #{default => false})}
|
|
||||||
, {replayq_max_total_bytes, sc(emqx_schema:bytesize(), #{default => "1024MB"})}
|
|
||||||
].
|
|
||||||
|
|
||||||
conn_type(type) -> hoconsc:enum([mqtt, rpc]);
|
|
||||||
conn_type(_) -> undefined.
|
|
||||||
|
|
||||||
proto_ver(type) -> hoconsc:enum([v3, v4, v5]);
|
|
||||||
proto_ver(default) -> v4;
|
|
||||||
proto_ver(_) -> undefined.
|
|
||||||
|
|
||||||
start_type(type) -> hoconsc:enum([auto, manual]);
|
|
||||||
start_type(default) -> auto;
|
|
||||||
start_type(_) -> undefined.
|
|
||||||
|
|
||||||
forwards(type) -> hoconsc:array(binary());
|
|
||||||
forwards(default) -> [];
|
|
||||||
forwards(_) -> undefined.
|
|
||||||
|
|
||||||
sc(Type, Meta) -> hoconsc:mk(Type, Meta).
|
|
||||||
|
|
@ -1,69 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_mqtt_sup).
|
|
||||||
-behaviour(supervisor).
|
|
||||||
|
|
||||||
-include("emqx_bridge_mqtt.hrl").
|
|
||||||
-include_lib("emqx/include/logger.hrl").
|
|
||||||
|
|
||||||
|
|
||||||
%% APIs
|
|
||||||
-export([ start_link/0
|
|
||||||
]).
|
|
||||||
|
|
||||||
-export([ create_bridge/1
|
|
||||||
, drop_bridge/1
|
|
||||||
, bridges/0
|
|
||||||
]).
|
|
||||||
|
|
||||||
%% supervisor callbacks
|
|
||||||
-export([init/1]).
|
|
||||||
|
|
||||||
-define(WORKER_SUP, emqx_bridge_worker_sup).
|
|
||||||
|
|
||||||
start_link() ->
|
|
||||||
supervisor:start_link({local, ?MODULE}, ?MODULE, []).
|
|
||||||
|
|
||||||
init([]) ->
|
|
||||||
SupFlag = #{strategy => one_for_one,
|
|
||||||
intensity => 100,
|
|
||||||
period => 10},
|
|
||||||
{ok, {SupFlag, []}}.
|
|
||||||
|
|
||||||
bridge_spec(Config) ->
|
|
||||||
#{id => maps:get(name, Config),
|
|
||||||
start => {emqx_bridge_worker, start_link, [Config]},
|
|
||||||
restart => permanent,
|
|
||||||
shutdown => 5000,
|
|
||||||
type => worker,
|
|
||||||
modules => [emqx_bridge_worker]}.
|
|
||||||
|
|
||||||
-spec(bridges() -> [{node(), map()}]).
|
|
||||||
bridges() ->
|
|
||||||
[{Name, emqx_bridge_worker:status(Name)}
|
|
||||||
|| {Name, _Pid, _, _} <- supervisor:which_children(?MODULE)].
|
|
||||||
|
|
||||||
create_bridge(Config) ->
|
|
||||||
supervisor:start_child(?MODULE, bridge_spec(Config)).
|
|
||||||
|
|
||||||
drop_bridge(Name) ->
|
|
||||||
case supervisor:terminate_child(?MODULE, Name) of
|
|
||||||
ok ->
|
|
||||||
supervisor:delete_child(?MODULE, Name);
|
|
||||||
{error, Error} ->
|
|
||||||
{error, Error}
|
|
||||||
end.
|
|
||||||
|
|
@ -1,95 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
%% @doc This module implements EMQX Bridge transport layer based on gen_rpc.
|
|
||||||
|
|
||||||
-module(emqx_bridge_rpc).
|
|
||||||
|
|
||||||
-export([ start/1
|
|
||||||
, send/2
|
|
||||||
, stop/1
|
|
||||||
]).
|
|
||||||
|
|
||||||
%% Internal exports
|
|
||||||
-export([ handle_send/1
|
|
||||||
, heartbeat/2
|
|
||||||
]).
|
|
||||||
|
|
||||||
-type ack_ref() :: emqx_bridge_worker:ack_ref().
|
|
||||||
-type batch() :: emqx_bridge_worker:batch().
|
|
||||||
-define(HEARTBEAT_INTERVAL, timer:seconds(1)).
|
|
||||||
|
|
||||||
-define(RPC, emqx_rpc).
|
|
||||||
|
|
||||||
start(#{node := RemoteNode}) ->
|
|
||||||
case poke(RemoteNode) of
|
|
||||||
ok ->
|
|
||||||
Pid = proc_lib:spawn_link(?MODULE, heartbeat, [self(), RemoteNode]),
|
|
||||||
{ok, #{client_pid => Pid, remote_node => RemoteNode}};
|
|
||||||
Error ->
|
|
||||||
Error
|
|
||||||
end.
|
|
||||||
|
|
||||||
stop(#{client_pid := Pid}) when is_pid(Pid) ->
|
|
||||||
Ref = erlang:monitor(process, Pid),
|
|
||||||
unlink(Pid),
|
|
||||||
Pid ! stop,
|
|
||||||
receive
|
|
||||||
{'DOWN', Ref, process, Pid, _Reason} ->
|
|
||||||
ok
|
|
||||||
after
|
|
||||||
1000 ->
|
|
||||||
exit(Pid, kill)
|
|
||||||
end,
|
|
||||||
ok.
|
|
||||||
|
|
||||||
%% @doc Callback for `emqx_bridge_connect' behaviour
|
|
||||||
-spec send(#{remote_node := atom(), _ => _}, batch()) -> {ok, ack_ref()} | {error, any()}.
|
|
||||||
send(#{remote_node := RemoteNode}, Batch) ->
|
|
||||||
case ?RPC:call(RemoteNode, ?MODULE, handle_send, [Batch]) of
|
|
||||||
ok ->
|
|
||||||
Ref = make_ref(),
|
|
||||||
self() ! {batch_ack, Ref},
|
|
||||||
{ok, Ref};
|
|
||||||
{badrpc, Reason} -> {error, Reason}
|
|
||||||
end.
|
|
||||||
|
|
||||||
%% @doc Handle send on receiver side.
|
|
||||||
-spec handle_send(batch()) -> ok.
|
|
||||||
handle_send(Batch) ->
|
|
||||||
lists:foreach(fun(Msg) -> emqx_broker:publish(Msg) end, Batch).
|
|
||||||
|
|
||||||
%% @hidden Heartbeat loop
|
|
||||||
heartbeat(Parent, RemoteNode) ->
|
|
||||||
Interval = ?HEARTBEAT_INTERVAL,
|
|
||||||
receive
|
|
||||||
stop -> exit(normal)
|
|
||||||
after
|
|
||||||
Interval ->
|
|
||||||
case poke(RemoteNode) of
|
|
||||||
ok ->
|
|
||||||
?MODULE:heartbeat(Parent, RemoteNode);
|
|
||||||
{error, Reason} ->
|
|
||||||
Parent ! {disconnected, self(), Reason},
|
|
||||||
exit(normal)
|
|
||||||
end
|
|
||||||
end.
|
|
||||||
|
|
||||||
poke(RemoteNode) ->
|
|
||||||
case ?RPC:call(RemoteNode, erlang, node, []) of
|
|
||||||
RemoteNode -> ok;
|
|
||||||
{badrpc, Reason} -> {error, Reason}
|
|
||||||
end.
|
|
||||||
|
|
@ -1,42 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_rpc_tests).
|
|
||||||
-include_lib("eunit/include/eunit.hrl").
|
|
||||||
|
|
||||||
send_and_ack_test() ->
|
|
||||||
%% delegate from emqx_rpc to rpc for unit test
|
|
||||||
meck:new(emqx_rpc, [passthrough, no_history]),
|
|
||||||
meck:expect(emqx_rpc, call, 4,
|
|
||||||
fun(Node, Module, Fun, Args) ->
|
|
||||||
rpc:call(Node, Module, Fun, Args)
|
|
||||||
end),
|
|
||||||
meck:expect(emqx_rpc, cast, 4,
|
|
||||||
fun(Node, Module, Fun, Args) ->
|
|
||||||
rpc:cast(Node, Module, Fun, Args)
|
|
||||||
end),
|
|
||||||
meck:new(emqx_bridge_worker, [passthrough, no_history]),
|
|
||||||
try
|
|
||||||
{ok, #{client_pid := Pid, remote_node := Node}} = emqx_bridge_rpc:start(#{node => node()}),
|
|
||||||
{ok, Ref} = emqx_bridge_rpc:send(#{remote_node => Node}, []),
|
|
||||||
receive
|
|
||||||
{batch_ack, Ref} ->
|
|
||||||
ok
|
|
||||||
end,
|
|
||||||
ok = emqx_bridge_rpc:stop( #{client_pid => Pid})
|
|
||||||
after
|
|
||||||
meck:unload(emqx_rpc)
|
|
||||||
end.
|
|
||||||
|
|
@ -1,38 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_stub_conn).
|
|
||||||
|
|
||||||
-export([ start/1
|
|
||||||
, send/2
|
|
||||||
, stop/1
|
|
||||||
]).
|
|
||||||
|
|
||||||
-type ack_ref() :: emqx_bridge_worker:ack_ref().
|
|
||||||
-type batch() :: emqx_bridge_worker:batch().
|
|
||||||
|
|
||||||
start(#{client_pid := Pid} = Cfg) ->
|
|
||||||
Pid ! {self(), ?MODULE, ready},
|
|
||||||
{ok, Cfg}.
|
|
||||||
|
|
||||||
stop(_) -> ok.
|
|
||||||
|
|
||||||
%% @doc Callback for `emqx_bridge_connect' behaviour
|
|
||||||
-spec send(_, batch()) -> {ok, ack_ref()} | {error, any()}.
|
|
||||||
send(#{client_pid := Pid}, Batch) ->
|
|
||||||
Ref = make_ref(),
|
|
||||||
Pid ! {stub_message, self(), Ref, Batch},
|
|
||||||
{ok, Ref}.
|
|
||||||
|
|
@ -1,372 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_worker_SUITE).
|
|
||||||
|
|
||||||
-compile(export_all).
|
|
||||||
-compile(nowarn_export_all).
|
|
||||||
|
|
||||||
-include_lib("eunit/include/eunit.hrl").
|
|
||||||
-include_lib("common_test/include/ct.hrl").
|
|
||||||
-include_lib("emqx/include/emqx_mqtt.hrl").
|
|
||||||
-include_lib("emqx/include/emqx.hrl").
|
|
||||||
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
|
|
||||||
|
|
||||||
-define(wait(For, Timeout), emqx_ct_helpers:wait_for(?FUNCTION_NAME, ?LINE, fun() -> For end, Timeout)).
|
|
||||||
|
|
||||||
-define(SNK_WAIT(WHAT), ?assertMatch({ok, _}, ?block_until(#{?snk_kind := WHAT}, 2000, 1000))).
|
|
||||||
|
|
||||||
receive_messages(Count) ->
|
|
||||||
receive_messages(Count, []).
|
|
||||||
|
|
||||||
receive_messages(0, Msgs) ->
|
|
||||||
Msgs;
|
|
||||||
receive_messages(Count, Msgs) ->
|
|
||||||
receive
|
|
||||||
{publish, Msg} ->
|
|
||||||
receive_messages(Count-1, [Msg|Msgs]);
|
|
||||||
_Other ->
|
|
||||||
receive_messages(Count, Msgs)
|
|
||||||
after 1000 ->
|
|
||||||
Msgs
|
|
||||||
end.
|
|
||||||
|
|
||||||
all() ->
|
|
||||||
lists:filtermap(
|
|
||||||
fun({FunName, _Arity}) ->
|
|
||||||
case atom_to_list(FunName) of
|
|
||||||
"t_" ++ _ -> {true, FunName};
|
|
||||||
_ -> false
|
|
||||||
end
|
|
||||||
end,
|
|
||||||
?MODULE:module_info(exports)).
|
|
||||||
|
|
||||||
init_per_suite(Config) ->
|
|
||||||
case node() of
|
|
||||||
nonode@nohost -> net_kernel:start(['emqx@127.0.0.1', longnames]);
|
|
||||||
_ -> ok
|
|
||||||
end,
|
|
||||||
emqx_ct_helpers:start_apps([emqx_bridge_mqtt]),
|
|
||||||
emqx_logger:set_log_level(error),
|
|
||||||
[{log_level, error} | Config].
|
|
||||||
|
|
||||||
end_per_suite(_Config) ->
|
|
||||||
emqx_ct_helpers:stop_apps([emqx_bridge_mqtt]).
|
|
||||||
|
|
||||||
init_per_testcase(_TestCase, Config) ->
|
|
||||||
ok = snabbkaffe:start_trace(),
|
|
||||||
Config.
|
|
||||||
|
|
||||||
end_per_testcase(_TestCase, _Config) ->
|
|
||||||
ok = snabbkaffe:stop().
|
|
||||||
|
|
||||||
t_rpc_mngr(_Config) ->
|
|
||||||
Name = "rpc_name",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [<<"mngr">>],
|
|
||||||
forward_mountpoint => <<"forwarded">>,
|
|
||||||
start_type => auto,
|
|
||||||
config => #{
|
|
||||||
conn_type => rpc,
|
|
||||||
node => node()
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Pid} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
?assertEqual([<<"mngr">>], emqx_bridge_worker:get_forwards(Name)),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_present(Name, "mngr")),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_present(Name, "mngr2")),
|
|
||||||
?assertEqual([<<"mngr2">>, <<"mngr">>], emqx_bridge_worker:get_forwards(Name)),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_absent(Name, "mngr2")),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_absent(Name, "mngr3")),
|
|
||||||
?assertEqual([<<"mngr">>], emqx_bridge_worker:get_forwards(Name)),
|
|
||||||
?assertEqual({error, no_remote_subscription_support},
|
|
||||||
emqx_bridge_worker:ensure_subscription_present(Name, <<"t">>, 0)),
|
|
||||||
?assertEqual({error, no_remote_subscription_support},
|
|
||||||
emqx_bridge_worker:ensure_subscription_absent(Name, <<"t">>)),
|
|
||||||
ok = emqx_bridge_worker:stop(Pid).
|
|
||||||
|
|
||||||
t_mqtt_mngr(_Config) ->
|
|
||||||
Name = "mqtt_name",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [<<"mngr">>],
|
|
||||||
forward_mountpoint => <<"forwarded">>,
|
|
||||||
start_type => auto,
|
|
||||||
config => #{
|
|
||||||
address => "127.0.0.1:1883",
|
|
||||||
conn_type => mqtt,
|
|
||||||
clientid => <<"client1">>,
|
|
||||||
keepalive => 300,
|
|
||||||
subscriptions => [#{topic => <<"t/#">>, qos => 1}]
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Pid} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
?assertEqual([<<"mngr">>], emqx_bridge_worker:get_forwards(Name)),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_present(Name, "mngr")),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_present(Name, "mngr2")),
|
|
||||||
?assertEqual([<<"mngr2">>, <<"mngr">>], emqx_bridge_worker:get_forwards(Name)),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_absent(Name, "mngr2")),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_forward_absent(Name, "mngr3")),
|
|
||||||
?assertEqual([<<"mngr">>], emqx_bridge_worker:get_forwards(Name)),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_subscription_present(Name, <<"t">>, 0)),
|
|
||||||
?assertEqual(ok, emqx_bridge_worker:ensure_subscription_absent(Name, <<"t">>)),
|
|
||||||
?assertEqual([{<<"t/#">>,1}], emqx_bridge_worker:get_subscriptions(Name)),
|
|
||||||
ok = emqx_bridge_worker:stop(Pid).
|
|
||||||
|
|
||||||
%% A loopback RPC to local node
|
|
||||||
t_rpc(_Config) ->
|
|
||||||
Name = "rpc",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [<<"t_rpc/#">>],
|
|
||||||
forward_mountpoint => <<"forwarded">>,
|
|
||||||
start_type => auto,
|
|
||||||
config => #{
|
|
||||||
conn_type => rpc,
|
|
||||||
node => node()
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Pid} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
{ok, ConnPid} = emqtt:start_link([{clientid, <<"ClientId">>}]),
|
|
||||||
{ok, _Props} = emqtt:connect(ConnPid),
|
|
||||||
{ok, _Props, [1]} = emqtt:subscribe(ConnPid, {<<"forwarded/t_rpc/one">>, ?QOS_1}),
|
|
||||||
timer:sleep(100),
|
|
||||||
{ok, _PacketId} = emqtt:publish(ConnPid, <<"t_rpc/one">>, <<"hello">>, ?QOS_1),
|
|
||||||
timer:sleep(100),
|
|
||||||
?assertEqual(1, length(receive_messages(1))),
|
|
||||||
emqtt:disconnect(ConnPid),
|
|
||||||
emqx_bridge_worker:stop(Pid).
|
|
||||||
|
|
||||||
%% Full data loopback flow explained:
|
|
||||||
%% mqtt-client ----> local-broker ---(local-subscription)--->
|
|
||||||
%% bridge(export) --- (mqtt-connection)--> local-broker ---(remote-subscription) -->
|
|
||||||
%% bridge(import) --> mqtt-client
|
|
||||||
t_mqtt(_Config) ->
|
|
||||||
SendToTopic = <<"t_mqtt/one">>,
|
|
||||||
SendToTopic2 = <<"t_mqtt/two">>,
|
|
||||||
SendToTopic3 = <<"t_mqtt/three">>,
|
|
||||||
Mountpoint = <<"forwarded/${node}/">>,
|
|
||||||
Name = "mqtt",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [SendToTopic],
|
|
||||||
forward_mountpoint => Mountpoint,
|
|
||||||
start_type => auto,
|
|
||||||
config => #{
|
|
||||||
address => "127.0.0.1:1883",
|
|
||||||
conn_type => mqtt,
|
|
||||||
clientid => <<"client1">>,
|
|
||||||
keepalive => 300,
|
|
||||||
subscriptions => [#{topic => SendToTopic2, qos => 1}],
|
|
||||||
receive_mountpoint => <<"receive/aws/">>
|
|
||||||
},
|
|
||||||
queue => #{
|
|
||||||
replayq_dir => "data/t_mqtt/",
|
|
||||||
replayq_seg_bytes => 10000,
|
|
||||||
batch_bytes_limit => 1000,
|
|
||||||
batch_count_limit => 10
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Pid} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
?assertEqual([{SendToTopic2, 1}], emqx_bridge_worker:get_subscriptions(Name)),
|
|
||||||
ok = emqx_bridge_worker:ensure_subscription_present(Name, SendToTopic3, _QoS = 1),
|
|
||||||
?assertEqual([{SendToTopic3, 1},{SendToTopic2, 1}],
|
|
||||||
emqx_bridge_worker:get_subscriptions(Name)),
|
|
||||||
{ok, ConnPid} = emqtt:start_link([{clientid, <<"client-1">>}]),
|
|
||||||
{ok, _Props} = emqtt:connect(ConnPid),
|
|
||||||
emqtt:subscribe(ConnPid, <<"forwarded/+/t_mqtt/one">>, 1),
|
|
||||||
%% message from a different client, to avoid getting terminated by no-local
|
|
||||||
Max = 10,
|
|
||||||
Msgs = lists:seq(1, Max),
|
|
||||||
lists:foreach(fun(I) ->
|
|
||||||
{ok, _PacketId} = emqtt:publish(ConnPid, SendToTopic, integer_to_binary(I), ?QOS_1)
|
|
||||||
end, Msgs),
|
|
||||||
?assertEqual(10, length(receive_messages(200))),
|
|
||||||
|
|
||||||
emqtt:subscribe(ConnPid, <<"receive/aws/t_mqtt/two">>, 1),
|
|
||||||
%% message from a different client, to avoid getting terminated by no-local
|
|
||||||
Max = 10,
|
|
||||||
Msgs = lists:seq(1, Max),
|
|
||||||
lists:foreach(fun(I) ->
|
|
||||||
{ok, _PacketId} = emqtt:publish(ConnPid, SendToTopic2, integer_to_binary(I), ?QOS_1)
|
|
||||||
end, Msgs),
|
|
||||||
?assertEqual(10, length(receive_messages(200))),
|
|
||||||
|
|
||||||
emqtt:disconnect(ConnPid),
|
|
||||||
ok = emqx_bridge_worker:stop(Pid).
|
|
||||||
|
|
||||||
t_stub_normal(Config) when is_list(Config) ->
|
|
||||||
Name = "stub_normal",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [<<"t_stub_normal/#">>],
|
|
||||||
forward_mountpoint => <<"forwarded">>,
|
|
||||||
start_type => auto,
|
|
||||||
config => #{
|
|
||||||
conn_type => emqx_bridge_stub_conn,
|
|
||||||
client_pid => self()
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Pid} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
receive
|
|
||||||
{Pid, emqx_bridge_stub_conn, ready} -> ok
|
|
||||||
after
|
|
||||||
5000 ->
|
|
||||||
error(timeout)
|
|
||||||
end,
|
|
||||||
{ok, ConnPid} = emqtt:start_link([{clientid, <<"ClientId">>}]),
|
|
||||||
{ok, _} = emqtt:connect(ConnPid),
|
|
||||||
{ok, _PacketId} = emqtt:publish(ConnPid, <<"t_stub_normal/one">>, <<"hello">>, ?QOS_1),
|
|
||||||
receive
|
|
||||||
{stub_message, WorkerPid, BatchRef, _Batch} ->
|
|
||||||
WorkerPid ! {batch_ack, BatchRef},
|
|
||||||
ok
|
|
||||||
after
|
|
||||||
5000 ->
|
|
||||||
error(timeout)
|
|
||||||
end,
|
|
||||||
?SNK_WAIT(inflight_drained),
|
|
||||||
?SNK_WAIT(replayq_drained),
|
|
||||||
emqtt:disconnect(ConnPid),
|
|
||||||
ok = emqx_bridge_worker:stop(Pid).
|
|
||||||
|
|
||||||
t_stub_overflow(_Config) ->
|
|
||||||
Topic = <<"t_stub_overflow/one">>,
|
|
||||||
MaxInflight = 20,
|
|
||||||
Name = "stub_overflow",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [<<"t_stub_overflow/one">>],
|
|
||||||
forward_mountpoint => <<"forwarded">>,
|
|
||||||
start_type => auto,
|
|
||||||
max_inflight => MaxInflight,
|
|
||||||
config => #{
|
|
||||||
conn_type => emqx_bridge_stub_conn,
|
|
||||||
client_pid => self()
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Worker} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
{ok, ConnPid} = emqtt:start_link([{clientid, <<"ClientId">>}]),
|
|
||||||
{ok, _} = emqtt:connect(ConnPid),
|
|
||||||
lists:foreach(
|
|
||||||
fun(I) ->
|
|
||||||
Data = integer_to_binary(I),
|
|
||||||
_ = emqtt:publish(ConnPid, Topic, Data, ?QOS_1)
|
|
||||||
end, lists:seq(1, MaxInflight * 2)),
|
|
||||||
?SNK_WAIT(inflight_full),
|
|
||||||
Acks = stub_receive(MaxInflight),
|
|
||||||
lists:foreach(fun({Pid, Ref}) -> Pid ! {batch_ack, Ref} end, Acks),
|
|
||||||
Acks2 = stub_receive(MaxInflight),
|
|
||||||
lists:foreach(fun({Pid, Ref}) -> Pid ! {batch_ack, Ref} end, Acks2),
|
|
||||||
?SNK_WAIT(inflight_drained),
|
|
||||||
?SNK_WAIT(replayq_drained),
|
|
||||||
emqtt:disconnect(ConnPid),
|
|
||||||
ok = emqx_bridge_worker:stop(Worker).
|
|
||||||
|
|
||||||
t_stub_random_order(_Config) ->
|
|
||||||
Topic = <<"t_stub_random_order/a">>,
|
|
||||||
MaxInflight = 10,
|
|
||||||
Name = "stub_random_order",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [Topic],
|
|
||||||
forward_mountpoint => <<"forwarded">>,
|
|
||||||
start_type => auto,
|
|
||||||
max_inflight => MaxInflight,
|
|
||||||
config => #{
|
|
||||||
conn_type => emqx_bridge_stub_conn,
|
|
||||||
client_pid => self()
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Worker} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
ClientId = <<"ClientId">>,
|
|
||||||
{ok, ConnPid} = emqtt:start_link([{clientid, ClientId}]),
|
|
||||||
{ok, _} = emqtt:connect(ConnPid),
|
|
||||||
lists:foreach(
|
|
||||||
fun(I) ->
|
|
||||||
Data = integer_to_binary(I),
|
|
||||||
_ = emqtt:publish(ConnPid, Topic, Data, ?QOS_1)
|
|
||||||
end, lists:seq(1, MaxInflight)),
|
|
||||||
Acks = stub_receive(MaxInflight),
|
|
||||||
lists:foreach(fun({Pid, Ref}) -> Pid ! {batch_ack, Ref} end,
|
|
||||||
lists:reverse(Acks)),
|
|
||||||
?SNK_WAIT(inflight_drained),
|
|
||||||
?SNK_WAIT(replayq_drained),
|
|
||||||
emqtt:disconnect(ConnPid),
|
|
||||||
ok = emqx_bridge_worker:stop(Worker).
|
|
||||||
|
|
||||||
t_stub_retry_inflight(_Config) ->
|
|
||||||
Topic = <<"to_stub_retry_inflight/a">>,
|
|
||||||
MaxInflight = 10,
|
|
||||||
Name = "stub_retry_inflight",
|
|
||||||
Cfg = #{
|
|
||||||
name => Name,
|
|
||||||
forwards => [Topic],
|
|
||||||
forward_mountpoint => <<"forwarded">>,
|
|
||||||
reconnect_interval => 10,
|
|
||||||
start_type => auto,
|
|
||||||
max_inflight => MaxInflight,
|
|
||||||
config => #{
|
|
||||||
conn_type => emqx_bridge_stub_conn,
|
|
||||||
client_pid => self()
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{ok, Worker} = emqx_bridge_mqtt_sup:create_bridge(Cfg),
|
|
||||||
ClientId = <<"ClientId2">>,
|
|
||||||
case ?block_until(#{?snk_kind := connected, inflight := 0}, 2000, 1000) of
|
|
||||||
{ok, #{inflight := 0}} -> ok;
|
|
||||||
Other -> ct:fail("~p", [Other])
|
|
||||||
end,
|
|
||||||
{ok, ConnPid} = emqtt:start_link([{clientid, ClientId}]),
|
|
||||||
{ok, _} = emqtt:connect(ConnPid),
|
|
||||||
lists:foreach(
|
|
||||||
fun(I) ->
|
|
||||||
Data = integer_to_binary(I),
|
|
||||||
_ = emqtt:publish(ConnPid, Topic, Data, ?QOS_1)
|
|
||||||
end, lists:seq(1, MaxInflight)),
|
|
||||||
%% receive acks but do not ack
|
|
||||||
Acks1 = stub_receive(MaxInflight),
|
|
||||||
?assertEqual(MaxInflight, length(Acks1)),
|
|
||||||
%% simulate a disconnect
|
|
||||||
Worker ! {disconnected, self(), test},
|
|
||||||
?SNK_WAIT(disconnected),
|
|
||||||
case ?block_until(#{?snk_kind := connected, inflight := MaxInflight}, 2000, 20) of
|
|
||||||
{ok, _} -> ok;
|
|
||||||
Error -> ct:fail("~p", [Error])
|
|
||||||
end,
|
|
||||||
%% expect worker to retry inflight, so to receive acks again
|
|
||||||
Acks2 = stub_receive(MaxInflight),
|
|
||||||
?assertEqual(MaxInflight, length(Acks2)),
|
|
||||||
lists:foreach(fun({Pid, Ref}) -> Pid ! {batch_ack, Ref} end,
|
|
||||||
lists:reverse(Acks2)),
|
|
||||||
?SNK_WAIT(inflight_drained),
|
|
||||||
?SNK_WAIT(replayq_drained),
|
|
||||||
emqtt:disconnect(ConnPid),
|
|
||||||
ok = emqx_bridge_worker:stop(Worker).
|
|
||||||
|
|
||||||
stub_receive(N) ->
|
|
||||||
stub_receive(N, []).
|
|
||||||
|
|
||||||
stub_receive(0, Acc) -> lists:reverse(Acc);
|
|
||||||
stub_receive(N, Acc) ->
|
|
||||||
receive
|
|
||||||
{stub_message, WorkerPid, BatchRef, _Batch} ->
|
|
||||||
stub_receive(N - 1, [{WorkerPid, BatchRef} | Acc])
|
|
||||||
after
|
|
||||||
5000 ->
|
|
||||||
lists:reverse(Acc)
|
|
||||||
end.
|
|
||||||
|
|
@ -1,135 +0,0 @@
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
||||||
%%
|
|
||||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
%% you may not use this file except in compliance with the License.
|
|
||||||
%% You may obtain a copy of the License at
|
|
||||||
%%
|
|
||||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
%%
|
|
||||||
%% Unless required by applicable law or agreed to in writing, software
|
|
||||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
%% See the License for the specific language governing permissions and
|
|
||||||
%% limitations under the License.
|
|
||||||
%%--------------------------------------------------------------------
|
|
||||||
|
|
||||||
-module(emqx_bridge_worker_tests).
|
|
||||||
|
|
||||||
-include_lib("eunit/include/eunit.hrl").
|
|
||||||
-include_lib("emqx/include/emqx.hrl").
|
|
||||||
-include_lib("emqx/include/emqx_mqtt.hrl").
|
|
||||||
|
|
||||||
-define(BRIDGE_NAME, test).
|
|
||||||
-define(BRIDGE_REG_NAME, emqx_bridge_worker_test).
|
|
||||||
-define(WAIT(PATTERN, TIMEOUT),
|
|
||||||
receive
|
|
||||||
PATTERN ->
|
|
||||||
ok
|
|
||||||
after
|
|
||||||
TIMEOUT ->
|
|
||||||
error(timeout)
|
|
||||||
end).
|
|
||||||
|
|
||||||
%% stub callbacks
|
|
||||||
-export([start/1, send/2, stop/1]).
|
|
||||||
|
|
||||||
start(#{connect_result := Result, test_pid := Pid, test_ref := Ref}) ->
|
|
||||||
case is_pid(Pid) of
|
|
||||||
true -> Pid ! {connection_start_attempt, Ref};
|
|
||||||
false -> ok
|
|
||||||
end,
|
|
||||||
Result.
|
|
||||||
|
|
||||||
send(SendFun, Batch) when is_function(SendFun, 2) ->
|
|
||||||
SendFun(Batch).
|
|
||||||
|
|
||||||
stop(_Pid) -> ok.
|
|
||||||
|
|
||||||
%% bridge worker should retry connecting remote node indefinitely
|
|
||||||
% reconnect_test() ->
|
|
||||||
% emqx_metrics:start_link(),
|
|
||||||
% emqx_bridge_worker:register_metrics(),
|
|
||||||
% Ref = make_ref(),
|
|
||||||
% Config = make_config(Ref, self(), {error, test}),
|
|
||||||
% {ok, Pid} = emqx_bridge_worker:start_link(?BRIDGE_NAME, Config),
|
|
||||||
% %% assert name registered
|
|
||||||
% ?assertEqual(Pid, whereis(?BRIDGE_REG_NAME)),
|
|
||||||
% ?WAIT({connection_start_attempt, Ref}, 1000),
|
|
||||||
% %% expect same message again
|
|
||||||
% ?WAIT({connection_start_attempt, Ref}, 1000),
|
|
||||||
% ok = emqx_bridge_worker:stop(?BRIDGE_REG_NAME),
|
|
||||||
% emqx_metrics:stop(),
|
|
||||||
% ok.
|
|
||||||
|
|
||||||
%% connect first, disconnect, then connect again
|
|
||||||
disturbance_test() ->
|
|
||||||
emqx_metrics:start_link(),
|
|
||||||
emqx_bridge_worker:register_metrics(),
|
|
||||||
Ref = make_ref(),
|
|
||||||
TestPid = self(),
|
|
||||||
Config = make_config(Ref, TestPid, {ok, #{client_pid => TestPid}}),
|
|
||||||
{ok, Pid} = emqx_bridge_worker:start_link(Config#{name => disturbance}),
|
|
||||||
?assertEqual(Pid, whereis(emqx_bridge_worker_disturbance)),
|
|
||||||
?WAIT({connection_start_attempt, Ref}, 1000),
|
|
||||||
Pid ! {disconnected, TestPid, test},
|
|
||||||
?WAIT({connection_start_attempt, Ref}, 1000),
|
|
||||||
emqx_metrics:stop(),
|
|
||||||
ok = emqx_bridge_worker:stop(Pid).
|
|
||||||
|
|
||||||
% % %% buffer should continue taking in messages when disconnected
|
|
||||||
% buffer_when_disconnected_test_() ->
|
|
||||||
% {timeout, 10000, fun test_buffer_when_disconnected/0}.
|
|
||||||
|
|
||||||
% test_buffer_when_disconnected() ->
|
|
||||||
% Ref = make_ref(),
|
|
||||||
% Nums = lists:seq(1, 100),
|
|
||||||
% Sender = spawn_link(fun() -> receive {bridge, Pid} -> sender_loop(Pid, Nums, _Interval = 5) end end),
|
|
||||||
% SenderMref = monitor(process, Sender),
|
|
||||||
% Receiver = spawn_link(fun() -> receive {bridge, Pid} -> receiver_loop(Pid, Nums, _Interval = 1) end end),
|
|
||||||
% ReceiverMref = monitor(process, Receiver),
|
|
||||||
% SendFun = fun(Batch) ->
|
|
||||||
% BatchRef = make_ref(),
|
|
||||||
% Receiver ! {batch, BatchRef, Batch},
|
|
||||||
% {ok, BatchRef}
|
|
||||||
% end,
|
|
||||||
% Config0 = make_config(Ref, false, {ok, #{client_pid => undefined}}),
|
|
||||||
% Config = Config0#{reconnect_delay_ms => 100},
|
|
||||||
% emqx_metrics:start_link(),
|
|
||||||
% emqx_bridge_worker:register_metrics(),
|
|
||||||
% {ok, Pid} = emqx_bridge_worker:start_link(?BRIDGE_NAME, Config),
|
|
||||||
% Sender ! {bridge, Pid},
|
|
||||||
% Receiver ! {bridge, Pid},
|
|
||||||
% ?assertEqual(Pid, whereis(?BRIDGE_REG_NAME)),
|
|
||||||
% Pid ! {disconnected, Ref, test},
|
|
||||||
% ?WAIT({'DOWN', SenderMref, process, Sender, normal}, 5000),
|
|
||||||
% ?WAIT({'DOWN', ReceiverMref, process, Receiver, normal}, 1000),
|
|
||||||
% ok = emqx_bridge_worker:stop(?BRIDGE_REG_NAME),
|
|
||||||
% emqx_metrics:stop().
|
|
||||||
|
|
||||||
manual_start_stop_test() ->
|
|
||||||
emqx_metrics:start_link(),
|
|
||||||
emqx_bridge_worker:register_metrics(),
|
|
||||||
Ref = make_ref(),
|
|
||||||
TestPid = self(),
|
|
||||||
BridgeName = manual_start_stop,
|
|
||||||
Config0 = make_config(Ref, TestPid, {ok, #{client_pid => TestPid}}),
|
|
||||||
Config = Config0#{start_type := manual},
|
|
||||||
{ok, Pid} = emqx_bridge_worker:start_link(Config#{name => BridgeName}),
|
|
||||||
%% call ensure_started again should yeld the same result
|
|
||||||
ok = emqx_bridge_worker:ensure_started(BridgeName),
|
|
||||||
emqx_bridge_worker:ensure_stopped(BridgeName),
|
|
||||||
emqx_metrics:stop(),
|
|
||||||
ok = emqx_bridge_worker:stop(Pid).
|
|
||||||
|
|
||||||
make_config(Ref, TestPid, Result) ->
|
|
||||||
#{
|
|
||||||
start_type => auto,
|
|
||||||
reconnect_interval => 50,
|
|
||||||
config => #{
|
|
||||||
test_pid => TestPid,
|
|
||||||
test_ref => Ref,
|
|
||||||
conn_type => ?MODULE,
|
|
||||||
connect_result => Result
|
|
||||||
}
|
|
||||||
}.
|
|
||||||
|
|
@ -17,7 +17,8 @@
|
||||||
%% By accident, We have always been using the upstream fork due to
|
%% By accident, We have always been using the upstream fork due to
|
||||||
%% eredis_cluster's dependency getting resolved earlier.
|
%% eredis_cluster's dependency getting resolved earlier.
|
||||||
%% Here we pin 1.5.2 to avoid surprises in the future.
|
%% Here we pin 1.5.2 to avoid surprises in the future.
|
||||||
{poolboy, {git, "https://github.com/emqx/poolboy.git", {tag, "1.5.2"}}}
|
{poolboy, {git, "https://github.com/emqx/poolboy.git", {tag, "1.5.2"}}},
|
||||||
|
{emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.4.3"}}}
|
||||||
]}.
|
]}.
|
||||||
|
|
||||||
{shell, [
|
{shell, [
|
||||||
|
|
|
||||||
|
|
@ -13,7 +13,8 @@
|
||||||
epgsql,
|
epgsql,
|
||||||
mysql,
|
mysql,
|
||||||
mongodb,
|
mongodb,
|
||||||
emqx
|
emqx,
|
||||||
|
emqtt
|
||||||
]},
|
]},
|
||||||
{env,[]},
|
{env,[]},
|
||||||
{modules, []},
|
{modules, []},
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@
|
||||||
-export([start/2, stop/1]).
|
-export([start/2, stop/1]).
|
||||||
|
|
||||||
start(_StartType, _StartArgs) ->
|
start(_StartType, _StartArgs) ->
|
||||||
|
emqx_connector_mqtt_worker:register_metrics(),
|
||||||
emqx_connector_sup:start_link().
|
emqx_connector_sup:start_link().
|
||||||
|
|
||||||
stop(_State) ->
|
stop(_State) ->
|
||||||
|
|
|
||||||
|
|
@ -18,6 +18,16 @@
|
||||||
-include_lib("typerefl/include/types.hrl").
|
-include_lib("typerefl/include/types.hrl").
|
||||||
-include_lib("emqx_resource/include/emqx_resource_behaviour.hrl").
|
-include_lib("emqx_resource/include/emqx_resource_behaviour.hrl").
|
||||||
|
|
||||||
|
-behaviour(supervisor).
|
||||||
|
|
||||||
|
%% API and callbacks for supervisor
|
||||||
|
-export([ start_link/0
|
||||||
|
, init/1
|
||||||
|
, create_bridge/1
|
||||||
|
, drop_bridge/1
|
||||||
|
, bridges/0
|
||||||
|
]).
|
||||||
|
|
||||||
%% callbacks of behaviour emqx_resource
|
%% callbacks of behaviour emqx_resource
|
||||||
-export([ on_start/2
|
-export([ on_start/2
|
||||||
, on_stop/2
|
, on_stop/2
|
||||||
|
|
@ -36,55 +46,42 @@ roots() ->
|
||||||
[{config, #{type => hoconsc:ref(?MODULE, "config")}}].
|
[{config, #{type => hoconsc:ref(?MODULE, "config")}}].
|
||||||
|
|
||||||
fields("config") ->
|
fields("config") ->
|
||||||
[ {server, hoconsc:mk(emqx_schema:ip_port(), #{default => "127.0.0.1:1883"})}
|
emqx_connector_mqtt_schema:fields("config").
|
||||||
, {reconnect_interval, hoconsc:mk(emqx_schema:duration_ms(), #{default => "30s"})}
|
|
||||||
, {proto_ver, fun proto_ver/1}
|
|
||||||
, {bridge_mode, hoconsc:mk(boolean(), #{default => true})}
|
|
||||||
, {clientid_prefix, hoconsc:mk(string(), #{default => ""})}
|
|
||||||
, {username, hoconsc:mk(string())}
|
|
||||||
, {password, hoconsc:mk(string())}
|
|
||||||
, {clean_start, hoconsc:mk(boolean(), #{default => true})}
|
|
||||||
, {keepalive, hoconsc:mk(integer(), #{default => 300})}
|
|
||||||
, {retry_interval, hoconsc:mk(emqx_schema:duration_ms(), #{default => "30s"})}
|
|
||||||
, {max_inflight, hoconsc:mk(integer(), #{default => 32})}
|
|
||||||
, {replayq, hoconsc:mk(hoconsc:ref(?MODULE, "replayq"))}
|
|
||||||
, {in, hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, "in")), #{default => []})}
|
|
||||||
, {out, hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, "out")), #{default => []})}
|
|
||||||
] ++ emqx_connector_schema_lib:ssl_fields();
|
|
||||||
|
|
||||||
fields("in") ->
|
%% ===================================================================
|
||||||
[ {subscribe_remote_topic, #{type => binary(), nullable => false}}
|
%% supervisor APIs
|
||||||
, {local_topic, hoconsc:mk(binary(), #{default => <<"${topic}">>})}
|
start_link() ->
|
||||||
, {subscribe_qos, hoconsc:mk(qos(), #{default => 1})}
|
supervisor:start_link({local, ?MODULE}, ?MODULE, []).
|
||||||
] ++ common_inout_confs();
|
|
||||||
|
|
||||||
fields("out") ->
|
init([]) ->
|
||||||
[ {subscribe_local_topic, #{type => binary(), nullable => false}}
|
SupFlag = #{strategy => one_for_one,
|
||||||
, {remote_topic, hoconsc:mk(binary(), #{default => <<"${topic}">>})}
|
intensity => 100,
|
||||||
] ++ common_inout_confs();
|
period => 10},
|
||||||
|
{ok, {SupFlag, []}}.
|
||||||
|
|
||||||
fields("replayq") ->
|
bridge_spec(Config) ->
|
||||||
[ {dir, hoconsc:union([boolean(), string()])}
|
#{id => maps:get(name, Config),
|
||||||
, {seg_bytes, hoconsc:mk(emqx_schema:bytesize(), #{default => "100MB"})}
|
start => {emqx_connector_mqtt_worker, start_link, [Config]},
|
||||||
, {offload, hoconsc:mk(boolean(), #{default => false})}
|
restart => permanent,
|
||||||
, {max_total_bytes, hoconsc:mk(emqx_schema:bytesize(), #{default => "1024MB"})}
|
shutdown => 5000,
|
||||||
].
|
type => worker,
|
||||||
|
modules => [emqx_connector_mqtt_worker]}.
|
||||||
|
|
||||||
common_inout_confs() ->
|
-spec(bridges() -> [{node(), map()}]).
|
||||||
[{id, #{type => binary(), nullable => false}}] ++ publish_confs().
|
bridges() ->
|
||||||
|
[{Name, emqx_connector_mqtt_worker:status(Name)}
|
||||||
|
|| {Name, _Pid, _, _} <- supervisor:which_children(?MODULE)].
|
||||||
|
|
||||||
publish_confs() ->
|
create_bridge(Config) ->
|
||||||
[ {qos, hoconsc:mk(qos(), #{default => <<"${qos}">>})}
|
supervisor:start_child(?MODULE, bridge_spec(Config)).
|
||||||
, {retain, hoconsc:mk(hoconsc:union([boolean(), binary()]), #{default => <<"${retain}">>})}
|
|
||||||
, {payload, hoconsc:mk(binary(), #{default => <<"${payload}">>})}
|
|
||||||
].
|
|
||||||
|
|
||||||
qos() ->
|
drop_bridge(Name) ->
|
||||||
hoconsc:union([typerefl:integer(0), typerefl:integer(1), typerefl:integer(2), binary()]).
|
case supervisor:terminate_child(?MODULE, Name) of
|
||||||
|
ok ->
|
||||||
proto_ver(type) -> hoconsc:enum([v3, v4, v5]);
|
supervisor:delete_child(?MODULE, Name);
|
||||||
proto_ver(default) -> v4;
|
{error, Error} ->
|
||||||
proto_ver(_) -> undefined.
|
{error, Error}
|
||||||
|
end.
|
||||||
|
|
||||||
%% ===================================================================
|
%% ===================================================================
|
||||||
on_start(InstId, Conf) ->
|
on_start(InstId, Conf) ->
|
||||||
|
|
@ -105,7 +102,7 @@ on_start(InstId, Conf) ->
|
||||||
|
|
||||||
on_stop(InstId, #{}) ->
|
on_stop(InstId, #{}) ->
|
||||||
logger:info("stopping mqtt connector: ~p", [InstId]),
|
logger:info("stopping mqtt connector: ~p", [InstId]),
|
||||||
case emqx_bridge_mqtt_sup:drop_bridge(InstId) of
|
case ?MODULE:drop_bridge(InstId) of
|
||||||
ok -> ok;
|
ok -> ok;
|
||||||
{error, not_found} -> ok;
|
{error, not_found} -> ok;
|
||||||
{error, Reason} ->
|
{error, Reason} ->
|
||||||
|
|
@ -124,7 +121,7 @@ on_query(InstId, {publish_to_remote, Msg}, _AfterQuery, _State) ->
|
||||||
logger:debug("publish to remote node, connector: ~p, msg: ~p", [InstId, Msg]).
|
logger:debug("publish to remote node, connector: ~p, msg: ~p", [InstId, Msg]).
|
||||||
|
|
||||||
on_health_check(_InstId, #{sub_bridges := NameList} = State) ->
|
on_health_check(_InstId, #{sub_bridges := NameList} = State) ->
|
||||||
Results = [{Name, emqx_bridge_worker:ping(Name)} || Name <- NameList],
|
Results = [{Name, emqx_connector_mqtt_worker:ping(Name)} || Name <- NameList],
|
||||||
case lists:all(fun({_, pong}) -> true; ({_, _}) -> false end, Results) of
|
case lists:all(fun({_, pong}) -> true; ({_, _}) -> false end, Results) of
|
||||||
true -> {ok, State};
|
true -> {ok, State};
|
||||||
false -> {error, {some_sub_bridge_down, Results}, State}
|
false -> {error, {some_sub_bridge_down, Results}, State}
|
||||||
|
|
@ -155,7 +152,7 @@ create_channel(#{subscribe_local_topic := _, id := BridgeId} = OutConf, NamePref
|
||||||
subscriptions => undefined, forwards => OutConf}).
|
subscriptions => undefined, forwards => OutConf}).
|
||||||
|
|
||||||
create_sub_bridge(#{name := Name} = Conf) ->
|
create_sub_bridge(#{name := Name} = Conf) ->
|
||||||
case emqx_bridge_mqtt_sup:create_bridge(Conf) of
|
case ?MODULE:create_bridge(Conf) of
|
||||||
{ok, _Pid} ->
|
{ok, _Pid} ->
|
||||||
start_sub_bridge(Name);
|
start_sub_bridge(Name);
|
||||||
{error, {already_started, _Pid}} ->
|
{error, {already_started, _Pid}} ->
|
||||||
|
|
@ -165,7 +162,7 @@ create_sub_bridge(#{name := Name} = Conf) ->
|
||||||
end.
|
end.
|
||||||
|
|
||||||
start_sub_bridge(Name) ->
|
start_sub_bridge(Name) ->
|
||||||
case emqx_bridge_worker:ensure_started(Name) of
|
case emqx_connector_mqtt_worker:ensure_started(Name) of
|
||||||
ok -> {ok, Name};
|
ok -> {ok, Name};
|
||||||
{error, Reason} -> {error, Reason}
|
{error, Reason} -> {error, Reason}
|
||||||
end.
|
end.
|
||||||
|
|
|
||||||
|
|
@ -28,9 +28,19 @@ start_link() ->
|
||||||
|
|
||||||
init([]) ->
|
init([]) ->
|
||||||
SupFlags = #{strategy => one_for_all,
|
SupFlags = #{strategy => one_for_all,
|
||||||
intensity => 0,
|
intensity => 5,
|
||||||
period => 1},
|
period => 20},
|
||||||
ChildSpecs = [],
|
ChildSpecs = [
|
||||||
|
child_spec(emqx_connector_mqtt)
|
||||||
|
],
|
||||||
{ok, {SupFlags, ChildSpecs}}.
|
{ok, {SupFlags, ChildSpecs}}.
|
||||||
|
|
||||||
|
child_spec(Mod) ->
|
||||||
|
#{id => Mod,
|
||||||
|
start => {Mod, start_link, []},
|
||||||
|
restart => permanent,
|
||||||
|
shutdown => 3000,
|
||||||
|
type => supervisor,
|
||||||
|
modules => [Mod]}.
|
||||||
|
|
||||||
%% internal functions
|
%% internal functions
|
||||||
|
|
|
||||||
|
|
@ -16,7 +16,7 @@
|
||||||
|
|
||||||
%% @doc This module implements EMQX Bridge transport layer on top of MQTT protocol
|
%% @doc This module implements EMQX Bridge transport layer on top of MQTT protocol
|
||||||
|
|
||||||
-module(emqx_bridge_mqtt).
|
-module(emqx_connector_mqtt_mod).
|
||||||
|
|
||||||
-export([ start/1
|
-export([ start/1
|
||||||
, send/2
|
, send/2
|
||||||
|
|
@ -51,7 +51,7 @@ start(Config) ->
|
||||||
{Host, Port} = maps:get(server, Config),
|
{Host, Port} = maps:get(server, Config),
|
||||||
Mountpoint = maps:get(receive_mountpoint, Config, undefined),
|
Mountpoint = maps:get(receive_mountpoint, Config, undefined),
|
||||||
Subscriptions = maps:get(subscriptions, Config),
|
Subscriptions = maps:get(subscriptions, Config),
|
||||||
Vars = emqx_bridge_msg:make_pub_vars(Mountpoint, Subscriptions),
|
Vars = emqx_connector_mqtt_msg:make_pub_vars(Mountpoint, Subscriptions),
|
||||||
Handlers = make_hdlr(Parent, Vars),
|
Handlers = make_hdlr(Parent, Vars),
|
||||||
Config1 = Config#{
|
Config1 = Config#{
|
||||||
msg_handler => Handlers,
|
msg_handler => Handlers,
|
||||||
|
|
@ -161,7 +161,7 @@ handle_publish(Msg, undefined) ->
|
||||||
?LOG(error, "cannot publish to local broker as 'bridge.mqtt.<name>.in' not configured, msg: ~p", [Msg]);
|
?LOG(error, "cannot publish to local broker as 'bridge.mqtt.<name>.in' not configured, msg: ~p", [Msg]);
|
||||||
handle_publish(Msg, Vars) ->
|
handle_publish(Msg, Vars) ->
|
||||||
?LOG(debug, "publish to local broker, msg: ~p, vars: ~p", [Msg, Vars]),
|
?LOG(debug, "publish to local broker, msg: ~p, vars: ~p", [Msg, Vars]),
|
||||||
emqx_broker:publish(emqx_bridge_msg:to_broker_msg(Msg, Vars)).
|
emqx_broker:publish(emqx_connector_mqtt_msg:to_broker_msg(Msg, Vars)).
|
||||||
|
|
||||||
handle_disconnected(Reason, Parent) ->
|
handle_disconnected(Reason, Parent) ->
|
||||||
Parent ! {disconnected, self(), Reason}.
|
Parent ! {disconnected, self(), Reason}.
|
||||||
|
|
@ -14,7 +14,7 @@
|
||||||
%% limitations under the License.
|
%% limitations under the License.
|
||||||
%%--------------------------------------------------------------------
|
%%--------------------------------------------------------------------
|
||||||
|
|
||||||
-module(emqx_bridge_msg).
|
-module(emqx_connector_mqtt_msg).
|
||||||
|
|
||||||
-export([ to_binary/1
|
-export([ to_binary/1
|
||||||
, from_binary/1
|
, from_binary/1
|
||||||
|
|
@ -28,7 +28,6 @@
|
||||||
|
|
||||||
-include_lib("emqx/include/emqx.hrl").
|
-include_lib("emqx/include/emqx.hrl").
|
||||||
|
|
||||||
-include_lib("emqx_bridge_mqtt/include/emqx_bridge_mqtt.hrl").
|
|
||||||
-include_lib("emqtt/include/emqtt.hrl").
|
-include_lib("emqtt/include/emqtt.hrl").
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -56,13 +55,13 @@ make_pub_vars(Mountpoint, #{payload := _, qos := _, retain := _, local_topic :=
|
||||||
%% Shame that we have to know the callback module here
|
%% Shame that we have to know the callback module here
|
||||||
%% would be great if we can get rid of #mqtt_msg{} record
|
%% would be great if we can get rid of #mqtt_msg{} record
|
||||||
%% and use #message{} in all places.
|
%% and use #message{} in all places.
|
||||||
-spec to_remote_msg(emqx_bridge_rpc | emqx_bridge_worker, msg(), variables())
|
-spec to_remote_msg(emqx_bridge_rpc | emqx_connector_mqtt_mod, msg(), variables())
|
||||||
-> exp_msg().
|
-> exp_msg().
|
||||||
to_remote_msg(emqx_bridge_mqtt, #message{flags = Flags0} = Msg, Vars) ->
|
to_remote_msg(emqx_connector_mqtt_mod, #message{flags = Flags0} = Msg, Vars) ->
|
||||||
Retain0 = maps:get(retain, Flags0, false),
|
Retain0 = maps:get(retain, Flags0, false),
|
||||||
MapMsg = maps:put(retain, Retain0, emqx_message:to_map(Msg)),
|
MapMsg = maps:put(retain, Retain0, emqx_message:to_map(Msg)),
|
||||||
to_remote_msg(emqx_bridge_mqtt, MapMsg, Vars);
|
to_remote_msg(emqx_connector_mqtt_mod, MapMsg, Vars);
|
||||||
to_remote_msg(emqx_bridge_mqtt, MapMsg, #{topic := TopicToken, payload := PayloadToken,
|
to_remote_msg(emqx_connector_mqtt_mod, MapMsg, #{topic := TopicToken, payload := PayloadToken,
|
||||||
qos := QoSToken, retain := RetainToken, mountpoint := Mountpoint}) when is_map(MapMsg) ->
|
qos := QoSToken, retain := RetainToken, mountpoint := Mountpoint}) when is_map(MapMsg) ->
|
||||||
Topic = replace_vars_in_str(TopicToken, MapMsg),
|
Topic = replace_vars_in_str(TopicToken, MapMsg),
|
||||||
Payload = replace_vars_in_str(PayloadToken, MapMsg),
|
Payload = replace_vars_in_str(PayloadToken, MapMsg),
|
||||||
|
|
@ -0,0 +1,78 @@
|
||||||
|
%%--------------------------------------------------------------------
|
||||||
|
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||||
|
%%
|
||||||
|
%% Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
%% you may not use this file except in compliance with the License.
|
||||||
|
%% You may obtain a copy of the License at
|
||||||
|
%%
|
||||||
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
%%
|
||||||
|
%% Unless required by applicable law or agreed to in writing, software
|
||||||
|
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
%% See the License for the specific language governing permissions and
|
||||||
|
%% limitations under the License.
|
||||||
|
%%--------------------------------------------------------------------
|
||||||
|
|
||||||
|
-module(emqx_connector_mqtt_schema).
|
||||||
|
|
||||||
|
-include_lib("typerefl/include/types.hrl").
|
||||||
|
|
||||||
|
-behaviour(hocon_schema).
|
||||||
|
|
||||||
|
-export([ roots/0
|
||||||
|
, fields/1]).
|
||||||
|
|
||||||
|
roots() ->
|
||||||
|
[{config, #{type => hoconsc:ref(?MODULE, "config")}}].
|
||||||
|
|
||||||
|
fields("config") ->
|
||||||
|
[ {server, hoconsc:mk(emqx_schema:ip_port(), #{default => "127.0.0.1:1883"})}
|
||||||
|
, {reconnect_interval, hoconsc:mk(emqx_schema:duration_ms(), #{default => "30s"})}
|
||||||
|
, {proto_ver, fun proto_ver/1}
|
||||||
|
, {bridge_mode, hoconsc:mk(boolean(), #{default => true})}
|
||||||
|
, {clientid_prefix, hoconsc:mk(string(), #{default => ""})}
|
||||||
|
, {username, hoconsc:mk(string())}
|
||||||
|
, {password, hoconsc:mk(string())}
|
||||||
|
, {clean_start, hoconsc:mk(boolean(), #{default => true})}
|
||||||
|
, {keepalive, hoconsc:mk(integer(), #{default => 300})}
|
||||||
|
, {retry_interval, hoconsc:mk(emqx_schema:duration_ms(), #{default => "30s"})}
|
||||||
|
, {max_inflight, hoconsc:mk(integer(), #{default => 32})}
|
||||||
|
, {replayq, hoconsc:mk(hoconsc:ref(?MODULE, "replayq"))}
|
||||||
|
, {in, hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, "in")), #{default => []})}
|
||||||
|
, {out, hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, "out")), #{default => []})}
|
||||||
|
] ++ emqx_connector_schema_lib:ssl_fields();
|
||||||
|
|
||||||
|
fields("in") ->
|
||||||
|
[ {subscribe_remote_topic, #{type => binary(), nullable => false}}
|
||||||
|
, {local_topic, hoconsc:mk(binary(), #{default => <<"${topic}">>})}
|
||||||
|
, {subscribe_qos, hoconsc:mk(qos(), #{default => 1})}
|
||||||
|
] ++ common_inout_confs();
|
||||||
|
|
||||||
|
fields("out") ->
|
||||||
|
[ {subscribe_local_topic, #{type => binary(), nullable => false}}
|
||||||
|
, {remote_topic, hoconsc:mk(binary(), #{default => <<"${topic}">>})}
|
||||||
|
] ++ common_inout_confs();
|
||||||
|
|
||||||
|
fields("replayq") ->
|
||||||
|
[ {dir, hoconsc:union([boolean(), string()])}
|
||||||
|
, {seg_bytes, hoconsc:mk(emqx_schema:bytesize(), #{default => "100MB"})}
|
||||||
|
, {offload, hoconsc:mk(boolean(), #{default => false})}
|
||||||
|
, {max_total_bytes, hoconsc:mk(emqx_schema:bytesize(), #{default => "1024MB"})}
|
||||||
|
].
|
||||||
|
|
||||||
|
common_inout_confs() ->
|
||||||
|
[{id, #{type => binary(), nullable => false}}] ++ publish_confs().
|
||||||
|
|
||||||
|
publish_confs() ->
|
||||||
|
[ {qos, hoconsc:mk(qos(), #{default => <<"${qos}">>})}
|
||||||
|
, {retain, hoconsc:mk(hoconsc:union([boolean(), binary()]), #{default => <<"${retain}">>})}
|
||||||
|
, {payload, hoconsc:mk(binary(), #{default => <<"${payload}">>})}
|
||||||
|
].
|
||||||
|
|
||||||
|
qos() ->
|
||||||
|
hoconsc:union([typerefl:integer(0), typerefl:integer(1), typerefl:integer(2), binary()]).
|
||||||
|
|
||||||
|
proto_ver(type) -> hoconsc:enum([v3, v4, v5]);
|
||||||
|
proto_ver(default) -> v4;
|
||||||
|
proto_ver(_) -> undefined.
|
||||||
|
|
@ -19,7 +19,7 @@
|
||||||
%% to remote MQTT node/cluster via `connection' transport layer.
|
%% to remote MQTT node/cluster via `connection' transport layer.
|
||||||
%% In case `REMOTE' is also an EMQX node, `connection' is recommended to be
|
%% In case `REMOTE' is also an EMQX node, `connection' is recommended to be
|
||||||
%% the `gen_rpc' based implementation `emqx_bridge_rpc'. Otherwise `connection'
|
%% the `gen_rpc' based implementation `emqx_bridge_rpc'. Otherwise `connection'
|
||||||
%% has to be `emqx_bridge_mqtt'.
|
%% has to be `emqx_connector_mqtt_mod'.
|
||||||
%%
|
%%
|
||||||
%% ```
|
%% ```
|
||||||
%% +------+ +--------+
|
%% +------+ +--------+
|
||||||
|
|
@ -59,7 +59,7 @@
|
||||||
%% NOTES:
|
%% NOTES:
|
||||||
%% * Local messages are all normalised to QoS-1 when exporting to remote
|
%% * Local messages are all normalised to QoS-1 when exporting to remote
|
||||||
|
|
||||||
-module(emqx_bridge_worker).
|
-module(emqx_connector_mqtt_worker).
|
||||||
-behaviour(gen_statem).
|
-behaviour(gen_statem).
|
||||||
|
|
||||||
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
|
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
|
||||||
|
|
@ -106,7 +106,7 @@
|
||||||
-type id() :: atom() | string() | pid().
|
-type id() :: atom() | string() | pid().
|
||||||
-type qos() :: emqx_mqtt_types:qos().
|
-type qos() :: emqx_mqtt_types:qos().
|
||||||
-type config() :: map().
|
-type config() :: map().
|
||||||
-type batch() :: [emqx_bridge_msg:exp_msg()].
|
-type batch() :: [emqx_connector_mqtt_msg:exp_msg()].
|
||||||
-type ack_ref() :: term().
|
-type ack_ref() :: term().
|
||||||
-type topic() :: emqx_topic:topic().
|
-type topic() :: emqx_topic:topic().
|
||||||
|
|
||||||
|
|
@ -222,7 +222,7 @@ open_replayq(Name, QCfg) ->
|
||||||
false -> #{dir => filename:join([Dir, node(), Name]),
|
false -> #{dir => filename:join([Dir, node(), Name]),
|
||||||
seg_bytes => SegBytes, max_total_size => MaxTotalSize}
|
seg_bytes => SegBytes, max_total_size => MaxTotalSize}
|
||||||
end,
|
end,
|
||||||
replayq:open(QueueConfig#{sizer => fun emqx_bridge_msg:estimate_size/1,
|
replayq:open(QueueConfig#{sizer => fun emqx_connector_mqtt_msg:estimate_size/1,
|
||||||
marshaller => fun ?MODULE:msg_marshaller/1}).
|
marshaller => fun ?MODULE:msg_marshaller/1}).
|
||||||
|
|
||||||
pre_process_opts(#{subscriptions := InConf, forwards := OutConf} = ConnectOpts) ->
|
pre_process_opts(#{subscriptions := InConf, forwards := OutConf} = ConnectOpts) ->
|
||||||
|
|
@ -412,10 +412,10 @@ do_send(#{inflight := Inflight,
|
||||||
mountpoint := Mountpoint,
|
mountpoint := Mountpoint,
|
||||||
connect_opts := #{forwards := Forwards},
|
connect_opts := #{forwards := Forwards},
|
||||||
if_record_metrics := IfRecordMetrics} = State, QAckRef, [_ | _] = Batch) ->
|
if_record_metrics := IfRecordMetrics} = State, QAckRef, [_ | _] = Batch) ->
|
||||||
Vars = emqx_bridge_msg:make_pub_vars(Mountpoint, Forwards),
|
Vars = emqx_connector_mqtt_msg:make_pub_vars(Mountpoint, Forwards),
|
||||||
ExportMsg = fun(Message) ->
|
ExportMsg = fun(Message) ->
|
||||||
bridges_metrics_inc(IfRecordMetrics, 'bridge.mqtt.message_sent'),
|
bridges_metrics_inc(IfRecordMetrics, 'bridge.mqtt.message_sent'),
|
||||||
emqx_bridge_msg:to_remote_msg(Module, Message, Vars)
|
emqx_connector_mqtt_msg:to_remote_msg(Module, Message, Vars)
|
||||||
end,
|
end,
|
||||||
?LOG(debug, "publish to remote broker, msg: ~p, vars: ~p", [Batch, Vars]),
|
?LOG(debug, "publish to remote broker, msg: ~p, vars: ~p", [Batch, Vars]),
|
||||||
case Module:send(Connection, [ExportMsg(M) || M <- Batch]) of
|
case Module:send(Connection, [ExportMsg(M) || M <- Batch]) of
|
||||||
|
|
@ -501,8 +501,8 @@ disconnect(State) ->
|
||||||
State.
|
State.
|
||||||
|
|
||||||
%% Called only when replayq needs to dump it to disk.
|
%% Called only when replayq needs to dump it to disk.
|
||||||
msg_marshaller(Bin) when is_binary(Bin) -> emqx_bridge_msg:from_binary(Bin);
|
msg_marshaller(Bin) when is_binary(Bin) -> emqx_connector_mqtt_msg:from_binary(Bin);
|
||||||
msg_marshaller(Msg) -> emqx_bridge_msg:to_binary(Msg).
|
msg_marshaller(Msg) -> emqx_connector_mqtt_msg:to_binary(Msg).
|
||||||
|
|
||||||
format_mountpoint(undefined) ->
|
format_mountpoint(undefined) ->
|
||||||
undefined;
|
undefined;
|
||||||
|
|
@ -541,7 +541,7 @@ is_sensitive(_) -> false.
|
||||||
conn_type(rpc) ->
|
conn_type(rpc) ->
|
||||||
emqx_bridge_rpc;
|
emqx_bridge_rpc;
|
||||||
conn_type(mqtt) ->
|
conn_type(mqtt) ->
|
||||||
emqx_bridge_mqtt;
|
emqx_connector_mqtt_mod;
|
||||||
conn_type(Mod) when is_atom(Mod) ->
|
conn_type(Mod) when is_atom(Mod) ->
|
||||||
Mod.
|
Mod.
|
||||||
|
|
||||||
|
|
@ -37,11 +37,11 @@ send_and_ack_test() ->
|
||||||
try
|
try
|
||||||
Max = 1,
|
Max = 1,
|
||||||
Batch = lists:seq(1, Max),
|
Batch = lists:seq(1, Max),
|
||||||
{ok, Conn} = emqx_bridge_mqtt:start(#{address => "127.0.0.1:1883"}),
|
{ok, Conn} = emqx_connector_mqtt_mod:start(#{address => "127.0.0.1:1883"}),
|
||||||
% %% return last packet id as batch reference
|
% %% return last packet id as batch reference
|
||||||
{ok, _AckRef} = emqx_bridge_mqtt:send(Conn, Batch),
|
{ok, _AckRef} = emqx_connector_mqtt_mod:send(Conn, Batch),
|
||||||
|
|
||||||
ok = emqx_bridge_mqtt:stop(Conn)
|
ok = emqx_connector_mqtt_mod:stop(Conn)
|
||||||
after
|
after
|
||||||
meck:unload(emqtt)
|
meck:unload(emqtt)
|
||||||
end.
|
end.
|
||||||
|
|
@ -48,7 +48,6 @@
|
||||||
, emqx_statsd_schema
|
, emqx_statsd_schema
|
||||||
, emqx_authz_schema
|
, emqx_authz_schema
|
||||||
, emqx_auto_subscribe_schema
|
, emqx_auto_subscribe_schema
|
||||||
, emqx_bridge_mqtt_schema
|
|
||||||
, emqx_modules_schema
|
, emqx_modules_schema
|
||||||
, emqx_dashboard_schema
|
, emqx_dashboard_schema
|
||||||
, emqx_gateway_schema
|
, emqx_gateway_schema
|
||||||
|
|
|
||||||
|
|
@ -411,7 +411,7 @@ test_resource_status(PoolName) ->
|
||||||
IsConnected = fun(Worker) ->
|
IsConnected = fun(Worker) ->
|
||||||
case ecpool_worker:client(Worker) of
|
case ecpool_worker:client(Worker) of
|
||||||
{ok, Bridge} ->
|
{ok, Bridge} ->
|
||||||
try emqx_bridge_worker:status(Bridge) of
|
try emqx_connector_mqtt_worker:status(Bridge) of
|
||||||
connected -> true;
|
connected -> true;
|
||||||
_ -> false
|
_ -> false
|
||||||
catch _Error:_Reason ->
|
catch _Error:_Reason ->
|
||||||
|
|
@ -524,7 +524,7 @@ connect(Options = #{disk_cache := DiskCache, ecpool_worker_id := Id, pool_name :
|
||||||
end
|
end
|
||||||
end,
|
end,
|
||||||
Options2 = maps:without([ecpool_worker_id, pool_name, append], Options1),
|
Options2 = maps:without([ecpool_worker_id, pool_name, append], Options1),
|
||||||
emqx_bridge_worker:start_link(Options2#{name => name(Pool, Id)}).
|
emqx_connector_mqtt_worker:start_link(Options2#{name => name(Pool, Id)}).
|
||||||
name(Pool, Id) ->
|
name(Pool, Id) ->
|
||||||
list_to_atom(atom_to_list(Pool) ++ ":" ++ integer_to_list(Id)).
|
list_to_atom(atom_to_list(Pool) ++ ":" ++ integer_to_list(Id)).
|
||||||
pool_name(ResId) ->
|
pool_name(ResId) ->
|
||||||
|
|
|
||||||
|
|
@ -272,7 +272,6 @@ relx_apps(ReleaseType) ->
|
||||||
, emqx_bridge
|
, emqx_bridge
|
||||||
, emqx_rule_engine
|
, emqx_rule_engine
|
||||||
, emqx_rule_actions
|
, emqx_rule_actions
|
||||||
, emqx_bridge_mqtt
|
|
||||||
, emqx_modules
|
, emqx_modules
|
||||||
, emqx_management
|
, emqx_management
|
||||||
, emqx_dashboard
|
, emqx_dashboard
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue