test v0

kong/clustering/rpc/socket.lua

@@ -172,7 +172,7 @@ function _M:process_rpc_msg(payload, collection)
       -- TODO: async call by using a new manager of timer
       -- collection is not nil, it means it is a batch call
       -- we should call sync function
-      res, err = _M._dispatch(nil, self, dispatch_cb, payload, collection)
+      _M._dispatch(nil, self, dispatch_cb, payload, collection)
 
     else
 
@@ -181,17 +181,17 @@ function _M:process_rpc_msg(payload, collection)
       local name = string_format("JSON-RPC callback for node_id: %s, id: %d, method: %s",
                                  self.node_id, payload_id or 0, payload_method)
       res, err = kong.timer:named_at(name, 0, _M._dispatch, self, dispatch_cb, payload)
-    end
 
-    if not res and payload_id then
-      local reso, erro = self:push_response(new_error(payload_id, jsonrpc.INTERNAL_ERROR),
-                                            "unable to send \"INTERNAL_ERROR\" error back to client: ",
-                                            collection)
-      if not reso then
-        return nil, erro
-      end
+      if not res and payload_id then
+        local reso, erro = self:push_response(new_error(payload_id, jsonrpc.INTERNAL_ERROR),
+                                              "unable to send \"INTERNAL_ERROR\" error back to client: ",
+                                              collection)
+        if not reso then
+          return nil, erro
+        end
 
-      return nil, "unable to dispatch JSON-RPC callback: " .. err
+        return nil, "unable to dispatch JSON-RPC callback: " .. err
+      end
     end
 
   else

spec/02-integration/18-hybrid_rpc/06-batch-rpc_spec.lua

@@ -0,0 +1,90 @@
+local helpers = require "spec.helpers"
+local cjson = require("cjson.safe")
+
+-- register a test rpc service in custom plugin rpc-batch-test
+for _, strategy in helpers.each_strategy() do
+  describe("Hybrid Mode RPC #" .. strategy, function()
+
+    lazy_setup(function()
+      helpers.get_db_utils(strategy, { "routes", "services" })
+
+      assert(helpers.start_kong({
+        role = "control_plane",
+        cluster_cert = "spec/fixtures/kong_clustering.crt",
+        cluster_cert_key = "spec/fixtures/kong_clustering.key",
+        database = strategy,
+        cluster_listen = "127.0.0.1:9005",
+        nginx_conf = "spec/fixtures/custom_nginx.template",
+        cluster_rpc = "on",
+        plugins = "bundled,rpc-batch-test",
+        cluster_rpc_sync = "off",
+      }))
+
+      assert(helpers.start_kong({
+        role = "data_plane",
+        database = "off",
+        prefix = "servroot2",
+        cluster_cert = "spec/fixtures/kong_clustering.crt",
+        cluster_cert_key = "spec/fixtures/kong_clustering.key",
+        cluster_control_plane = "127.0.0.1:9005",
+        proxy_listen = "0.0.0.0:9002",
+        nginx_conf = "spec/fixtures/custom_nginx.template",
+        cluster_rpc = "on",
+        plugins = "bundled,rpc-batch-test",
+        cluster_rpc_sync = "off",
+      }))
+    end)
+
+    lazy_teardown(function()
+      helpers.stop_kong("servroot2")
+      helpers.stop_kong()
+    end)
+
+    describe("batch works", function()
+      it("DP calls CP via batching ", function()
+        local admin_client = helpers.admin_client(10000)
+        finally(function()
+          admin_client:close()
+        end)
+
+        local res = assert(admin_client:post("/services", {
+          body = { name = "mockbin-service", url = "https://127.0.0.1:15556/request", },
+          headers = {["Content-Type"] = "application/json"}
+        }))
+        assert.res_status(201, res)
+
+        res = assert(admin_client:post("/services/mockbin-service/routes", {
+          body = { paths = { "/" }, },
+          headers = {["Content-Type"] = "application/json"}
+        }))
+        local body = assert.res_status(201, res)
+        local json = cjson.decode(body)
+        local route_id = json.id
+
+        -- add a plugin for route
+        res = assert(admin_client:post("/routes/" .. route_id .. "/plugins", {
+          body = { name = "rpc-batch-test" },
+          headers = {["Content-Type"] = "application/json"}
+        }))
+        assert.res_status(201, res)
+
+        helpers.wait_until(function()
+          local proxy_client = helpers.http_client("127.0.0.1", 9002)
+
+          res = proxy_client:send({
+            method  = "GET",
+            path    = "/",
+          })
+
+          local status = res and res.status
+          proxy_client:close()
+          if status == 200 then
+            return true
+          end
+        end, 10)
+
+        assert.logfile().has.line("kong.test.batch called", true)
+      end)
+    end)
+  end)
+end -- for _, strategy

spec/fixtures/custom_plugins/kong/plugins/rpc-batch-test/handler.lua

@@ -0,0 +1,27 @@
+local RpcBatchTestHandler = {
+  VERSION = "1.0",
+  PRIORITY = 1000,
+}
+
+
+function RpcBatchTestHandler:init_worker()
+  kong.rpc.callbacks:register("kong.test.batch", function(node_id, greeting)
+    ngx.log(ngx.DEBUG, "kong.test.batch called")
+    return "hello ".. greeting
+  end)
+end
+
+
+function RpcBatchTestHandler:access()
+  kong.rpc:set_batch(1)
+
+  local res, err = kong.rpc:call("control_plane", "kong.test.batch", "world")
+  if not res then
+    return kong.response.exit(500, err)
+  end
+
+  return kong.response.exit(200, res)
+end
+
+
+return RpcBatchTestHandler

spec/fixtures/custom_plugins/kong/plugins/rpc-batch-test/schema.lua

@@ -0,0 +1,12 @@
+return {
+  name = "rpc-batch-test",
+  fields = {
+    {
+      config = {
+        type = "record",
+        fields = {
+        },
+      },
+    },
+  },
+}