sokol-luajit/lua/copas.lua

2084 lines
60 KiB
Lua

-------------------------------------------------------------------------------
-- Copas - Coroutine Oriented Portable Asynchronous Services
--
-- A dispatcher based on coroutines that can be used by TCP/IP servers.
-- Uses LuaSocket as the interface with the TCP/IP stack.
--
-- Authors: Andre Carregal, Javier Guerra, and Fabio Mascarenhas
-- Contributors: Diego Nehab, Mike Pall, David Burgess, Leonardo Godinho,
-- Thomas Harning Jr., and Gary NG
--
-- Copyright 2005-2013 - Kepler Project (www.keplerproject.org), 2015-2024 Thijs Schreijer
--
-- $Id: copas.lua,v 1.37 2009/04/07 22:09:52 carregal Exp $
-------------------------------------------------------------------------------
if package.loaded["socket.http"] and (_VERSION=="Lua 5.1") then -- obsolete: only for Lua 5.1 compatibility
error("you must require copas before require'ing socket.http")
end
if package.loaded["copas.http"] and (_VERSION=="Lua 5.1") then -- obsolete: only for Lua 5.1 compatibility
error("you must require copas before require'ing copas.http")
end
-- load either LuaSocket, or LuaSystem
-- note: with luasocket we don't use 'sleep' but 'select' with no sockets
local socket, system, system_sleep do
if pcall(require, "socket") then
-- found LuaSocket
socket = require "socket"
end
-- try LuaSystem as fallback
if pcall(require, "system") then
system = require "system"
system_sleep = system.sleep -- system.sleep will be patched later on
end
if not (socket or system) then
error("Neither LuaSocket nor LuaSystem found, Copas requires at least one of them")
end
end
local binaryheap = require "binaryheap"
local gettime = (socket or system).gettime
local ssl -- only loaded upon demand
local WATCH_DOG_TIMEOUT = 120
local UDP_DATAGRAM_MAX = (socket or {})._DATAGRAMSIZE or 8192
local TIMEOUT_PRECISION = 0.1 -- 100ms
local fnil = function() end
local coroutine_create = coroutine.create
local coroutine_running = coroutine.running
local coroutine_yield = coroutine.yield
local coroutine_resume = coroutine.resume
local coroutine_status = coroutine.status
-- nil-safe versions for pack/unpack
local _unpack = unpack or table.unpack
local unpack = function(t, i, j) return _unpack(t, i or 1, j or t.n or #t) end
local pack = function(...) return { n = select("#", ...), ...} end
local pcall = pcall
if _VERSION=="Lua 5.1" and not jit then -- obsolete: only for Lua 5.1 compatibility
pcall = require("coxpcall").pcall
coroutine_running = require("coxpcall").running
end
if socket then
-- Redefines LuaSocket functions with coroutine safe versions (pure Lua)
-- (this allows the use of socket.http from within copas)
local err_mt = {
__tostring = function (self)
return "Copas 'try' error intermediate table: '"..tostring(self[1].."'")
end,
}
local function statusHandler(status, ...)
if status then return ... end
local err = (...)
if type(err) == "table" and getmetatable(err) == err_mt then
return nil, err[1]
else
error(err)
end
end
function socket.protect(func)
return function (...)
return statusHandler(pcall(func, ...))
end
end
function socket.newtry(finalizer)
return function (...)
local status = (...)
if not status then
pcall(finalizer or fnil, select(2, ...))
error(setmetatable({ (select(2, ...)) }, err_mt), 0)
end
return ...
end
end
socket.try = socket.newtry()
end
-- Setup the Copas meta table to auto-load submodules and define a default method
local copas do
local submodules = { "ftp", "http", "lock", "queue", "semaphore", "smtp", "timer" }
for i, key in ipairs(submodules) do
submodules[key] = true
submodules[i] = nil
end
copas = setmetatable({},{
__index = function(self, key)
if submodules[key] then
self[key] = require("copas."..key)
submodules[key] = nil
return rawget(self, key)
end
end,
__call = function(self, ...)
return self.loop(...)
end,
})
end
-- Meta information is public even if beginning with an "_"
copas._COPYRIGHT = "Copyright (C) 2005-2013 Kepler Project, 2015-2024 Thijs Schreijer"
copas._DESCRIPTION = "Coroutine Oriented Portable Asynchronous Services"
copas._VERSION = "Copas 4.7.1"
-- Close the socket associated with the current connection after the handler finishes
copas.autoclose = true
-- indicator for the loop running
copas.running = false
-- gettime method from either LuaSocket or LuaSystem: time in (fractional) seconds, since epoch.
copas.gettime = gettime
-------------------------------------------------------------------------------
-- Object names, to track names of thread/coroutines and sockets
-------------------------------------------------------------------------------
local object_names = setmetatable({}, {
__mode = "k",
__index = function(self, key)
local name = tostring(key)
if key ~= nil then
rawset(self, key, name)
end
return name
end
})
-------------------------------------------------------------------------------
-- Simple set implementation
-- adds a FIFO queue for each socket in the set
-------------------------------------------------------------------------------
local function newsocketset()
local set = {}
do -- set implementation
local reverse = {}
-- Adds a socket to the set, does nothing if it exists
-- @return skt if added, or nil if it existed
function set:insert(skt)
if not reverse[skt] then
self[#self + 1] = skt
reverse[skt] = #self
return skt
end
end
-- Removes socket from the set, does nothing if not found
-- @return skt if removed, or nil if it wasn't in the set
function set:remove(skt)
local index = reverse[skt]
if index then
reverse[skt] = nil
local top = self[#self]
self[#self] = nil
if top ~= skt then
reverse[top] = index
self[index] = top
end
return skt
end
end
end
do -- queues implementation
local fifo_queues = setmetatable({},{
__mode = "k", -- auto collect queue if socket is gone
__index = function(self, skt) -- auto create fifo queue if not found
local newfifo = {}
self[skt] = newfifo
return newfifo
end,
})
-- pushes an item in the fifo queue for the socket.
function set:push(skt, itm)
local queue = fifo_queues[skt]
queue[#queue + 1] = itm
end
-- pops an item from the fifo queue for the socket
function set:pop(skt)
local queue = fifo_queues[skt]
return table.remove(queue, 1)
end
end
return set
end
-- Threads immediately resumable
local _resumable = {} do
local resumelist = {}
function _resumable:push(co)
resumelist[#resumelist + 1] = co
end
function _resumable:clear_resumelist()
local lst = resumelist
resumelist = {}
return lst
end
function _resumable:done()
return resumelist[1] == nil
end
function _resumable:count()
return #resumelist + #_resumable
end
end
-- Similar to the socket set above, but tailored for the use of
-- sleeping threads
local _sleeping = {} do
local heap = binaryheap.minUnique()
local lethargy = setmetatable({}, { __mode = "k" }) -- list of coroutines sleeping without a wakeup time
-- Required base implementation
-----------------------------------------
_sleeping.insert = fnil
_sleeping.remove = fnil
-- push a new timer on the heap
function _sleeping:push(sleeptime, co)
if sleeptime < 0 then
lethargy[co] = true
elseif sleeptime == 0 then
_resumable:push(co)
else
heap:insert(gettime() + sleeptime, co)
end
end
-- find the thread that should wake up to the time, if any
function _sleeping:pop(time)
if time < (heap:peekValue() or math.huge) then
return
end
return heap:pop()
end
-- additional methods for time management
-----------------------------------------
function _sleeping:getnext() -- returns delay until next sleep expires, or nil if there is none
local t = heap:peekValue()
if t then
-- never report less than 0, because select() might block
return math.max(t - gettime(), 0)
end
end
function _sleeping:wakeup(co)
if lethargy[co] then
lethargy[co] = nil
_resumable:push(co)
return
end
if heap:remove(co) then
_resumable:push(co)
end
end
function _sleeping:cancel(co)
lethargy[co] = nil
heap:remove(co)
end
-- @param tos number of timeouts running
function _sleeping:done(tos)
-- return true if we have nothing more to do
-- the timeout task doesn't qualify as work (fallbacks only),
-- the lethargy also doesn't qualify as work ('dead' tasks),
-- but the combination of a timeout + a lethargy can be work
return heap:size() == 1 -- 1 means only the timeout-timer task is running
and not (tos > 0 and next(lethargy))
end
-- gets number of threads in binaryheap and lethargy
function _sleeping:status()
local c = 0
for _ in pairs(lethargy) do c = c + 1 end
return heap:size(), c
end
end -- _sleeping
-------------------------------------------------------------------------------
-- Tracking coroutines and sockets
-------------------------------------------------------------------------------
local _servers = newsocketset() -- servers being handled
local _threads = setmetatable({}, {__mode = "k"}) -- registered threads added with addthread()
local _canceled = setmetatable({}, {__mode = "k"}) -- threads that are canceled and pending removal
local _autoclose = setmetatable({}, {__mode = "kv"}) -- sockets (value) to close when a thread (key) exits
local _autoclose_r = setmetatable({}, {__mode = "kv"}) -- reverse: sockets (key) to close when a thread (value) exits
-- for each socket we log the last read and last write times to enable the
-- watchdog to follow up if it takes too long.
-- tables contain the time, indexed by the socket
local _reading_log = {}
local _writing_log = {}
local _closed = {} -- track sockets that have been closed (list/array)
local _reading = newsocketset() -- sockets currently being read
local _writing = newsocketset() -- sockets currently being written
local _isSocketTimeout = { -- set of errors indicating a socket-timeout
["timeout"] = true, -- default LuaSocket timeout
["wantread"] = true, -- LuaSec specific timeout
["wantwrite"] = true, -- LuaSec specific timeout
}
-------------------------------------------------------------------------------
-- Coroutine based socket timeouts.
-------------------------------------------------------------------------------
local user_timeouts_connect
local user_timeouts_send
local user_timeouts_receive
do
local timeout_mt = {
__mode = "k",
__index = function(self, skt)
-- if there is no timeout found, we insert one automatically, to block forever
self[skt] = math.huge
return self[skt]
end,
}
user_timeouts_connect = setmetatable({}, timeout_mt)
user_timeouts_send = setmetatable({}, timeout_mt)
user_timeouts_receive = setmetatable({}, timeout_mt)
end
local useSocketTimeoutErrors = setmetatable({},{ __mode = "k" })
-- sto = socket-time-out
local sto_timeout, sto_timed_out, sto_change_queue, sto_error do
local socket_register = setmetatable({}, { __mode = "k" }) -- socket by coroutine
local operation_register = setmetatable({}, { __mode = "k" }) -- operation "read"/"write" by coroutine
local timeout_flags = setmetatable({}, { __mode = "k" }) -- true if timedout, by coroutine
local function socket_callback(co)
local skt = socket_register[co]
local queue = operation_register[co]
-- flag the timeout and resume the coroutine
timeout_flags[co] = true
_resumable:push(co)
-- clear the socket from the current queue
if queue == "read" then
_reading:remove(skt)
elseif queue == "write" then
_writing:remove(skt)
else
error("bad queue name; expected 'read'/'write', got: "..tostring(queue))
end
end
-- Sets a socket timeout.
-- Calling it as `sto_timeout()` will cancel the timeout.
-- @param queue (string) the queue the socket is currently in, must be either "read" or "write"
-- @param skt (socket) the socket on which to operate
-- @param use_connect_to (bool) timeout to use is determined based on queue (read/write) or if this
-- is truthy, it is the connect timeout.
-- @return true
function sto_timeout(skt, queue, use_connect_to)
local co = coroutine_running()
socket_register[co] = skt
operation_register[co] = queue
timeout_flags[co] = nil
if skt then
local to = (use_connect_to and user_timeouts_connect[skt]) or
(queue == "read" and user_timeouts_receive[skt]) or
user_timeouts_send[skt]
copas.timeout(to, socket_callback)
else
copas.timeout(0)
end
return true
end
-- Changes the timeout to a different queue (read/write).
-- Only usefull with ssl-handshakes and "wantread", "wantwrite" errors, when
-- the queue has to be changed, so the timeout handler knows where to find the socket.
-- @param queue (string) the new queue the socket is in, must be either "read" or "write"
-- @return true
function sto_change_queue(queue)
operation_register[coroutine_running()] = queue
return true
end
-- Responds with `true` if the operation timed-out.
function sto_timed_out()
return timeout_flags[coroutine_running()]
end
-- Returns the proper timeout error
function sto_error(err)
return useSocketTimeoutErrors[coroutine_running()] and err or "timeout"
end
end
-------------------------------------------------------------------------------
-- Coroutine based socket I/O functions.
-------------------------------------------------------------------------------
-- Returns "tcp"" for plain TCP and "ssl" for ssl-wrapped sockets, so truthy
-- for tcp based, and falsy for udp based.
local isTCP do
local lookup = {
tcp = "tcp",
SSL = "ssl",
}
function isTCP(socket)
return lookup[tostring(socket):sub(1,3)]
end
end
function copas.close(skt, ...)
_closed[#_closed+1] = skt
return skt:close(...)
end
-- nil or negative is indefinitly
function copas.settimeout(skt, timeout)
timeout = timeout or -1
if type(timeout) ~= "number" then
return nil, "timeout must be 'nil' or a number"
end
return copas.settimeouts(skt, timeout, timeout, timeout)
end
-- negative is indefinitly, nil means do not change
function copas.settimeouts(skt, connect, send, read)
if connect ~= nil and type(connect) ~= "number" then
return nil, "connect timeout must be 'nil' or a number"
end
if connect then
if connect < 0 then
connect = nil
end
user_timeouts_connect[skt] = connect
end
if send ~= nil and type(send) ~= "number" then
return nil, "send timeout must be 'nil' or a number"
end
if send then
if send < 0 then
send = nil
end
user_timeouts_send[skt] = send
end
if read ~= nil and type(read) ~= "number" then
return nil, "read timeout must be 'nil' or a number"
end
if read then
if read < 0 then
read = nil
end
user_timeouts_receive[skt] = read
end
return true
end
-- reads a pattern from a client and yields to the reading set on timeouts
-- UDP: a UDP socket expects a second argument to be a number, so it MUST
-- be provided as the 'pattern' below defaults to a string. Will throw a
-- 'bad argument' error if omitted.
function copas.receive(client, pattern, part)
local s, err
pattern = pattern or "*l"
local current_log = _reading_log
sto_timeout(client, "read")
repeat
s, err, part = client:receive(pattern, part)
-- guarantees that high throughput doesn't take other threads to starvation
if (math.random(100) > 90) then
copas.pause()
end
if s then
current_log[client] = nil
sto_timeout()
return s, err, part
elseif not _isSocketTimeout[err] then
current_log[client] = nil
sto_timeout()
return s, err, part
elseif sto_timed_out() then
current_log[client] = nil
return nil, sto_error(err), part
end
if err == "wantwrite" then -- wantwrite may be returned during SSL renegotiations
current_log = _writing_log
current_log[client] = gettime()
sto_change_queue("write")
coroutine_yield(client, _writing)
else
current_log = _reading_log
current_log[client] = gettime()
sto_change_queue("read")
coroutine_yield(client, _reading)
end
until false
end
-- receives data from a client over UDP. Not available for TCP.
-- (this is a copy of receive() method, adapted for receivefrom() use)
function copas.receivefrom(client, size)
local s, err, port
size = size or UDP_DATAGRAM_MAX
sto_timeout(client, "read")
repeat
s, err, port = client:receivefrom(size) -- upon success err holds ip address
-- garantees that high throughput doesn't take other threads to starvation
if (math.random(100) > 90) then
copas.pause()
end
if s then
_reading_log[client] = nil
sto_timeout()
return s, err, port
elseif err ~= "timeout" then
_reading_log[client] = nil
sto_timeout()
return s, err, port
elseif sto_timed_out() then
_reading_log[client] = nil
return nil, sto_error(err), port
end
_reading_log[client] = gettime()
coroutine_yield(client, _reading)
until false
end
-- same as above but with special treatment when reading chunks,
-- unblocks on any data received.
function copas.receivepartial(client, pattern, part)
local s, err
pattern = pattern or "*l"
local orig_size = #(part or "")
local current_log = _reading_log
sto_timeout(client, "read")
repeat
s, err, part = client:receive(pattern, part)
-- guarantees that high throughput doesn't take other threads to starvation
if (math.random(100) > 90) then
copas.pause()
end
if s or (type(part) == "string" and #part > orig_size) then
current_log[client] = nil
sto_timeout()
return s, err, part
elseif not _isSocketTimeout[err] then
current_log[client] = nil
sto_timeout()
return s, err, part
elseif sto_timed_out() then
current_log[client] = nil
return nil, sto_error(err), part
end
if err == "wantwrite" then
current_log = _writing_log
current_log[client] = gettime()
sto_change_queue("write")
coroutine_yield(client, _writing)
else
current_log = _reading_log
current_log[client] = gettime()
sto_change_queue("read")
coroutine_yield(client, _reading)
end
until false
end
copas.receivePartial = copas.receivepartial -- compat: receivePartial is deprecated
-- sends data to a client. The operation is buffered and
-- yields to the writing set on timeouts
-- Note: from and to parameters will be ignored by/for UDP sockets
function copas.send(client, data, from, to)
local s, err
from = from or 1
local lastIndex = from - 1
local current_log = _writing_log
sto_timeout(client, "write")
repeat
s, err, lastIndex = client:send(data, lastIndex + 1, to)
-- guarantees that high throughput doesn't take other threads to starvation
if (math.random(100) > 90) then
copas.pause()
end
if s then
current_log[client] = nil
sto_timeout()
return s, err, lastIndex
elseif not _isSocketTimeout[err] then
current_log[client] = nil
sto_timeout()
return s, err, lastIndex
elseif sto_timed_out() then
current_log[client] = nil
return nil, sto_error(err), lastIndex
end
if err == "wantread" then
current_log = _reading_log
current_log[client] = gettime()
sto_change_queue("read")
coroutine_yield(client, _reading)
else
current_log = _writing_log
current_log[client] = gettime()
sto_change_queue("write")
coroutine_yield(client, _writing)
end
until false
end
function copas.sendto(client, data, ip, port)
-- deprecated; for backward compatibility only, since UDP doesn't block on sending
return client:sendto(data, ip, port)
end
-- waits until connection is completed
function copas.connect(skt, host, port)
skt:settimeout(0)
local ret, err, tried_more_than_once
sto_timeout(skt, "write", true)
repeat
ret, err = skt:connect(host, port)
-- non-blocking connect on Windows results in error "Operation already
-- in progress" to indicate that it is completing the request async. So essentially
-- it is the same as "timeout"
if ret or (err ~= "timeout" and err ~= "Operation already in progress") then
_writing_log[skt] = nil
sto_timeout()
-- Once the async connect completes, Windows returns the error "already connected"
-- to indicate it is done, so that error should be ignored. Except when it is the
-- first call to connect, then it was already connected to something else and the
-- error should be returned
if (not ret) and (err == "already connected" and tried_more_than_once) then
return 1
end
return ret, err
elseif sto_timed_out() then
_writing_log[skt] = nil
return nil, sto_error(err)
end
tried_more_than_once = tried_more_than_once or true
_writing_log[skt] = gettime()
coroutine_yield(skt, _writing)
until false
end
-- Wraps a tcp socket in an ssl socket and configures it. If the socket was
-- already wrapped, it does nothing and returns the socket.
-- @param wrap_params the parameters for the ssl-context
-- @return wrapped socket, or throws an error
local function ssl_wrap(skt, wrap_params)
if isTCP(skt) == "ssl" then return skt end -- was already wrapped
if not wrap_params then
error("cannot wrap socket into a secure socket (using 'ssl.wrap()') without parameters/context")
end
ssl = ssl or require("ssl")
local nskt = assert(ssl.wrap(skt, wrap_params)) -- assert, because we do not want to silently ignore this one!!
nskt:settimeout(0) -- non-blocking on the ssl-socket
copas.settimeouts(nskt, user_timeouts_connect[skt],
user_timeouts_send[skt], user_timeouts_receive[skt]) -- copy copas user-timeout to newly wrapped one
local co = _autoclose_r[skt]
if co then
-- socket registered for autoclose, move registration to wrapped one
_autoclose[co] = nskt
_autoclose_r[skt] = nil
_autoclose_r[nskt] = co
end
local sock_name = object_names[skt]
if sock_name ~= tostring(skt) then
-- socket had a custom name, so copy it over
object_names[nskt] = sock_name
end
return nskt
end
-- For each luasec method we have a subtable, allows for future extension.
-- Required structure:
-- {
-- wrap = ... -- parameter to 'wrap()'; the ssl parameter table, or the context object
-- sni = { -- parameters to 'sni()'
-- names = string | table -- 1st parameter
-- strict = bool -- 2nd parameter
-- }
-- }
local function normalize_sslt(sslt)
local t = type(sslt)
local r = setmetatable({}, {
__index = function(self, key)
-- a bug if this happens, here as a sanity check, just being careful since
-- this is security stuff
error("accessing unknown 'ssl_params' table key: "..tostring(key))
end,
})
if t == "nil" then
r.wrap = false
r.sni = false
elseif t == "table" then
if sslt.mode or sslt.protocol then
-- has the mandatory fields for the ssl-params table for handshake
-- backward compatibility
r.wrap = sslt
r.sni = false
else
-- has the target definition, copy our known keys
r.wrap = sslt.wrap or false -- 'or false' because we do not want nils
r.sni = sslt.sni or false -- 'or false' because we do not want nils
end
elseif t == "userdata" then
-- it's an ssl-context object for the handshake
-- backward compatibility
r.wrap = sslt
r.sni = false
else
error("ssl parameters; did not expect type "..tostring(sslt))
end
return r
end
---
-- Peforms an (async) ssl handshake on a connected TCP client socket.
-- NOTE: if not ssl-wrapped already, then replace all previous socket references, with the returned new ssl wrapped socket
-- Throws error and does not return nil+error, as that might silently fail
-- in code like this;
-- copas.addserver(s1, function(skt)
-- skt = copas.wrap(skt, sparams)
-- skt:dohandshake() --> without explicit error checking, this fails silently and
-- skt:send(body) --> continues unencrypted
-- @param skt Regular LuaSocket CLIENT socket object
-- @param wrap_params Table with ssl parameters
-- @return wrapped ssl socket, or throws an error
function copas.dohandshake(skt, wrap_params)
ssl = ssl or require("ssl")
local nskt = ssl_wrap(skt, wrap_params)
sto_timeout(nskt, "write", true)
local queue
repeat
local success, err = nskt:dohandshake()
if success then
sto_timeout()
return nskt
elseif not _isSocketTimeout[err] then
sto_timeout()
error("TLS/SSL handshake failed: " .. tostring(err))
elseif sto_timed_out() then
return nil, sto_error(err)
elseif err == "wantwrite" then
sto_change_queue("write")
queue = _writing
elseif err == "wantread" then
sto_change_queue("read")
queue = _reading
else
error("TLS/SSL handshake failed: " .. tostring(err))
end
coroutine_yield(nskt, queue)
until false
end
-- flushes a client write buffer (deprecated)
function copas.flush()
end
-- wraps a TCP socket to use Copas methods (send, receive, flush and settimeout)
local _skt_mt_tcp = {
__tostring = function(self)
return tostring(self.socket).." (copas wrapped)"
end,
__index = {
send = function (self, data, from, to)
return copas.send (self.socket, data, from, to)
end,
receive = function (self, pattern, prefix)
if user_timeouts_receive[self.socket] == 0 then
return copas.receivepartial(self.socket, pattern, prefix)
end
return copas.receive(self.socket, pattern, prefix)
end,
receivepartial = function (self, pattern, prefix)
return copas.receivepartial(self.socket, pattern, prefix)
end,
flush = function (self)
return copas.flush(self.socket)
end,
settimeout = function (self, time)
return copas.settimeout(self.socket, time)
end,
settimeouts = function (self, connect, send, receive)
return copas.settimeouts(self.socket, connect, send, receive)
end,
-- TODO: socket.connect is a shortcut, and must be provided with an alternative
-- if ssl parameters are available, it will also include a handshake
connect = function(self, ...)
local res, err = copas.connect(self.socket, ...)
if res then
if self.ssl_params.sni then self:sni() end
if self.ssl_params.wrap then res, err = self:dohandshake() end
end
return res, err
end,
close = function(self, ...)
return copas.close(self.socket, ...)
end,
-- TODO: socket.bind is a shortcut, and must be provided with an alternative
bind = function(self, ...) return self.socket:bind(...) end,
-- TODO: is this DNS related? hence blocking?
getsockname = function(self, ...)
local ok, ip, port, family = pcall(self.socket.getsockname, self.socket, ...)
if ok then
return ip, port, family
else
return nil, "not implemented by LuaSec"
end
end,
getstats = function(self, ...) return self.socket:getstats(...) end,
setstats = function(self, ...) return self.socket:setstats(...) end,
listen = function(self, ...) return self.socket:listen(...) end,
accept = function(self, ...) return self.socket:accept(...) end,
setoption = function(self, ...)
local ok, res, err = pcall(self.socket.setoption, self.socket, ...)
if ok then
return res, err
else
return nil, "not implemented by LuaSec"
end
end,
getoption = function(self, ...)
local ok, val, err = pcall(self.socket.getoption, self.socket, ...)
if ok then
return val, err
else
return nil, "not implemented by LuaSec"
end
end,
-- TODO: is this DNS related? hence blocking?
getpeername = function(self, ...)
local ok, ip, port, family = pcall(self.socket.getpeername, self.socket, ...)
if ok then
return ip, port, family
else
return nil, "not implemented by LuaSec"
end
end,
shutdown = function(self, ...) return self.socket:shutdown(...) end,
sni = function(self, names, strict)
local sslp = self.ssl_params
self.socket = ssl_wrap(self.socket, sslp.wrap)
if names == nil then
names = sslp.sni.names
strict = sslp.sni.strict
end
return self.socket:sni(names, strict)
end,
dohandshake = function(self, wrap_params)
local nskt, err = copas.dohandshake(self.socket, wrap_params or self.ssl_params.wrap)
if not nskt then return nskt, err end
self.socket = nskt -- replace internal socket with the newly wrapped ssl one
return self
end,
getalpn = function(self, ...)
local ok, proto, err = pcall(self.socket.getalpn, self.socket, ...)
if ok then
return proto, err
else
return nil, "not a tls socket"
end
end,
getsniname = function(self, ...)
local ok, name, err = pcall(self.socket.getsniname, self.socket, ...)
if ok then
return name, err
else
return nil, "not a tls socket"
end
end,
}
}
-- wraps a UDP socket, copy of TCP one adapted for UDP.
local _skt_mt_udp = {__index = { }}
for k,v in pairs(_skt_mt_tcp) do _skt_mt_udp[k] = _skt_mt_udp[k] or v end
for k,v in pairs(_skt_mt_tcp.__index) do _skt_mt_udp.__index[k] = v end
_skt_mt_udp.__index.send = function(self, ...) return self.socket:send(...) end
_skt_mt_udp.__index.sendto = function(self, ...) return self.socket:sendto(...) end
_skt_mt_udp.__index.receive = function (self, size)
return copas.receive (self.socket, (size or UDP_DATAGRAM_MAX))
end
_skt_mt_udp.__index.receivefrom = function (self, size)
return copas.receivefrom (self.socket, (size or UDP_DATAGRAM_MAX))
end
-- TODO: is this DNS related? hence blocking?
_skt_mt_udp.__index.setpeername = function(self, ...) return self.socket:setpeername(...) end
_skt_mt_udp.__index.setsockname = function(self, ...) return self.socket:setsockname(...) end
-- do not close client, as it is also the server for udp.
_skt_mt_udp.__index.close = function(self, ...) return true end
_skt_mt_udp.__index.settimeouts = function (self, connect, send, receive)
return copas.settimeouts(self.socket, connect, send, receive)
end
---
-- Wraps a LuaSocket socket object in an async Copas based socket object.
-- @param skt The socket to wrap
-- @sslt (optional) Table with ssl parameters, use an empty table to use ssl with defaults
-- @return wrapped socket object
function copas.wrap (skt, sslt)
if (getmetatable(skt) == _skt_mt_tcp) or (getmetatable(skt) == _skt_mt_udp) then
return skt -- already wrapped
end
skt:settimeout(0)
if isTCP(skt) then
return setmetatable ({socket = skt, ssl_params = normalize_sslt(sslt)}, _skt_mt_tcp)
else
return setmetatable ({socket = skt}, _skt_mt_udp)
end
end
--- Wraps a handler in a function that deals with wrapping the socket and doing the
-- optional ssl handshake.
function copas.handler(handler, sslparams)
-- TODO: pass a timeout value to set, and use during handshake
return function (skt, ...)
skt = copas.wrap(skt, sslparams) -- this call will normalize the sslparams table
local sslp = skt.ssl_params
if sslp.sni then skt:sni(sslp.sni.names, sslp.sni.strict) end
if sslp.wrap then skt:dohandshake(sslp.wrap) end
return handler(skt, ...)
end
end
--------------------------------------------------
-- Error handling
--------------------------------------------------
local _errhandlers = setmetatable({}, { __mode = "k" }) -- error handler per coroutine
function copas.gettraceback(msg, co, skt)
local co_str = co == nil and "nil" or copas.getthreadname(co)
local skt_str = skt == nil and "nil" or copas.getsocketname(skt)
local msg_str = msg == nil and "" or tostring(msg)
if msg_str == "" then
msg_str = ("(coroutine: %s, socket: %s)"):format(msg_str, co_str, skt_str)
else
msg_str = ("%s (coroutine: %s, socket: %s)"):format(msg_str, co_str, skt_str)
end
if type(co) == "thread" then
-- regular Copas coroutine
return debug.traceback(co, msg_str)
end
-- not a coroutine, but the main thread, this happens if a timeout callback
-- (see `copas.timeout` causes an error (those callbacks run on the main thread).
return debug.traceback(msg_str, 2)
end
local function _deferror(msg, co, skt)
print(copas.gettraceback(msg, co, skt))
end
function copas.seterrorhandler(err, default)
assert(err == nil or type(err) == "function", "Expected the handler to be a function, or nil")
if default then
assert(err ~= nil, "Expected the handler to be a function when setting the default")
_deferror = err
else
_errhandlers[coroutine_running()] = err
end
end
copas.setErrorHandler = copas.seterrorhandler -- deprecated; old casing
function copas.geterrorhandler(co)
co = co or coroutine_running()
return _errhandlers[co] or _deferror
end
-- if `bool` is truthy, then the original socket errors will be returned in case of timeouts;
-- `timeout, wantread, wantwrite, Operation already in progress`. If falsy, it will always
-- return `timeout`.
function copas.useSocketTimeoutErrors(bool)
useSocketTimeoutErrors[coroutine_running()] = not not bool -- force to a boolean
end
-------------------------------------------------------------------------------
-- Thread handling
-------------------------------------------------------------------------------
local function _doTick (co, skt, ...)
if not co then return end
-- if a coroutine was canceled/removed, don't resume it
if _canceled[co] then
_canceled[co] = nil -- also clean up the registry
_threads[co] = nil
return
end
-- res: the socket (being read/write on) or the time to sleep
-- new_q: either _writing, _reading, or _sleeping
-- local time_before = gettime()
local ok, res, new_q = coroutine_resume(co, skt, ...)
-- local duration = gettime() - time_before
-- if duration > 1 then
-- duration = math.floor(duration * 1000)
-- pcall(_errhandlers[co] or _deferror, "task ran for "..tostring(duration).." milliseconds.", co, skt)
-- end
if new_q == _reading or new_q == _writing or new_q == _sleeping then
-- we're yielding to a new queue
new_q:insert (res)
new_q:push (res, co)
return
end
-- coroutine is terminating
if ok and coroutine_status(co) ~= "dead" then
-- it called coroutine.yield from a non-Copas function which is unexpected
ok = false
res = "coroutine.yield was called without a resume first, user-code cannot yield to Copas"
end
if not ok then
local k, e = pcall(_errhandlers[co] or _deferror, res, co, skt)
if not k then
print("Failed executing error handler: " .. tostring(e))
end
end
local skt_to_close = _autoclose[co]
if skt_to_close then
skt_to_close:close()
_autoclose[co] = nil
_autoclose_r[skt_to_close] = nil
end
_errhandlers[co] = nil
end
local _accept do
local client_counters = setmetatable({}, { __mode = "k" })
-- accepts a connection on socket input
function _accept(server_skt, handler)
local client_skt = server_skt:accept()
if client_skt then
local count = (client_counters[server_skt] or 0) + 1
client_counters[server_skt] = count
object_names[client_skt] = object_names[server_skt] .. ":client_" .. count
client_skt:settimeout(0)
copas.settimeouts(client_skt, user_timeouts_connect[server_skt], -- copy server socket timeout settings
user_timeouts_send[server_skt], user_timeouts_receive[server_skt])
local co = coroutine_create(handler)
object_names[co] = object_names[server_skt] .. ":handler_" .. count
if copas.autoclose then
_autoclose[co] = client_skt
_autoclose_r[client_skt] = co
end
_doTick(co, client_skt)
end
end
end
-------------------------------------------------------------------------------
-- Adds a server/handler pair to Copas dispatcher
-------------------------------------------------------------------------------
do
local function addTCPserver(server, handler, timeout, name)
server:settimeout(0)
if name then
object_names[server] = name
end
_servers[server] = handler
_reading:insert(server)
if timeout then
copas.settimeout(server, timeout)
end
end
local function addUDPserver(server, handler, timeout, name)
server:settimeout(0)
local co = coroutine_create(handler)
if name then
object_names[server] = name
end
object_names[co] = object_names[server]..":handler"
_reading:insert(server)
if timeout then
copas.settimeout(server, timeout)
end
_doTick(co, server)
end
function copas.addserver(server, handler, timeout, name)
if isTCP(server) then
addTCPserver(server, handler, timeout, name)
else
addUDPserver(server, handler, timeout, name)
end
end
end
function copas.removeserver(server, keep_open)
local skt = server
local mt = getmetatable(server)
if mt == _skt_mt_tcp or mt == _skt_mt_udp then
skt = server.socket
end
_servers:remove(skt)
_reading:remove(skt)
if keep_open then
return true
end
return server:close()
end
-------------------------------------------------------------------------------
-- Adds an new coroutine thread to Copas dispatcher
-------------------------------------------------------------------------------
function copas.addnamedthread(name, handler, ...)
if type(name) == "function" and type(handler) == "string" then
-- old call, flip args for compatibility
name, handler = handler, name
end
-- create a coroutine that skips the first argument, which is always the socket
-- passed by the scheduler, but `nil` in case of a task/thread
local thread = coroutine_create(function(_, ...)
copas.pause()
return handler(...)
end)
if name then
object_names[thread] = name
end
_threads[thread] = true -- register this thread so it can be removed
_doTick (thread, nil, ...)
return thread
end
function copas.addthread(handler, ...)
return copas.addnamedthread(nil, handler, ...)
end
function copas.removethread(thread)
-- if the specified coroutine is registered, add it to the canceled table so
-- that next time it tries to resume it exits.
_canceled[thread] = _threads[thread or 0]
_sleeping:cancel(thread)
end
-------------------------------------------------------------------------------
-- Sleep/pause management functions
-------------------------------------------------------------------------------
-- yields the current coroutine and wakes it after 'sleeptime' seconds.
-- If sleeptime < 0 then it sleeps until explicitly woken up using 'wakeup'
-- TODO: deprecated, remove in next major
function copas.sleep(sleeptime)
coroutine_yield((sleeptime or 0), _sleeping)
end
-- yields the current coroutine and wakes it after 'sleeptime' seconds.
-- if sleeptime < 0 then it sleeps 0 seconds.
function copas.pause(sleeptime)
if sleeptime and sleeptime > 0 then
coroutine_yield(sleeptime, _sleeping)
else
coroutine_yield(0, _sleeping)
end
end
-- patch luasystem to use copas.pause instead of system_sleep
if package.loaded["system"] then
package.loaded["system"].sleep = copas.pause
end
-- yields the current coroutine until explicitly woken up using 'wakeup'
function copas.pauseforever()
coroutine_yield(-1, _sleeping)
end
-- Wakes up a sleeping coroutine 'co'.
function copas.wakeup(co)
_sleeping:wakeup(co)
end
-------------------------------------------------------------------------------
-- Timeout management
-------------------------------------------------------------------------------
do
local timeout_register = setmetatable({}, { __mode = "k" })
local time_out_thread
local timerwheel = require("timerwheel").new({
now = gettime,
precision = TIMEOUT_PRECISION,
ringsize = math.floor(60*60*24/TIMEOUT_PRECISION), -- ring size 1 day
err_handler = function(err)
return _deferror(err, time_out_thread)
end,
})
time_out_thread = copas.addnamedthread("copas_core_timer", function()
while true do
copas.pause(TIMEOUT_PRECISION)
timerwheel:step()
end
end)
-- get the number of timeouts running
function copas.gettimeouts()
return timerwheel:count()
end
--- Sets the timeout for the current coroutine.
-- @param delay delay (seconds), use 0 (or math.huge) to cancel the timerout
-- @param callback function with signature: `function(coroutine)` where coroutine is the routine that timed-out
-- @return true
function copas.timeout(delay, callback)
local co = coroutine_running()
local existing_timer = timeout_register[co]
if existing_timer then
timerwheel:cancel(existing_timer)
end
if delay > 0 and delay ~= math.huge then
timeout_register[co] = timerwheel:set(delay, callback, co)
elseif delay == 0 or delay == math.huge then
timeout_register[co] = nil
else
error("timout value must be greater than or equal to 0, got: "..tostring(delay))
end
return true
end
end
-------------------------------------------------------------------------------
-- main tasks: manage readable and writable socket sets
-------------------------------------------------------------------------------
-- a task is an object with a required method `step()` that deals with a
-- single step for that task.
local _tasks = {} do
function _tasks:add(tsk)
_tasks[#_tasks + 1] = tsk
end
end
-- a task to check ready to read events
local _readable_task = {} do
_readable_task._events = {}
local function tick(skt)
local handler = _servers[skt]
if handler then
_accept(skt, handler)
else
_reading:remove(skt)
_doTick(_reading:pop(skt), skt)
end
end
function _readable_task:step()
for _, skt in ipairs(self._events) do
tick(skt)
end
end
_tasks:add(_readable_task)
end
-- a task to check ready to write events
local _writable_task = {} do
_writable_task._events = {}
local function tick(skt)
_writing:remove(skt)
_doTick(_writing:pop(skt), skt)
end
function _writable_task:step()
for _, skt in ipairs(self._events) do
tick(skt)
end
end
_tasks:add(_writable_task)
end
-- sleeping threads task
local _sleeping_task = {} do
function _sleeping_task:step()
local now = gettime()
local co = _sleeping:pop(now)
while co do
-- we're pushing them to _resumable, since that list will be replaced before
-- executing. This prevents tasks running twice in a row with pause(0) for example.
-- So here we won't execute, but at _resumable step which is next
_resumable:push(co)
co = _sleeping:pop(now)
end
end
_tasks:add(_sleeping_task)
end
-- resumable threads task
local _resumable_task = {} do
function _resumable_task:step()
-- replace the resume list before iterating, so items placed in there
-- will indeed end up in the next copas step, not in this one, and not
-- create a loop
local resumelist = _resumable:clear_resumelist()
for _, co in ipairs(resumelist) do
_doTick(co)
end
end
_tasks:add(_resumable_task)
end
-------------------------------------------------------------------------------
-- Checks for reads and writes on sockets
-------------------------------------------------------------------------------
local _select_plain do
local last_cleansing = 0
local duration = function(t2, t1) return t2-t1 end
if not socket then
-- socket module unavailable, switch to luasystem sleep
_select_plain = system_sleep
else
-- use socket.select to handle socket-io
_select_plain = function(timeout)
local err
local now = gettime()
-- remove any closed sockets to prevent select from hanging on them
if _closed[1] then
for i, skt in ipairs(_closed) do
_closed[i] = { _reading:remove(skt), _writing:remove(skt) }
end
end
_readable_task._events, _writable_task._events, err = socket.select(_reading, _writing, timeout)
local r_events, w_events = _readable_task._events, _writable_task._events
-- inject closed sockets in readable/writeable task so they can error out properly
if _closed[1] then
for i, skts in ipairs(_closed) do
_closed[i] = nil
r_events[#r_events+1] = skts[1]
w_events[#w_events+1] = skts[2]
end
end
if duration(now, last_cleansing) > WATCH_DOG_TIMEOUT then
last_cleansing = now
-- Check all sockets selected for reading, and check how long they have been waiting
-- for data already, without select returning them as readable
for skt,time in pairs(_reading_log) do
if not r_events[skt] and duration(now, time) > WATCH_DOG_TIMEOUT then
-- This one timedout while waiting to become readable, so move
-- it in the readable list and try and read anyway, despite not
-- having been returned by select
_reading_log[skt] = nil
r_events[#r_events + 1] = skt
r_events[skt] = #r_events
end
end
-- Do the same for writing
for skt,time in pairs(_writing_log) do
if not w_events[skt] and duration(now, time) > WATCH_DOG_TIMEOUT then
_writing_log[skt] = nil
w_events[#w_events + 1] = skt
w_events[skt] = #w_events
end
end
end
if err == "timeout" and #r_events + #w_events > 0 then
return nil
else
return err
end
end
end
end
-------------------------------------------------------------------------------
-- Dispatcher loop step.
-- Listen to client requests and handles them
-- Returns false if no socket-data was handled, or true if there was data
-- handled (or nil + error message)
-------------------------------------------------------------------------------
local copas_stats
local min_ever, max_ever
local _select = _select_plain
-- instrumented version of _select() to collect stats
local _select_instrumented = function(timeout)
if copas_stats then
local step_duration = gettime() - copas_stats.step_start
copas_stats.duration_max = math.max(copas_stats.duration_max, step_duration)
copas_stats.duration_min = math.min(copas_stats.duration_min, step_duration)
copas_stats.duration_tot = copas_stats.duration_tot + step_duration
copas_stats.steps = copas_stats.steps + 1
else
copas_stats = {
duration_max = -1,
duration_min = 999999,
duration_tot = 0,
steps = 0,
}
end
local err = _select_plain(timeout)
local now = gettime()
copas_stats.time_start = copas_stats.time_start or now
copas_stats.step_start = now
return err
end
function copas.step(timeout)
-- Need to wake up the select call in time for the next sleeping event
if not _resumable:done() then
timeout = 0
else
timeout = math.min(_sleeping:getnext(), timeout or math.huge)
end
local err = _select(timeout)
for _, tsk in ipairs(_tasks) do
tsk:step()
end
if err then
if err == "timeout" then
if timeout + 0.01 > TIMEOUT_PRECISION and math.random(100) > 90 then
-- we were idle, so occasionally do a GC sweep to ensure lingering
-- sockets are closed, and we don't accidentally block the loop from
-- exiting
collectgarbage()
end
return false
end
return nil, err
end
return true
end
-------------------------------------------------------------------------------
-- Check whether there is something to do.
-- returns false if there are no sockets for read/write nor tasks scheduled
-- (which means Copas is in an empty spin)
-------------------------------------------------------------------------------
function copas.finished()
return #_reading == 0 and #_writing == 0 and _resumable:done() and _sleeping:done(copas.gettimeouts())
end
local resetexit do
local exit_semaphore, exiting
function resetexit()
exit_semaphore = copas.semaphore.new(1, 0, math.huge)
exiting = false
end
-- Signals tasks to exit. But only if they check for it. By calling `copas.exiting`
-- they can check if they should exit. Or by calling `copas.waitforexit` they can
-- wait until the exit signal is given.
function copas.exit()
if exiting then return end
exiting = true
exit_semaphore:destroy()
end
-- returns whether Copas is in the process of exiting. Exit can be started by
-- calling `copas.exit()`.
function copas.exiting()
return exiting
end
-- Pauses the current coroutine until Copas is exiting. To be used as an exit
-- signal for tasks that need to clean up before exiting.
function copas.waitforexit()
exit_semaphore:take(1)
end
end
local _getstats do
local _getstats_instrumented, _getstats_plain
function _getstats_plain(enable)
-- this function gets hit if turned off, so turn on if true
if enable == true then
_select = _select_instrumented
_getstats = _getstats_instrumented
-- reset stats
min_ever = nil
max_ever = nil
copas_stats = nil
end
return {}
end
-- convert from seconds to millisecs, with microsec precision
local function useconds(t)
return math.floor((t * 1000000) + 0.5) / 1000
end
-- convert from seconds to seconds, with millisec precision
local function mseconds(t)
return math.floor((t * 1000) + 0.5) / 1000
end
function _getstats_instrumented(enable)
if enable == false then
_select = _select_plain
_getstats = _getstats_plain
-- instrumentation disabled, so switch to the plain implementation
return _getstats(enable)
end
if (not copas_stats) or (copas_stats.step == 0) then
return {}
end
local stats = copas_stats
copas_stats = nil
min_ever = math.min(min_ever or 9999999, stats.duration_min)
max_ever = math.max(max_ever or 0, stats.duration_max)
stats.duration_min_ever = min_ever
stats.duration_max_ever = max_ever
stats.duration_avg = stats.duration_tot / stats.steps
stats.step_start = nil
stats.time_end = gettime()
stats.time_tot = stats.time_end - stats.time_start
stats.time_avg = stats.time_tot / stats.steps
stats.duration_avg = useconds(stats.duration_avg)
stats.duration_max = useconds(stats.duration_max)
stats.duration_max_ever = useconds(stats.duration_max_ever)
stats.duration_min = useconds(stats.duration_min)
stats.duration_min_ever = useconds(stats.duration_min_ever)
stats.duration_tot = useconds(stats.duration_tot)
stats.time_avg = useconds(stats.time_avg)
stats.time_start = mseconds(stats.time_start)
stats.time_end = mseconds(stats.time_end)
stats.time_tot = mseconds(stats.time_tot)
return stats
end
_getstats = _getstats_plain
end
function copas.status(enable_stats)
local res = _getstats(enable_stats)
res.running = not not copas.running
res.timeout = copas.gettimeouts()
res.timer, res.inactive = _sleeping:status()
res.read = #_reading
res.write = #_writing
res.active = _resumable:count()
return res
end
-------------------------------------------------------------------------------
-- Dispatcher endless loop.
-- Listen to client requests and handles them forever
-------------------------------------------------------------------------------
function copas.loop(initializer, timeout)
if type(initializer) == "function" then
copas.addnamedthread("copas_initializer", initializer)
else
timeout = initializer or timeout
end
resetexit()
copas.running = true
while true do
copas.step(timeout)
if copas.finished() then
if copas.exiting() then
break
end
copas.exit()
end
end
copas.running = false
end
-------------------------------------------------------------------------------
-- Naming sockets and coroutines.
-------------------------------------------------------------------------------
do
local function realsocket(skt)
local mt = getmetatable(skt)
if mt == _skt_mt_tcp or mt == _skt_mt_udp then
return skt.socket
else
return skt
end
end
function copas.setsocketname(name, skt)
assert(type(name) == "string", "expected arg #1 to be a string")
skt = assert(realsocket(skt), "expected arg #2 to be a socket")
object_names[skt] = name
end
function copas.getsocketname(skt)
skt = assert(realsocket(skt), "expected arg #1 to be a socket")
return object_names[skt]
end
end
function copas.setthreadname(name, coro)
assert(type(name) == "string", "expected arg #1 to be a string")
coro = coro or coroutine_running()
assert(type(coro) == "thread", "expected arg #2 to be a coroutine or nil")
object_names[coro] = name
end
function copas.getthreadname(coro)
coro = coro or coroutine_running()
assert(type(coro) == "thread", "expected arg #1 to be a coroutine or nil")
return object_names[coro]
end
-------------------------------------------------------------------------------
-- Debug functionality.
-------------------------------------------------------------------------------
do
copas.debug = {}
local log_core -- if truthy, the core-timer will also be logged
local debug_log -- function used as logger
local debug_yield = function(skt, queue)
local name = object_names[coroutine_running()]
if log_core or name ~= "copas_core_timer" then
if queue == _sleeping then
debug_log("yielding '", name, "' to SLEEP for ", skt," seconds")
elseif queue == _writing then
debug_log("yielding '", name, "' to WRITE on '", object_names[skt], "'")
elseif queue == _reading then
debug_log("yielding '", name, "' to READ on '", object_names[skt], "'")
else
debug_log("thread '", name, "' yielding to unexpected queue; ", tostring(queue), " (", type(queue), ")", debug.traceback())
end
end
return coroutine.yield(skt, queue)
end
local debug_resume = function(coro, skt, ...)
local name = object_names[coro]
if skt then
debug_log("resuming '", name, "' for socket '", object_names[skt], "'")
else
if log_core or name ~= "copas_core_timer" then
debug_log("resuming '", name, "'")
end
end
return coroutine.resume(coro, skt, ...)
end
local debug_create = function(f)
local f_wrapped = function(...)
local results = pack(f(...))
debug_log("exiting '", object_names[coroutine_running()], "'")
return unpack(results)
end
return coroutine.create(f_wrapped)
end
debug_log = fnil
-- enables debug output for all coroutine operations.
function copas.debug.start(logger, core)
log_core = core
debug_log = logger or print
coroutine_yield = debug_yield
coroutine_resume = debug_resume
coroutine_create = debug_create
end
-- disables debug output for coroutine operations.
function copas.debug.stop()
debug_log = fnil
coroutine_yield = coroutine.yield
coroutine_resume = coroutine.resume
coroutine_create = coroutine.create
end
do
local call_id = 0
-- Description table of socket functions for debug output.
-- each socket function name has TWO entries;
-- 'name_in' and 'name_out', each being an array of names/descriptions of respectively
-- input parameters and return values.
-- If either table has a 'callback' key, then that is a function that will be called
-- with the parameters/return-values for further inspection.
local args = {
settimeout_in = {
"socket ",
"seconds",
"mode ",
},
settimeout_out = {
"success",
"error ",
},
connect_in = {
"socket ",
"address",
"port ",
},
connect_out = {
"success",
"error ",
},
getfd_in = {
"socket ",
-- callback = function(...)
-- print(debug.traceback("called from:", 4))
-- end,
},
getfd_out = {
"fd",
},
send_in = {
"socket ",
"data ",
"idx-start",
"idx-end ",
},
send_out = {
"last-idx-send ",
"error ",
"err-last-idx-send",
},
receive_in = {
"socket ",
"pattern",
"prefix ",
},
receive_out = {
"received ",
"error ",
"partial data",
},
dirty_in = {
"socket",
-- callback = function(...)
-- print(debug.traceback("called from:", 4))
-- end,
},
dirty_out = {
"data in read-buffer",
},
close_in = {
"socket",
-- callback = function(...)
-- print(debug.traceback("called from:", 4))
-- end,
},
close_out = {
"success",
"error",
},
}
local function print_call(func, msg, ...)
print(msg)
local arg = pack(...)
local desc = args[func] or {}
for i = 1, math.max(arg.n, #desc) do
local value = arg[i]
if type(value) == "string" then
local xvalue = value:sub(1,30)
if xvalue ~= value then
xvalue = xvalue .."(...truncated)"
end
print("\t"..(desc[i] or i)..": '"..tostring(xvalue).."' ("..type(value).." #"..#value..")")
else
print("\t"..(desc[i] or i)..": '"..tostring(value).."' ("..type(value)..")")
end
end
if desc.callback then
desc.callback(...)
end
end
local debug_mt = {
__index = function(self, key)
local value = self.__original_socket[key]
if type(value) ~= "function" then
return value
end
return function(self2, ...)
local my_id = call_id + 1
call_id = my_id
local results
if self2 ~= self then
-- there is no self
print_call(tostring(key).."_in", my_id .. "-calling '"..tostring(key) .. "' with; ", self, ...)
results = pack(value(self, ...))
else
print_call(tostring(key).."_in", my_id .. "-calling '" .. tostring(key) .. "' with; ", self.__original_socket, ...)
results = pack(value(self.__original_socket, ...))
end
print_call(tostring(key).."_out", my_id .. "-results '"..tostring(key) .. "' returned; ", unpack(results))
return unpack(results)
end
end,
__tostring = function(self)
return tostring(self.__original_socket)
end
}
-- wraps a socket (copas or luasocket) in a debug version printing all calls
-- and their parameters/return values. Extremely noisy!
-- returns the wrapped socket.
-- NOTE: only for plain sockets, will not support TLS
function copas.debug.socket(original_skt)
if (getmetatable(original_skt) == _skt_mt_tcp) or (getmetatable(original_skt) == _skt_mt_udp) then
-- already wrapped as Copas socket, so recurse with the original luasocket one
original_skt.socket = copas.debug.socket(original_skt.socket)
return original_skt
end
local proxy = setmetatable({
__original_socket = original_skt
}, debug_mt)
return proxy
end
end
end
return copas