feat(mqtt-proxy): support using route's upstream

apisix/stream/plugins/mqtt-proxy.lua

@@ -41,7 +41,7 @@ local schema = {
             },
         }
     },
-    required = {"protocol_name", "protocol_level", "upstream"},
+    required = {"protocol_name", "protocol_level"},
 }
 
 
@@ -164,6 +164,10 @@ function _M.preread(conf, ctx)
 
     core.log.info("mqtt client id: ", res.client_id)
 
+    if not conf.upstream then
+        return
+    end
+
     local host = conf.upstream.host
     if not host then
         host = conf.upstream.ip

docs/en/latest/plugins/mqtt-proxy.md

@@ -41,6 +41,7 @@ And this plugin both support MQTT protocol [3.1.*](http://docs.oasis-open.org/mq
 | -------------- | ------- | ----------- | ------- | ----- | -------------------------------------------------------------------------------------- |
 | protocol_name  | string  | required    |         |       | Name of protocol, should be `MQTT` in normal.                                          |
 | protocol_level | integer | required    |         |       | Level of protocol, it should be `4` for MQTT `3.1.*`. it should be `5` for MQTT `5.0`. |
+| upstream       | object  | deprecated  |         |       | Use separate upstream in the route instead.                                            |
 | upstream.host  | string  | required    |         |       | the IP or host of upstream, will forward current request to.                           |
 | upstream.ip    | string  | deprecated  |         |       | Use "host" instead. IP address of upstream, will forward current request to.|
 | upstream.port  | number  | required    |         |       | Port of upstream, will forward current request to.                                     |
@@ -73,12 +74,16 @@ curl http://127.0.0.1:9080/apisix/admin/stream_routes/1 -H 'X-API-KEY: edd1c9f03
     "plugins": {
         "mqtt-proxy": {
             "protocol_name": "MQTT",
-            "protocol_level": 4,
-            "upstream": {
-                "host": "127.0.0.1",
-                "port": 1980
-            }
+            "protocol_level": 4
         }
+    },
+    "upstream": {
+        "type": "roundrobin",
+        "nodes": [{
+            "host": "127.0.0.1",
+            "port": 1980,
+            "weight": 1
+        }]
     }
 }'
 ```

docs/zh/latest/plugins/mqtt-proxy.md

@@ -40,6 +40,7 @@ title: mqtt-proxy
 | -------------- | ------- | ------ | ------ | ------ | ------------------------------------------------------ |
 | protocol_name  | string  | 必须   |        |        | 协议名称，正常情况下应为“ MQTT”                        |
 | protocol_level | integer | 必须   |        |        | 协议级别，MQTT `3.1.*` 应为 `4` ，MQTT `5.0` 应是`5`。 |
+| upstream       | object  | 废弃   |         |       | 推荐改用 route 上配置的上游信息                                            |
 | upstream.host  | string  | 必须   |        |        | 将当前请求转发到的上游的 IP 地址或域名                  |
 | upstream.ip    | string  | 废弃   |        |        | 推荐使用“host”代替。将当前请求转发到的上游的 IP 地址                       |
 | upstream.port  | number  | 必须   |        |        | 将当前请求转发到的上游的端口                           |
@@ -71,12 +72,16 @@ curl http://127.0.0.1:9080/apisix/admin/stream_routes/1 -H 'X-API-KEY: edd1c9f03
     "plugins": {
         "mqtt-proxy": {
             "protocol_name": "MQTT",
-            "protocol_level": 4,
-            "upstream": {
-                "host": "127.0.0.1",
-                "port": 1980
-            }
+            "protocol_level": 4
         }
+    },
+    "upstream": {
+        "type": "roundrobin",
+        "nodes": [{
+            "host": "127.0.0.1",
+            "port": 1980,
+            "weight": 1
+        }]
     }
 }'
 ```

t/stream-plugin/mqtt-proxy.t

@@ -265,3 +265,56 @@ passed
 --- error_log
 failed to parse domain: loc, error:
 --- timeout: 10
+
+
+
+=== TEST 11: set route with upstream
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local code, body = t('/apisix/admin/stream_routes/1',
+                ngx.HTTP_PUT,
+                [[{
+                    "remote_addr": "127.0.0.1",
+                    "server_port": 1985,
+                    "plugins": {
+                        "mqtt-proxy": {
+                            "protocol_name": "MQTT",
+                            "protocol_level": 4
+                        }
+                    },
+                    "upstream": {
+                        "type": "roundrobin",
+                        "nodes": [{
+                            "host": "127.0.0.1",
+                            "port": 1995,
+                            "weight": 1
+                        }]
+                    }
+                }]]
+                )
+
+            if code >= 300 then
+                ngx.status = code
+            end
+            ngx.say(body)
+        }
+    }
+--- request
+GET /t
+--- response_body
+passed
+--- no_error_log
+[error]
+
+
+
+=== TEST 12: hit route
+--- stream_enable
+--- stream_request eval
+"\x10\x0f\x00\x04\x4d\x51\x54\x54\x04\x02\x00\x3c\x00\x03\x66\x6f\x6f"
+--- stream_response
+hello world
+--- no_error_log
+[error]