Merge pull request #46 from moteus/master

Support `socket_action` interface.

doc/lcurl.ldoc

@@ -24,6 +24,13 @@ function easy() end
 --
 -- @tparam[opt] table options
 -- @treturn[1] multi new curl multi object
+--
+-- @usage
+-- m = curl.multi{
+--   socketfunction = handle_socket;
+--   timerfunction  = start_timeout;
+-- }
+--
 function multi() end
 
 --- Create Share object
@@ -433,6 +440,18 @@ function info_read() end
 -- c:setopt{maxconnects = 10}
 function setopt() end
 
+--- Perform socket action.
+--
+-- @tparam[opt=curl.SOCKET_TIMEOUT] number socket
+-- @tparam[opt=0] number mask
+-- @treturn multi self
+--
+-- @usage
+-- c:socket_action()
+-- c:socket_action(sock_fd, curl.CSELECT_IN)
+-- c:socket_action(sock_fd, curl.CSELECT_OUT)
+function socket_action() end
+
 --- Set timer callback.
 --
 -- @tparam function timer timer callback

examples/cURLv3/multi-uv.lua

@@ -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)

src/l52util.c

@@ -155,3 +155,27 @@ void *lutil_newudatap_impl(lua_State *L, size_t size, const void *p){
   lutil_setmetatablep(L, p);
   return obj;
 }
+
+void lutil_pushint64(lua_State *L, int64_t v){
+  if(sizeof(lua_Integer) >= sizeof(int64_t)){
+    lua_pushinteger(L, (lua_Integer)v);
+    return;
+  }
+  lua_pushnumber(L, (lua_Number)v);
+}
+
+int64_t lutil_checkint64(lua_State *L, int idx){
+  if(sizeof(lua_Integer) >= sizeof(int64_t))
+    return luaL_checkinteger(L, idx);
+  return (int64_t)luaL_checknumber(L, idx);
+}
+
+int64_t lutil_optint64(lua_State *L, int idx, int64_t v){
+  if(sizeof(lua_Integer) >= sizeof(int64_t))
+    return luaL_optinteger(L, idx, v);
+  return (int64_t)luaL_optnumber(L, idx, v);
+}
+
+void lutil_pushnvalues(lua_State *L, int n){
+  for(;n;--n) lua_pushvalue(L, -n);
+}

src/l52util.h

@@ -13,47 +13,52 @@
 
 #include "lua.h"
 #include "lauxlib.h"
+#include <stdint.h>
 
 #if LUA_VERSION_NUM >= 503 /* Lua 5.3 */
 
-#ifndef luaL_optint
-# define luaL_optint luaL_optinteger
-#endif
-
 #ifndef luaL_checkint
-# define luaL_checkint luaL_checkinteger
+#define luaL_checkint  luaL_checkinteger
 #endif
 
 #ifndef luaL_checklong
-# define luaL_checklong luaL_checkinteger
+#define luaL_checklong  luaL_checkinteger
+#endif
+
+#ifndef luaL_optint
+#define luaL_optint luaL_optinteger
+#endif
+
+#ifndef luaL_optlong
+#define luaL_optlong luaL_optinteger
 #endif
 
 #endif
 
+#if LUA_VERSION_NUM >= 502 /* Lua 5.2 */
 
-#if LUA_VERSION_NUM >= 502 // lua 5.2
+/* lua_rawgetp */
+/* lua_rawsetp */
+/* luaL_setfuncs */
+/* lua_absindex */
 
-// lua_rawgetp
-// lua_rawsetp
-// luaL_setfuncs
-// lua_absindex
 #ifndef lua_objlen
-
 #define lua_objlen      lua_rawlen
-
 #endif
 
 int   luaL_typerror (lua_State *L, int narg, const char *tname);
 
 #ifndef luaL_register
-
 void luaL_register (lua_State *L, const char *libname, const luaL_Reg *l);
-
 #endif
 
-#else                      // lua 5.1
+#ifndef lua_equal
+#define lua_equal(L,idx1,idx2) lua_compare(L,(idx1),(idx2),LUA_OPEQ)
+#endif
 
