fix: run init_worker of apisix.admin module in stream subsystem

apisix/init.lua

@@ -1173,6 +1173,9 @@ function _M.stream_init_worker()
 
     require("apisix.events").init_worker()
 
+    -- for admin api of standalone mode, we need to startup background timer and patch schema etc.
+    require("apisix.admin.init").init_worker()
+
     local discovery = require("apisix.discovery.init").discovery
     if discovery and discovery.init_worker then
         discovery.init_worker()

t/admin/standalone.t

@@ -273,7 +273,7 @@ X-Digest: t13
 === TEST 14: configure stream route
 --- request
 PUT /apisix/admin/configs
-{"stream_routes":[{"server_addr":"127.0.0.1","server_port":1985,"id":1,"upstream":{"nodes":{"127.0.0.1:1995":1},"type":"roundrobin"}}]}
+{"stream_routes":[{"modifiedIndex": 1, "server_addr":"127.0.0.1","server_port":1985,"id":1,"upstream":{"nodes":{"127.0.0.1:1995":1},"type":"roundrobin"}}]}
 --- more_headers
 X-API-KEY: edd1c9f034335f136f87ad84b625c8f1
 X-Digest: t14
@@ -286,30 +286,69 @@ X-Digest: t14
 location /stream_request {
     content_by_lua_block {
         ngx.sleep(1)  -- wait for the stream route to take effect
-        local sock, err = ngx.socket.tcp()
-        assert(sock, err)
 
-        local ok, err = sock:connect("127.0.0.1", 1985)
-        if not ok then
-            ngx.say("connect to stream server error: ", err)
-            return
-        end
-        local bytes, err = sock:send("mmm")
-        if not bytes then
-            ngx.say("send stream request error: ", err)
-            return
+        local tcp_request = function(host, port)
+            local sock, err = ngx.socket.tcp()
+            assert(sock, err)
+
+            local ok, err = sock:connect(host, port)
+            if not ok then
+                ngx.say("connect to stream server error: ", err)
+                return
+            end
+            local bytes, err = sock:send("mmm")
+            if not bytes then
+                ngx.say("send stream request error: ", err)
+                return
+            end
+
+            local data, err = sock:receive("*a")
+            if not data then
+                sock:close()
+                ngx.say("receive stream response error: ", err)
+                return
+            end
+            sock:close()
+            ngx.print(data)
         end
 
-        local data, err = sock:receive("*a")
-        if not data then
-            sock:close()
-            ngx.say("receive stream response error: ", err)
+        tcp_request("127.0.0.1", 1985)
+
+        -- update the stream route in runtime to confirm the new stream route takes effect
+        local t = require("lib.test_admin").test
+        local code, body = t('/apisix/admin/configs',
+            ngx.HTTP_PUT,
+            [[{
+              "stream_routes": [
+                {
+                  "modifiedIndex": 2,
+                  "server_addr": "127.0.0.2",
+                  "server_port": 1985,
+                  "id": 1,
+                  "upstream": {
+                    "nodes": {
+                      "127.0.0.1:1995": 1
+                    },
+                    "type": "roundrobin"
+                  }
+                }
+              ]
+            }]], nil, { ["X-API-KEY"] = "edd1c9f034335f136f87ad84b625c8f1", ["X-Digest"] = "t15"}
+        )
+        if code ~= 202 then
+            ngx.print("failed to update stream route, code: ", code, ", body: ", body)
             return
         end
-        ngx.print(data)
+
+        ngx.sleep(1)
+
+        tcp_request("127.0.0.1", 1985)
+        tcp_request("127.0.0.2", 1985)
     }
 }
 --- request
 GET /stream_request
 --- response_body
 hello world
+receive stream response error: connection reset by peer
+hello world