lua中使用luasocket通信，写了一个简单的事件库

1.服务端，server.lua。功能有直接转发并执行命令，还有每日定时任务功能和普通定时任务功能，module如下：
提供的接口:
server.addrPort(addr, port)
server.runLoop()

#!/usr/bin/lua

local socket = require "socket"
local json = require "cjson.safe"
local log = require "log"

local M = {
    headLength = 10,
    address = "0.0.0.0",
    port = 9201,
    io = {},
    timer = {}
}

server = M

--外部调用
function M.addrPort(addr, port)
    M.address = addr
    M.port = port
end

function M.runLoop()
    if (false == M.createServerThread()) then
        log.error("run loop exit")
        return nil
    end
    M.doEvents()
end

--内部实现
function M.createNewIoEvent(_fd, _entry, _func, _event, _data)
    local eventInfo = {
        _fd = _fd,
        _entry = _entry,
        _event = _event,
        _data = _data,
        _func = _func
    }
    log.debug("create new io event, fd (%s), entry (%s), func (%s),  event (%s), data (%s)",
               _fd, _entry, _func, _event, _data)
    M.io[#M.io+1] = eventInfo
end

function M.createNewTimerEvent(_func, _at, _everyday_at, _interval, _repeat, _data)
    local eventInfo = {
        _func = _func,
        _at = _at,
        _everyday_at = _everyday_at,
        _interval = _interval,
        _repeat = _repeat,
        _data = _data
    }
    log.info("create new timer event, func (%s), at (%s), _everyday_at (%s)," ..
             " interval (%s),  repeat (%s), data (%s)",
              _func, _at, _everyday_at, _interval, _repeat, _data)
    M.timer[#M.timer+1] = eventInfo
end

function M.executeCommand(cmd)
    local file = assert(io.popen(cmd, 'r'))
    local data = file:read('*all')
    file:close()
    log.info("execute command (%s)", cmd)
    return data
end

function M.getEverydaySeconds()
    return tonumber(os.date("%H")) * 3600 +
           tonumber(os.date("%M")) * 60 +
           tonumber(os.date("%S"))
end

function M.addTimerTask(_func, _after, _everyday_at, _interval, _repeat, _data)
    local _at

    if (_everyday_at > 0) then
        if (_everyday_at >= M.getEverydaySeconds()) then
            _at = socket.gettime() - M.getEverydaySeconds() + _everyday_at
        else
            _at = socket.gettime() - M.getEverydaySeconds() + _everyday_at + 86400
        end
        log.debug("add new everyday timer task. everyday at (%s)", _at)
    else
        _at = socket.gettime() + _after
        log.debug("add new interval timer task. at (%s)", _at)
    end

    M.createNewTimerEvent(_func, _at, _everyday_at, _interval, _repeat, _data)
    return ""
end

function M.removeTimerTask(_data)
    local timer = M.timer
    for i=#timer, 1, -1 do
        if (timer[i]._data == _data) then
            log.info("remove timer task, data (%s)", _data)
            table.remove(timer, i)
        end
    end
    return ""
end

function M.packetStringData(s)
    local length = #s

    if (length >= 2^32) then
        return nil
    end

    local head = string.format("%0."..tostring(M.headLength).."u", #s)
    return head .. s
end

function M.packetAndSendData(_data, fd)
    local cmd = {
        cmd = 0,
        data = _data
    }

    local data = M.packetStringData(json.encode(cmd))
    local event = "closed"
    if (not data) then
        log.error("failed to packet data, data is to large. length (%s)", _data)
    end
    log.info("packet data successfully, start to send. fd (%s), data (%s)", fd, data)

    while true do
        --socket.sleep(1)
        fd:settimeout(0)
        local count, status = fd:send(data)

        if (status == "timeout") then
            log.info("send data timeout, now yield. fd (%s)", fd)
            coroutine.yield(fd)
        elseif (status == "closed") then
            log.info("closed by peer, fd (%s)", fd)
            event = "closed"
            break
        end

        if (nil ~= count) then
            if (count == #data) then
                log.info("send data completed, contine to read other command, fd (%s)", fd)
                event = "read"
                break
            else
                data = string.sub(data, count+1, -1)
                log.debug("send total %s, remain data (%s), fd (%s)", count, data, fd)
            end
        end
    end

    return event
end

function M.receiveAndParseData(fd)
    local event, rest
    local headlen = M.headLength
    local total = 0
    local data = ""

    log.info("start to receive data, fd (%s)", fd)
    while true do
        --socket.sleep(1)

        if (total == 0) then
            rest = headlen - #data
        else
            rest = total - #data
        end

        fd:settimeout(0)
        local s, status, partial = fd:receive(rest)
        local receive = (s or partial)
        data = data .. receive

        log.debug("new receive (%s), fd (%s)", receive, fd)

        if (#receive == 0) then -- partial is a string, so receive is a string
            if status == "timeout" then
                log.info("receive data timeout, now yield. fd (%s)", fd)
                coroutine.yield(fd)
            elseif status == "closed" then
                log.info("closed by peer, fd (%s)", fd)
                event = "closed"
                data = ""
                break
            end
        end

        if (#data >= 10) and (total == 0) then
            total = tonumber(string.sub(data, 1, 10))
            data = string.sub(data, 11, -1)
            log.info("receive head completed, len (%s), data len (%s)", 10, total)
        end

        if (#data > 0) and (#data == total) then
            log.info("receive data completed, len (%s), data (%s)", #data, data)
            event = "pending"
            data = data
            break
        end
    end

    return event, data
end

function M.findInfoNode(fd)
    local pos
    local io_ev = M.io
    for i=1, #io_ev do
        if io_ev[i]._fd == fd then
            pos = i
            break
        end
    end
    return io_ev[pos]
end

function M.clientCmdDispatchAndExecute(data)
    local cmdt = json.decode(data)

    log.info("dispatch and execute command.  data (%s)", data)

    if (cmdt.cmd == 1) then -- execute one script
            cmdt.data = M.executeCommand(cmdt.data)
    elseif (cmdt.cmd == 2) then -- add one timer task
            cmdt.data = M.addTimerTask(M.executeCommand, cmdt.after, cmdt.everyday_at,
                                       cmdt.interval, cmdt["repeat"], cmdt.data)
    elseif (cmdt.cmd == 3) then -- remove one timer task
            cmdt.data = M.removeTimerTask(cmdt.data)
    end

    return "write", cmdt.data
end

function M.doClient(client)
    local io_ev = M.findInfoNode(client)

    while (io_ev._event ~= "closed") do
        -- receive data
        if (io_ev._event == "read") then
            io_ev._event, io_ev._data = M.receiveAndParseData(client)
        end

        -- do real task
        if (io_ev._event == "pending") then
            io_ev._event, io_ev._data = io_ev._func(io_ev._data)
        end

        -- send data
        if (io_ev._event == "write") then
            io_ev._event = M.packetAndSendData(io_ev._data, client)
        end
    end
end

function M.acceptNewClient(server)
    server:settimeout(0)
    local newClient, status = server:accept()
    if newClient ~= nil then
        local entry = coroutine.create(function ()
            M.doClient(newClient)
        end)
        log.info("accept new client successfully, newClient (%s)", newClient)
        M.createNewIoEvent(newClient, entry, M.clientCmdDispatchAndExecute, "read", "")
    else
        log.debug("no need to accept new client, now yield. status (%s)", status)
        coroutine.yield(true, true)
    end
end

function M.doServer(server)
    log.debug("do server task")
    local io_ev = M.findInfoNode(server)
    while true do
        io_ev._func(server)
    end
end

function M.getSelectfd()
    local fd = {
        readfd = {},
        writefd = {}
    }

    local io_ev = M.io
    for i=#io_ev, 1, -1 do
        if io_ev[i]._event == "read" then
            log.debug("readfd (%s)", io_ev[i]._fd)
            fd.readfd[#fd.readfd+1] = io_ev[i]._fd
        elseif io_ev[i]._event == "write" then
            log.debug("writefd (%s)", io_ev[i]._fd)
            fd.writefd[#fd.writefd+1] = io_ev[i]._fd
        elseif io_ev[i]._event == "closed" then
            log.info("remove io event because of closed. fd (%s)", io_ev[i]._fd)
            table.remove(io_ev, i)
        end
    end

    return fd
end

function M.calculateSelectTime()
    local timer = M.timer

    if (#timer > 0) then
        local at = timer[1]._at
        for i=2, #timer do
            if (at > timer[i]._at) then
                at = timer[i]._at
            end
        end

        local now = socket.gettime()
        log.debug("now (%s), at (%s)", now, at)

        if (now >= at) then
            return 0
        else
            return at - now
        end
    else
        return nil
    end
end

function M.doIoEvent()
    local io_ev = M.io
    local dftime = M.calculateSelectTime()
    local fd = M.getSelectfd()

    socket.select(fd.readfd, fd.writefd, dftime)
    for i=#io_ev, 1, -1 do
        log.debug("coroutine resume, fd (%s)", io_ev[i]._fd)
        local status, res = coroutine.resume(io_ev[i]._entry)
        log.debug("resume response, status (%s), res (%s)", status, res)
        if (not status) then
            log.error("failed to resume coroutine, fd (%s)", io_ev[i]._fd)
        end

        if (not res) then
            log.info("io thread exit, fd (%s)", io_ev[i]._fd)
            table.remove(io_ev, i)
        end
    end
end

function M.doTimerEvent()
    local timer = M.timer
    local now = socket.gettime()

    for i=#timer, 1, -1 do
        if (timer[i]._at <= now) then
            log.info("execute timer event, data (%s), everyday_at (%s), interval (%s), repeat (%s), now(%s)",
                      timer[i]._data, timer[i]._everyday_at, timer[i]._interval, timer[i]._repeat, now)
            timer[i]._func(timer[i]._data)

            if (timer[i]._everyday_at < 0) then
                timer[i]._at = now + timer[i]._interval
                if (timer[i]._repeat > 0) then
                    timer[i]._repeat = timer[i]._repeat - 1
                    if (timer[i]._repeat == 0) then
                        log.info("remove timer event, data (%s), interval (%s), repeat (%s)",
                                     timer[i]._data, timer[i]._interval, timer[i]._repeat)
                        table.remove(timer, i)
                    end
                end
            else
                timer[i]._at = socket.gettime() - M.getEverydaySeconds()
                             + 86400 + timer[i]._everyday_at
            end
        end
    end
end

function M.doEvents()
    local io_ev = M.io
    local timer_ev = M.timer
    while true do
        M.doIoEvent()
        M.doTimerEvent()
    end
end

function M.createServer(srv)
    local server, ret

    server, err = socket.tcp()
    if (nil == server) then
        log.error("failed to create tcp socket, err (%s)", err)
        return nil, err
    end

    ret, err = server:setoption("reuseaddr", true)
    if (nil == ret) then
        log.error("failed to set option, err (%s)", err)
        return nil, err
    end

    ret, err = server:bind(srv.address, srv.port)
    if (nil == ret) then
        log.error("failed to bind server, err (%s)", err)
        return nil, err
    end

    ret, err = server:listen(100)
    if (nil == ret) then
        log.error("failed to create server listen, err (%s)", err)
        return nil, err
    end

    log.info("create tcp server successfully, fd (%s), addr (%s), port (%s)",
              server, srv.address, srv.port)
    return server
end

function M.createServerThread()
    local server = M.createServer(M)

    if (server == nil) then
        log.error("failed to create new server. addr (%s), port (%s)",
                   M.address, M.port)
        return false
    end

    local entry = coroutine.create(function ()
        M.doServer(server)
    end)
    M.createNewIoEvent(server, entry, M.acceptNewClient, "read", "")

    return true
end

return server

2.客服端，client.lua。功能主要是发送相应的命令给服务端执行，module如下：
提供的接口：
client.server(addr,port)
client.socketTimeout(sendTimeout, receiveTimeout)
client.send(_cmd, _data, _everyday_at, _after, _interval, _repeat)

#!/usr/bin/lua

local socket = require "socket"
local json = require "cjson.safe"
local log = require "log"

-- private data structure
local M = {
    sendTimeOut = 1,                -- Socket send timeout
    receiveTimeOut = 1, -- Socket receive timeout
    headLength = 10,                -- Data head length
    serverAddress = "127.0.0.1", -- Server ip address
    serverPort = 9201,                -- Server port
    data = {                        -- Real data
        cmd = 1,                    -- Command id, 0 - success, 1 - direct execution command, 2 - register timing task
        data = "",                    -- script
        everyday_at = -1,            -- everyday execute once at this time
        after = -1,                    -- Start timer task after seconds
        interval = -1,                -- Timer task interval
        ["repeat"] = -1                -- Execution times
    }
}

client = M

--外部调用
function M.server(addr, port)
    M.serverAddress = addr
    M.serverPort = port
end

function M.socketTimeout(sendTimeOut, receiveTimeOut)
    M.sendTimeOut = sendTimeOut
    M.receiveTimeOut = receiveTimeOut
end

function M.send(_cmd, _data, _everyday_at, _after, _interval, _repeat)
    local data = {
        cmd = _cmd,                    -- 命令
        data = _data,                -- 具体需要执行的命令
        everyday_at = _everyday_at,    -- 每日定时任务执行的时间，没有填-1
        after = _after,                -- 普通定时任务，之后几秒执行，没有填-1
        interval = _interval,        -- 普通定时任务，间隔几秒执行，没有填-1
        ["repeat"] = _repeat        -- 普通定时任务，执行几次
    }

    M.data = data
    return M.doClient()
end

--内部实现
function M.packetStringData(s)
    local length = #s

    if (length >= 2^32) then
        return nil
    end

    local head = string.format("%0.10u", #s)
    return head .. s
end

function M.sendData(fd, s)
    local data = s
    local ret
    log.debug("start to send data. data (%s), timeout (%s)", s, M.sendTimeOut)

    while #data > 0 do
        fd:settimeout(M.sendTimeOut)
        local count, status = fd:send(data)

        if (status == "timeout") then
            log.info("send data timeout")
            ret = status
            break
        elseif (status == "closed") then
            log.info("closed by peer")
            ret = status
            break
        end

        if (count ~= nil) then
            if (count == #data) then
                log.info("send data completed")
                break
            else
                data = string.sub(data, count+1, -1)
                log.debug("send total %s, remain data (%s)", count, data)
            end
        end
    end

    return ret
end

function M.receiveData(fd)
    local ret, rest
    local total = 0
    local data = ""

    log.debug("start to receive data. timout (%s)", M.receiveTimeOut)
    while true do
        if (total == 0) then
            rest = M.headLength - #data
        else
            rest = total - #data
        end

        fd:settimeout(M.receiveTimeOut)
        local s, status, partial = fd:receive(rest)
        local receive = (s or partial)
        data = data .. receive

        log.debug("new receive data (%s)", data)

        if (#receive == 0) then -- partial is a string, so receive is a string
            if status == "timeout" then
                log.info("receive data timeout.", fd)
                ret = "timeout"
                data = ""
                break
            elseif status == "closed" then
                log.info("closed by peer.", fd)
                ret = "closed"
                data = ""
                break
            end
        end

        if (#data >= 10) and (total == 0) then
            total = tonumber(string.sub(data, 1, 10))
            data = string.sub(data, 11, -1)
            log.info("receive head completed, len (%s), data len (%s)", 10, total)
        end

        if (#data > 0) and (#data == total) then
            log.info("receive data completed, len (%s), data (%s)", #data, data)
            data = data
            break
        end
    end

    return ret, data
end

function M.doClient()
    local err, ret
    local client = assert(socket.connect(M.serverAddress, M.serverPort))
    local data = M.packetStringData(json.encode(M.data))

    if (data == nil) then
        log.error("failed to packet data, client (%s)", json.encode(M))
        client:close(client)
        return nil
    end

    err = M.sendData(client, data)

    if (err == "timeout") or (err == "closed") then
        log.error("fail to send data, err (%s), client (%s)", err, json.encode(M))
        client:close(client)
        return nil
    end

    err, ret = M.receiveData(client)
    if (err == "timeout") or (err == "closed") then
        log.error("fail to receive data, err (%s), client (%s)", err, json.encode(M))
        client:close(client)
        return nil
    end

    log.info("do client successfully. ret (%s)", ret)
    return ret
end

return client

3.测试代码：
  客服端，test_client.lua

#!/usr/bin/lua

local server = require "server"
local log = require "log"

log.configure(4, "stdout", "./log")

server.addrPort("0.0.0.0", 9201)     -- 配置监听地址
server.runLoop()              -- 开启服务

  服务端，test_server.lua

#!/usr/bin/lua

local client = require "client"
local log = require "log"

log.configure(4, "stdout", "./log") -- 配置打印等级4，到标准输出

client.server("127.0.0.1", 9201)  -- 配置服务端ip和端口
client.socketTimeout(1, 1)    -- 配置数据发送和接收时间

-- 直接执行一次命令
local ret = client.send(1, "/bin/ls -a", -1, -1, -1, -1)
log.warn("%s", ret)

-- 每日定时任务，每日凌晨3点执行一次
local ret = client.send(2, "/bin/ls -l", 3 * 3600, -1, -1, -1)
log.warn("%s", ret)
-- 取消每日定时任务
local ret = client.send(3, "/bin/ls -l", -1, -1, -1, -1)
log.warn("%s", ret)

-- 两秒后执行，3秒执行一次，共执行4次
local ret = client.send(2, "/bin/ls -a", -1, 2, 3, 4)
log.warn("%s", ret)
-- 取消一般定时任务
local ret = client.send(3, "/bin/ls -a", -1, -1, -1, -1)
log.warn("%s", ret)

在两个不同的终端分别执行test_server.lua和test_client.lua就行了，可看到通信过程的打印和返回的数据。这个server.lua只能使用io.popen()执行命令，并不能执行用户定义的函数。没有实现自己注册命令的功能，这里做得比较失败了，有时间可以再改改。

4.注意事项
1) 首先用到了lua的cjson库，可以直接到官网下载，使用和下载地址：
2) 其次是luasokect库，使用和下载地址：
    这里主要要注意socket.receive()函数，这个函数接收数据的方式有多种，一定要小心，主要是它的第二个函数参数。
     '*a' : 设置这个参数之后，socket被closed之后才会返回，如果设置了阻塞方式，那么在socket关闭之前，这里将一直阻塞着，不会返回任何东西。
     '*l' : 设置这个参数之后，当收到数据中的字符有回车换行时才会返回，且回车换行在返回的数据中会被抹掉，如果收到的数据中没有回车换行，将永远不会返回。
     number : 这个表示缓冲中大于等于number个字符时，函数才返回，不然会一直阻塞。
    只要上边三种情况导致阻塞，就只能在sokect关闭的情况下才能返回。这时候缓冲中剩余的由于数据也会全部返回，并存放在返回的第三个参数partial中，而返回的第一个参数为nil。解决阻塞的办法就是在调用receive函数之前调用socket:settimeout()函数，设置阻塞的时间。超时过后，receive函数就会返回，这时如果缓冲中有数据，那么会一起返回，在partial中。
3) 关于用到的log函数则是上一遍博客介绍的打印module。