-// functions form lua 5.2
+#else                      /* Lua 5.1 */
+
+/* functions form lua 5.2 */
 
 # define lua_absindex(L, i) (((i)>0)?(i):((i)<=LUA_REGISTRYINDEX?(i):(lua_gettop(L)+(i)+1)))
 # define lua_rawlen  lua_objlen
@@ -78,5 +83,12 @@ int   lutil_createmetap   (lua_State *L, const void *p, const luaL_Reg *methods,
 
 void *lutil_newudatap_impl     (lua_State *L, size_t size, const void *p);
 
-#endif
+void lutil_pushint64(lua_State *L, int64_t v);
 
+int64_t lutil_checkint64(lua_State *L, int idx);
+
+int64_t lutil_optint64(lua_State *L, int idx, int64_t v);
+
+void lutil_pushnvalues(lua_State *L, int n);
+
+#endif

src/lcmulti.c

@@ -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);
@@ -73,7 +74,10 @@ static int lcurl_multi_cleanup(lua_State *L){
 
   luaL_unref(L, LCURL_LUA_REGISTRY, p->tm.cb_ref);
   luaL_unref(L, LCURL_LUA_REGISTRY, p->tm.ud_ref);
+  luaL_unref(L, LCURL_LUA_REGISTRY, p->sc.cb_ref);
+  luaL_unref(L, LCURL_LUA_REGISTRY, p->sc.ud_ref);
   p->tm.cb_ref = p->tm.ud_ref = LUA_NOREF;
+  p->sc.cb_ref = p->sc.ud_ref = LUA_NOREF;
   return 0;
 }
 
@@ -226,6 +230,20 @@ static int lcurl_multi_timeout(lua_State *L){
   return 1;
 }
 
