Fix. Init socket callback with nil.
Fix. `multi.setopt` supports socket callback. Change. `socket_acttion` without args means timeout action. Change. cURLv3 multi supports options in callback. Add. `multi-uv` example.master
parent
619fb1b08d
commit
e7712b9eb2
|
@ -0,0 +1,162 @@
|
|||
local curl = require "cURL"
|
||||
local uv = require "lluv"
|
||||
|
||||
local fprintf = function(f, ...) f:write((string.format(...))) end
|
||||
local printf = function(...) fprintf(io.stdout, ...) end
|
||||
|
||||
local stderr = io.stderr
|
||||
|
||||
local timeout, curl_handle
|
||||
|
||||
local ACTION_NAMES = {
|
||||
[curl.POLL_IN ] = "POLL_IN";
|
||||
[curl.POLL_INOUT ] = "POLL_INOUT";
|
||||
[curl.POLL_OUT ] = "POLL_OUT";
|
||||
[curl.POLL_NONE ] = "POLL_NONE";
|
||||
[curl.POLL_REMOVE ] = "POLL_REMOVE";
|
||||
}
|
||||
local EVENT_NAMES = {
|
||||
[ uv.READABLE ] = "READABLE";
|
||||
[ uv.WRITABLE ] = "WRITABLE";
|
||||
[ uv.READABLE + uv.WRITABLE ] = "READABLE + WRITABLE";
|
||||
}
|
||||
local FLAGS = {
|
||||
[ uv.READABLE ] = curl.CSELECT_IN;
|
||||
[ uv.WRITABLE ] = curl.CSELECT_OUT;
|
||||
[ uv.READABLE + uv.WRITABLE ] = curl.CSELECT_IN + curl.CSELECT_OUT;
|
||||
|
||||
}
|
||||
|
||||
local trace = function() end or print
|
||||
|
||||
local FILES, CONTEXT = {}, {}
|
||||
|
||||
function create_curl_context(sockfd)
|
||||
local context = {
|
||||
sockfd = sockfd;
|
||||
poll_handle = uv.poll_socket(sockfd);
|
||||
}
|
||||
context.poll_handle.data = context
|
||||
|
||||
return context
|
||||
end
|
||||
|
||||
function destroy_curl_context(context)
|
||||
context.poll_handle:close()
|
||||
end
|
||||
|
||||
function add_download(url, num)
|
||||
local filename = tostring(num) .. ".download"
|
||||
local file = io.open(filename, "w")
|
||||
if not file then
|
||||
fprintf(stderr, "Error opening %s\n", filename)
|
||||
return
|
||||
end
|
||||
|
||||
local handle = curl.easy{
|
||||
url = url;
|
||||
writefunction = file;
|
||||
}
|
||||
|
||||
FILES[handle] = file
|
||||
|
||||
curl_handle:add_handle(handle)
|
||||
fprintf(stderr, "Added download %s -> %s\n", url, filename);
|
||||
end
|
||||
|
||||
function check_multi_info()
|
||||
while true do
|
||||
local easy, ok, err = curl_handle:info_read()
|
||||
if not easy then curl_handle:close() error(err) end
|
||||
if easy == 0 then break end
|
||||
|
||||
local context = CONTEXT[e]
|
||||
if context then destroy_curl_context(context) end
|
||||
local file = FILES[easy]
|
||||
if file then FILES[easy] = nil, file:close() end
|
||||
local done_url = easy:getinfo_effective_url()
|
||||
easy:close()
|
||||
if ok then
|
||||
printf("%s DONE\n", done_url);
|
||||
elseif data == "error" then
|
||||
printf("%s ERROR - %s\n", done_url, tostring(err));
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
function curl_perform(handle, err, events)
|
||||
-- calls by libuv --
|
||||
trace("UV::POLL", handle, err, EVENT_NAMES[events] or events)
|
||||
|
||||
local flags = assert(FLAGS[events], ("unknown event:" .. events))
|
||||
|
||||
context = handle.data
|
||||
|
||||
curl_handle:socket_action(context.sockfd, flags)
|
||||
|
||||
check_multi_info()
|
||||
end
|
||||
|
||||
function on_timeout(timer)
|
||||
-- calls by libuv --
|
||||
trace("UV::TIMEOUT", timer)
|
||||
|
||||
local running_handles, err = curl_handle:socket_action()
|
||||
|
||||
check_multi_info()
|
||||
end
|
||||
|
||||
function start_timeout(timeout_ms)
|
||||
-- calls by curl --
|
||||
trace("CURL::TIMEOUT", timeout_ms)
|
||||
|
||||
-- 0 means directly call socket_action, but we'll do it in a bit
|
||||
if timeout_ms <= 0 then timeout_ms = 1 end
|
||||
|
||||
timeout:stop():start(timeout_ms, 0, on_timeout)
|
||||
end
|
||||
|
||||
local handle_socket = function(...)
|
||||
local ok, err = pcall(handle_socket_impl, ...)
|
||||
if not ok then uv.defer(function() error(err) end) end
|
||||
end
|
||||
|
||||
function handle_socket_impl(easy, s, action)
|
||||
-- calls by curl --
|
||||
|
||||
trace("CURL::SOCKET", easy, s, ACTION_NAMES[action] or action)
|
||||
|
||||
local curl_context = CONTEXT[easy] or create_curl_context(s)
|
||||
CONTEXT[easy] = curl_context
|
||||
|
||||
assert(curl_context.sockfd == s)
|
||||
|
||||
if action == curl.POLL_IN then
|
||||
curl_context.poll_handle:start(uv.READABLE, curl_perform)
|
||||
elseif action == curl.POLL_OUT then
|
||||
curl_context.poll_handle:start(uv.WRITABLE, curl_perform)
|
||||
elseif action == curl.POLL_REMOVE then
|
||||
CONTEXT[easy] = nil
|
||||
destroy_curl_context(curl_context)
|
||||
end
|
||||
end
|
||||
|
||||
timeout = uv.timer()
|
||||
|
||||
curl_handle = curl.multi{
|
||||
socketfunction = handle_socket;
|
||||
timerfunction = start_timeout;
|
||||
}
|
||||
|
||||
curl_handle = curl.multi{
|
||||
socketfunction = handle_socket;
|
||||
timerfunction = start_timeout;
|
||||
}
|
||||
|
||||
for i = 1, math.huge do
|
||||
local url = arg[i]
|
||||
if not url then break end
|
||||
add_download(url, i)
|
||||
end
|
||||
|
||||
uv.run(loop, UV_RUN_DEFAULT)
|
|
@ -43,6 +43,7 @@ int lcurl_multi_create(lua_State *L, int error_mode){
|
|||
lcurl_util_new_weak_table(L, "v");
|
||||
p->h_ref = luaL_ref(L, LCURL_LUA_REGISTRY);
|
||||
p->tm.cb_ref = p->tm.ud_ref = LUA_NOREF;
|
||||
p->sc.cb_ref = p->sc.ud_ref = LUA_NOREF;
|
||||
|
||||
if(lua_type(L, 1) == LUA_TTABLE){
|
||||
int ret = lcurl_utils_apply_options(L, 1, 2, 1, p->err_mode, LCURL_ERROR_MULTI, CURLM_UNKNOWN_OPTION);
|
||||
|
@ -228,9 +229,11 @@ static int lcurl_multi_timeout(lua_State *L){
|
|||
|
||||
static int lcurl_multi_socket_action(lua_State *L){
|
||||
lcurl_multi_t *p = lcurl_getmulti(L);
|
||||
curl_socket_t s = lutil_checkint64(L, 2);
|
||||
int n, mask = lutil_checkint64(L, 3);
|
||||
CURLMcode code = curl_multi_socket_action(p->curl, s, mask, &n);
|
||||
curl_socket_t s = lutil_optint64(L, 2, CURL_SOCKET_TIMEOUT);
|
||||
CURLMcode code; int n, mask;
|
||||
if(s == CURL_SOCKET_TIMEOUT) mask = 0;
|
||||
else mask = lutil_checkint64(L, 3);
|
||||
code = curl_multi_socket_action(p->curl, s, mask, &n);
|
||||
if(code != CURLM_OK){
|
||||
lcurl_fail_ex(L, p->err_mode, LCURL_ERROR_MULTI, code);
|
||||
}
|
||||
|
@ -427,7 +430,8 @@ static int lcurl_multi_setopt(lua_State *L){
|
|||
#define OPT_ENTRY(l, N, T, S) case CURLMOPT_##N: return lcurl_multi_set_##N(L);
|
||||
switch(opt){
|
||||
#include "lcoptmulti.h"
|
||||
OPT_ENTRY(timerfunction, TIMERFUNCTION, TTT, 0)
|
||||
OPT_ENTRY(timerfunction, TIMERFUNCTION, TTT, 0)
|
||||
OPT_ENTRY(socketfunction, SOCKETFUNCTION, TTT, 0)
|
||||
}
|
||||
#undef OPT_ENTRY
|
||||
|
||||
|
|
|
@ -520,8 +520,9 @@ local Multi = class(curl.multi) do
|
|||
local add_handle = wrap_function("add_handle")
|
||||
local remove_handle = wrap_function("remove_handle")
|
||||
|
||||
function Multi:__init()
|
||||
function Multi:__init(opt)
|
||||
self._easy = {n = 0}
|
||||
if opt then self:setopt(opt) end
|
||||
return self
|
||||
end
|
||||
|
||||
|
|
Loading…
Reference in New Issue