apisix/plugins/http-dubbo.lua

--
-- Licensed to the Apache Software Foundation (ASF) under one or more
-- contributor license agreements.  See the NOTICE file distributed with
-- this work for additional information regarding copyright ownership.
-- The ASF licenses this file to You 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.
--

local require = require
local core = require("apisix.core")
local pairs = pairs
local str_format = string.format
local bit = require("bit")
local rshift = bit.rshift
local band = bit.band
local char = string.char
local tostring = tostring
local ngx = ngx
local type = type
local plugin_name = "http-dubbo"


local schema = {
    type = "object",
    properties = {
        service_name = {
            type = "string",
            minLength = 1,
        },
        service_version = {
            type = "string",
            pattern = [[^\d+\.\d+\.\d+]],
            default ="0.0.0"
        },
        method = {
            type = "string",
            minLength = 1,
        },
        params_type_desc = {
            type = "string",
            default = ""
        },
        serialization_header_key = {
            type = "string"
        },
        serialized = {
            type = "boolean",
            default = false
        },
        connect_timeout={
            type = "number",
            default = 6000
        },
        read_timeout={
            type = "number",
            default = 6000
        },
        send_timeout={
            type = "number",
            default = 6000
        }
    },
    required = { "service_name", "method" },
}

local _M = {
    version = 0.1,
    priority = 504,
    name = plugin_name,
    schema = schema,
}

function _M.check_schema(conf)
    return core.schema.check(schema, conf)
end


local function str_int32(int)
    return char(band(rshift(int, 24), 0xff),
            band(rshift(int, 16), 0xff),
            band(rshift(int, 8), 0xff),
            band(int, 0xff))
end


local function parse_dubbo_header(header)
    for i = 1, 16 do
        local currentByte = header:byte(i)
        if not currentByte then
            return nil
        end
    end

    local magic_number = str_format("%04x", header:byte(1) * 256 + header:byte(2))
    local message_flag = header:byte(3)
    local status = header:byte(4)
    local request_id = 0
    for i = 5, 12 do
        request_id = request_id * 256 + header:byte(i)
    end

    local byte13Val = header:byte(13) * 256 * 256 * 256
    local byte14Val = header:byte(14) * 256 * 256
    local data_length = byte13Val + byte14Val + header:byte(15) * 256 + header:byte(16)

    local is_request = bit.band(bit.rshift(message_flag, 7), 0x01) == 1 and 1 or 0
    local is_two_way = bit.band(bit.rshift(message_flag, 6), 0x01) == 1 and 1 or 0
    local is_event = bit.band(bit.rshift(message_flag, 5), 0x01) == 1 and 1 or 0

    return {
        magic_number = magic_number,
        message_flag = message_flag,
        is_request = is_request,
        is_two_way = is_two_way,
        is_event = is_event,
        status = status,
        request_id = request_id,
        data_length = data_length
    }
end


local function string_to_json_string(str)
    local result = "\""
    for i = 1, #str do
        local byte = core.string.sub(str, i, i)
        if byte == "\\" then
            result = result .. "\\\\"
        elseif byte == "\n" then
            result = result .. "\\n"
        elseif byte == "\t" then
            result = result .. "\\t"
        elseif byte == "\r" then
            result = result .. "\\r"
        elseif byte == "\b" then
            result = result .. "\\b"
        elseif byte == "\f" then
            result = result .. "\\f"
        elseif byte == "\"" then
            result = result .. "\\\""
        else
            result = result .. byte
        end
    end
    return result .. "\""
end


local function get_dubbo_request(conf, ctx)
    -- use dubbo and fastjson
    local first_byte4 = "\xda\xbb\xc6\x00"

    local requestId = "\x00\x00\x00\x00\x00\x00\x00\x01"
    local version = "\"2.0.2\"\n"
    local service = "\"" .. conf.service_name .. "\"" .. "\n"

    local service_version = "\"" ..  conf.service_version .. "\"" .. "\n"
    local method_name = "\"" .. conf.method .. "\"" .. "\n"

    local params_desc = "\"" .. conf.params_type_desc .. "\"" .. "\n"
    local params = ""
    local serialized = conf.serialized
    if conf.serialization_header_key then
        local serialization_header = core.request.header(ctx, conf.serialization_header_key)
        serialized = serialization_header == "true"
    end
    if serialized then
        params = core.request.get_body()
        if params then
            local end_of_params = core.string.sub(params, -1)
            if end_of_params ~= "\n" then
                params = params .. "\n"
            end
        end
    else
        local body_data = core.request.get_body()
        if body_data then
            local lua_object = core.json.decode(body_data);
            for _, v in pairs(lua_object) do
                local pt = type(v)
                if pt == "nil" then
                    params = params .. "null" .. "\n"
                elseif pt == "string" then
                    params = params .. string_to_json_string(v) .. "\n"
                elseif pt == "number" then
                    params = params .. tostring(v) .. "\n"
                else
                    params = params .. core.json.encode(v) .. "\n"
                end
            end
        end

    end
    local attachments = "{}\n"
    if params == nil then
        params = ""
    end
    local payload = #version + #service + #service_version
            + #method_name + #params_desc + #params + #attachments
    return {
        first_byte4,
        requestId,
        str_int32(payload),
        version,
        service,
        service_version,
        method_name,
        params_desc,
        params,
        attachments
    }
end


function _M.before_proxy(conf, ctx)
    local sock = ngx.socket.tcp()

    sock:settimeouts(conf.connect_timeout, conf.send_timeout, conf.read_timeout)
    local ok, err = sock:connect(ctx.picked_server.host, ctx.picked_server.port)
    if not ok then
        sock:close()
        core.log.error("failed to connect to upstream ", err)
        return 502
    end
    local request = get_dubbo_request(conf, ctx)
    local bytes, _ = sock:send(request)
    if bytes > 0 then
        local header, _ = sock:receiveany(16);
        if header then
            local header_info = parse_dubbo_header(header)
            if header_info and header_info.status == 20 then
                local readline = sock:receiveuntil("\n")
                local body_status, _, _ = readline()
                if body_status then
                    local response_status = core.string.sub(body_status, 1, 1)
                    if response_status == "2" or response_status == "5" then
                        sock:close()
                        return 200
                    elseif response_status == "1" or response_status == "4" then
                        local body, _, _ = readline()
                        sock:close()
                        return 200, body
                    end
                end
            end
        end
    end
    sock:close()
    return 500

end

return _M