+static int lcurl_multi_socket_action(lua_State *L){
+  lcurl_multi_t *p = lcurl_getmulti(L);
+  curl_socket_t s  = lutil_optint64(L, 2, CURL_SOCKET_TIMEOUT);
+  CURLMcode code; int n, mask;
+  if(s == CURL_SOCKET_TIMEOUT) mask = lutil_optint64(L, 3, 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);
+  }
+  lua_pushinteger(L, n);
+  return 1;
+}
+
 //{ OPTIONS
 static int lcurl_opt_set_long_(lua_State *L, int opt){
   lcurl_multi_t *p = lcurl_getmulti(L);
@@ -313,7 +331,7 @@ static int lcurl_multi_set_callback(lua_State *L,
   return 1;
 }
 
-//{Timer
+//{ Timer
 
 int lcurl_multi_timer_callback(CURLM *multi, long ms, void *arg){
   lcurl_multi_t *p = arg;
@@ -347,7 +365,6 @@ int lcurl_multi_timer_callback(CURLM *multi, long ms, void *arg){
   return ret;
 }
 
-
 static int lcurl_multi_set_TIMERFUNCTION(lua_State *L){
   lcurl_multi_t *p = lcurl_getmulti(L);
   return lcurl_multi_set_callback(L, p, &p->tm,
@@ -358,6 +375,43 @@ static int lcurl_multi_set_TIMERFUNCTION(lua_State *L){
 
 //}
 
+//{ Socket
+
+static int lcurl_multi_socket_callback(CURL *easy, curl_socket_t s, int what, void *arg, void *socketp){
+  lcurl_multi_t *p = arg;
+  lua_State *L = p->L;
+  lcurl_easy_t *e;
+  int n, top = lua_gettop(L);
+
+  n = lcurl_util_push_cb(L, &p->sc);
+
+  lua_rawgeti(L, LCURL_LUA_REGISTRY, p->h_ref);
+  lua_rawgetp(L, -1, easy);
+  e = lcurl_geteasy_at(L, -1);
+  lua_remove(L, -2);
+  lutil_pushint64(L, s);
+  lua_pushinteger(L, what);
+
+  if(lua_pcall(L, n+2, 0, 0)){
+    assert(lua_gettop(L) >= top);
+    lua_settop(L, top);
+    return -1; //! @todo break perform
+  }
+
+  lua_settop(L, top);
+  return 0;
+}
+
+static int lcurl_multi_set_SOCKETFUNCTION(lua_State *L){
+  lcurl_multi_t *p = lcurl_getmulti(L);
+  return lcurl_multi_set_callback(L, p, &p->sc,
+    CURLMOPT_SOCKETFUNCTION, CURLMOPT_SOCKETDATA,
+    "socket", lcurl_multi_socket_callback
+  );
+}
+
+//}
+
 //}
 
 static int lcurl_multi_setopt(lua_State *L){
@@ -380,6 +434,7 @@ static int lcurl_multi_setopt(lua_State *L){
   switch(opt){
     #include "lcoptmulti.h"
     OPT_ENTRY(timerfunction,  TIMERFUNCTION,  TTT, 0)
+    OPT_ENTRY(socketfunction, SOCKETFUNCTION, TTT, 0)
   }
 #undef OPT_ENTRY
 
@@ -396,10 +451,12 @@ static const struct luaL_Reg lcurl_multi_methods[] = {
   {"setopt",           lcurl_multi_setopt           },
   {"wait",             lcurl_multi_wait             },
   {"timeout",          lcurl_multi_timeout          },
+  {"socket_action",    lcurl_multi_socket_action    },
 
 #define OPT_ENTRY(L, N, T, S) { "setopt_"#L, lcurl_multi_set_##N },
   #include "lcoptmulti.h"
   OPT_ENTRY(timerfunction,  TIMERFUNCTION,  TTT, 0)
+  OPT_ENTRY(socketfunction, SOCKETFUNCTION, TTT, 0)
 #undef OPT_ENTRY
 
   {"close",            lcurl_multi_cleanup          },
@@ -412,6 +469,7 @@ static const lcurl_const_t lcurl_multi_opt[] = {
 #define OPT_ENTRY(L, N, T, S) { "OPT_MULTI_"#N, CURLMOPT_##N },
   #include "lcoptmulti.h"
   OPT_ENTRY(timerfunction,  TIMERFUNCTION,  TTT, 0)
+  OPT_ENTRY(socketfunction, SOCKETFUNCTION, TTT, 0)
 #undef OPT_ENTRY
 
   {NULL, 0}

src/lcmulti.h

@@ -20,6 +20,7 @@ typedef struct lcurl_multi_tag{
   int err_mode;
   int h_ref;
   lcurl_callback_t tm;
+  lcurl_callback_t sc;
 }lcurl_multi_t;
 
 int lcurl_multi_create(lua_State *L, int error_mode);

src/lcopteasy.h

@@ -264,6 +264,17 @@ FLG_ENTRY( HTTP_VERSION_2_0   )
 FLG_ENTRY( READFUNC_PAUSE     ) /*7.18.0*/
 FLG_ENTRY( WRITEFUNC_PAUSE    ) /*7.18.0*/
 
+FLG_ENTRY( POLL_IN            ) /*7.14.0*/
+FLG_ENTRY( POLL_INOUT         ) /*7.14.0*/
+FLG_ENTRY( POLL_NONE          ) /*7.14.0*/
+FLG_ENTRY( POLL_OUT           ) /*7.14.0*/
+FLG_ENTRY( POLL_REMOVE        ) /*7.14.0*/
+FLG_ENTRY( SOCKET_TIMEOUT     ) /*7.14.0*/
+
+FLG_ENTRY( CSELECT_ERR        ) /*7.16.3*/
+FLG_ENTRY( CSELECT_IN         ) /*7.16.3*/
+FLG_ENTRY( CSELECT_OUT        ) /*7.16.3*/
+
 #ifdef OPT_ENTRY_IS_NULL
 #  undef OPT_ENTRY
 #endif
@@ -271,3 +282,4 @@ FLG_ENTRY( WRITEFUNC_PAUSE    ) /*7.18.0*/
 #ifdef FLG_ENTRY_IS_NULL
 #  undef FLG_ENTRY
 #endif
+

src/lua/cURL/impl/cURL.lua

@@ -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