---
--- Generated by UUSEC(https://www.uusec.com/)
--- Created by Safe3.
--- DateTime: 2022/9/21 20:37
---
local producer = require("resty.kafka.producer")
local log = require("waf.log")

local _M = {
    version = 0.1,
    name = "kafka-logger"
}

--[[
function _M.req_filter(waf)

end

function _M.resp_header_filter(waf)

end

function _M.resp_body_filter(waf)

end
--]]

local function kafkaLog(_, brokerList, info)
    local kp = producer:new(brokerList, { producer_type = "async" })
    local key = "key"
    local message = log.encodeJson(info)
    local ok, err = kp:send("waf-log", key, message)
    if not ok then
        log.errLog(_M.name, " send err: ", err)
    end
end

function _M.log(waf)
    local brokerList = {
        {
            host = "127.0.0.1",
            port = 9092,

            sasl_config = {
                mechanism = "PLAIN",
                user = "USERNAME",
                password = "PASSWORD",
            },
        }
    }

    if waf.msg then
        local country, province, city = log.ip2loc(waf.ip)
        local info = { rule_id = waf.rule_id, ip = waf.ip, host = waf.host, url = waf.reqUri, data = waf.msg, country = country, province = province, city = city }
        log.broker(kafkaLog, brokerList, info)
    end
end

return _M