brejela
/
lua-zmq
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add support for zmq 3.0/3.1

Browse Source
zmq3.0
Robert G. Jakabosky 15 years ago
parent ffeb63cf7d
commit b1157e3aaa
2 changed files with 298 additions and 134 deletions
Show Stats Download Patch File Download Diff File

267
src/socket.nobj.lua
Unescape Escape View File

@ -21,18 +21,26 @@
object "ZMQ_Socket" {
error_on_null = "get_zmq_strerror()",
c_source [[
/* detect zmq version >= 2.1.0 */
/* detect zmq version */
#define VERSION_2_0 1
#define VERSION_2_1 0
#define VERSION_3_0 0
#if defined(ZMQ_VERSION)
#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION(2,1,0))
#undef VERSION_2_1
#define VERSION_2_1 1
#endif
# if (ZMQ_VERSION >= ZMQ_MAKE_VERSION(2,1,0))
# undef VERSION_2_1
# define VERSION_2_1 1
# endif
# if (ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0))
# undef VERSION_2_0
# define VERSION_2_0 0
# undef VERSION_3_0
# define VERSION_3_0 1
# endif
#endif
/* detect really old ZeroMQ 2.0.x series. */
#if !defined(ZMQ_RCVMORE)
#error "Your version of ZeroMQ is too old. Please upgrade to version 2.1 or to the latest 2.0.x"
# error "Your version of ZeroMQ is too old. Please upgrade to version 2.1 or to the latest 2.0.x"
#endif
typedef struct ZMQ_Socket ZMQ_Socket;
@ -44,24 +52,63 @@ typedef SOCKET socket_t;
typedef int socket_t;
#endif
#if ZMQ_VERSION_MAJOR == 2
# define zmq_sendmsg zmq_send
# define zmq_recvmsg zmq_recv
#endif
/* socket option types. */
#define OPT_TYPE_NONE 0
#define OPT_TYPE_INT 1
#define OPT_TYPE_UINT32 2
#define OPT_TYPE_UINT64 3
#define OPT_TYPE_INT64 4
#define OPT_TYPE_STR 5
#define OPT_TYPE_BLOB 5
#define OPT_TYPE_FD 6
static const int opt_types[] = {
#if VERSION_3_0
OPT_TYPE_NONE, /* 0 unused */
OPT_TYPE_INT, /* 1 ZMQ_HWM for backwards compatibility */
OPT_TYPE_NONE, /* 2 unused */
OPT_TYPE_NONE, /* 3 unused */
OPT_TYPE_UINT64, /* 4 ZMQ_AFFINITY */
OPT_TYPE_BLOB, /* 5 ZMQ_IDENTITY */
OPT_TYPE_BLOB, /* 6 ZMQ_SUBSCRIBE */
OPT_TYPE_BLOB, /* 7 ZMQ_UNSUBSCRIBE */
OPT_TYPE_INT, /* 8 ZMQ_RATE */
OPT_TYPE_INT, /* 9 ZMQ_RECOVERY_IVL */
OPT_TYPE_INT, /* 10 ZMQ_MCAST_LOOP */
OPT_TYPE_INT, /* 11 ZMQ_SNDBUF */
OPT_TYPE_INT, /* 12 ZMQ_RCVBUF */
OPT_TYPE_INT, /* 13 ZMQ_RCVMORE */
OPT_TYPE_FD, /* 14 ZMQ_FD */
OPT_TYPE_INT, /* 15 ZMQ_EVENTS */
OPT_TYPE_INT, /* 16 ZMQ_TYPE */
OPT_TYPE_INT, /* 17 ZMQ_LINGER */
OPT_TYPE_INT, /* 18 ZMQ_RECONNECT_IVL */
OPT_TYPE_INT, /* 19 ZMQ_BACKLOG */
OPT_TYPE_NONE, /* 20 unused */
OPT_TYPE_INT, /* 21 ZMQ_RECONNECT_IVL_MAX */
OPT_TYPE_INT64, /* 22 ZMQ_MAXMSGSIZE */
OPT_TYPE_INT, /* 23 ZMQ_SNDHWM */
OPT_TYPE_INT, /* 24 ZMQ_RCVHWM */
OPT_TYPE_INT, /* 25 ZMQ_MULTICAST_HOPS */
OPT_TYPE_NONE, /* 26 unused */
OPT_TYPE_INT, /* 27 ZMQ_RCVTIMEO */
OPT_TYPE_INT, /* 28 ZMQ_SNDTIMEO */
OPT_TYPE_INT, /* 29 ZMQ_RCVLABEL */
# define MAX_OPTS 29
#else
/* options for versions below 3.0 */
OPT_TYPE_NONE, /* 0 unused */
OPT_TYPE_UINT64, /* 1 ZMQ_HWM */
OPT_TYPE_NONE, /* 2 unused */
OPT_TYPE_INT64, /* 3 ZMQ_SWAP */
OPT_TYPE_UINT64, /* 4 ZMQ_AFFINITY */
OPT_TYPE_STR, /* 5 ZMQ_IDENTITY */
OPT_TYPE_STR, /* 6 ZMQ_SUBSCRIBE */
OPT_TYPE_STR, /* 7 ZMQ_UNSUBSCRIBE */
OPT_TYPE_BLOB, /* 5 ZMQ_IDENTITY */
OPT_TYPE_BLOB, /* 6 ZMQ_SUBSCRIBE */
OPT_TYPE_BLOB, /* 7 ZMQ_UNSUBSCRIBE */
OPT_TYPE_INT64, /* 8 ZMQ_RATE */
OPT_TYPE_INT64, /* 9 ZMQ_RECOVERY_IVL */
OPT_TYPE_INT64, /* 10 ZMQ_MCAST_LOOP */
@ -69,20 +116,22 @@ static const int opt_types[] = {
OPT_TYPE_UINT64, /* 12 ZMQ_RCVBUF */
OPT_TYPE_INT64, /* 13 ZMQ_RCVMORE */
#if VERSION_2_1
# if VERSION_2_1
OPT_TYPE_FD, /* 14 ZMQ_FD */
OPT_TYPE_UINT32, /* 15 ZMQ_EVENTS */
OPT_TYPE_INT, /* 16 ZMQ_TYPE */
OPT_TYPE_INT, /* 17 ZMQ_LINGER */
OPT_TYPE_INT, /* 18 ZMQ_RECONNECT_IVL */
OPT_TYPE_INT, /* 19 ZMQ_BACKLOG */
OPT_TYPE_INT64, /* 20 ZMQ_RECOVERY_IVL_MSEC */
OPT_TYPE_INT, /* 21 ZMQ_RECONNECT_IVL_MAX */
# define MAX_OPTS 21
# else
# define MAX_OPTS 13
# endif
#endif
};
#if VERSION_2_1
#define MAX_OPTS ZMQ_BACKLOG
#else
#define MAX_OPTS ZMQ_RCVMORE
#endif
]],
@ -100,32 +149,83 @@ int zmq_setsockopt (void *s, int option, const void *optval, size_t optvallen);
int zmq_getsockopt (void *s, int option, void *optval, size_t *optvallen);
]],
ffi_source[[
local option_types = {
[zmq.HWM] = 'uint64_t[1]',
[zmq.SWAP] = 'int64_t[1]',
[zmq.AFFINITY] = 'uint64_t[1]',
[zmq.IDENTITY] = 'string',
[zmq.SUBSCRIBE] = 'string',
[zmq.UNSUBSCRIBE] = 'string',
[zmq.RATE] = 'int64_t[1]',
[zmq.RECOVERY_IVL] = 'int64_t[1]',
[zmq.MCAST_LOOP] = 'int64_t[1]',
[zmq.SNDBUF] = 'uint64_t[1]',
[zmq.RCVBUF] = 'uint64_t[1]',
[zmq.RCVMORE] = 'int64_t[1]',
[zmq.FD] = 'int[1]',
[zmq.EVENTS] = 'uint32_t[1]',
[zmq.TYPE] = 'int[1]',
[zmq.LINGER] = 'int[1]',
[zmq.RECONNECT_IVL] = 'int[1]',
[zmq.BACKLOG] = 'int[1]',
}
local option_len = {}
local option_tmps = {}
for k,v in pairs(option_types) do
if v ~= 'string' then
option_len[k] = ffi.sizeof(v)
option_tmps[k] = ffi.new(v, 0)
local option_types = {}
local option_tmps
do
local zmq = _M
local zver = zmq.version()
local str_opt_temp_len = 255
option_tmps = setmetatable({}, {__index = function(tab, ctype)
local temp
if ctype == 'string' then
tmp = ffi.new('uint8_t[?]', str_opt_temp_len)
else
tmp = ffi.new(ctype, 0)
end
rawset(tab, ctype, tmp)
return tmp
end})
local function def_opt(opt, ctype)
if not opt then return end
option_types[opt] = ctype
if ctype == 'string' then
option_len[opt] = str_opt_temp_len
else
option_len[opt] = ffi.sizeof(ctype)
end
end
if zver[1] == 2 then
def_opt(zmq.HWM, 'uint64_t[1]')
def_opt(zmq.SWAP, 'int64_t[1]')
def_opt(zmq.AFFINITY, 'uint64_t[1]')
def_opt(zmq.IDENTITY, 'string')
def_opt(zmq.SUBSCRIBE, 'string')
def_opt(zmq.UNSUBSCRIBE, 'string')
def_opt(zmq.RATE, 'int64_t[1]')
def_opt(zmq.RECOVERY_IVL, 'int64_t[1]')
def_opt(zmq.RECOVERY_IVL_MSEC, 'int64_t[1]')
def_opt(zmq.MCAST_LOOP, 'int64_t[1]')
def_opt(zmq.SNDBUF, 'uint64_t[1]')
def_opt(zmq.RCVBUF, 'uint64_t[1]')
def_opt(zmq.RCVMORE, 'int64_t[1]')
def_opt(zmq.FD, 'int[1]')
def_opt(zmq.EVENTS, 'uint32_t[1]')
def_opt(zmq.TYPE, 'int[1]')
def_opt(zmq.LINGER, 'int[1]')
def_opt(zmq.RECONNECT_IVL, 'int[1]')
def_opt(zmq.BACKLOG, 'int[1]')
def_opt(zmq.RECONNECT_IVL_MAX, 'int[1]')
elseif zver[1] == 3 then
def_opt(zmq.AFFINITY, 'uint64_t[1]')
def_opt(zmq.IDENTITY, 'string')
def_opt(zmq.SUBSCRIBE, 'string')
def_opt(zmq.UNSUBSCRIBE, 'string')
def_opt(zmq.RATE, 'int[1]')
def_opt(zmq.RECOVERY_IVL, 'int[1]')
def_opt(zmq.RECOVERY_IVL_MSEC, 'int[1]')
def_opt(zmq.MCAST_LOOP, 'int[1]')
def_opt(zmq.SNDBUF, 'int[1]')
def_opt(zmq.RCVBUF, 'int[1]')
def_opt(zmq.RCVMORE, 'int[1]')
def_opt(zmq.FD, 'int[1]')
def_opt(zmq.EVENTS, 'int[1]')
def_opt(zmq.TYPE, 'int[1]')
def_opt(zmq.LINGER, 'int[1]')
def_opt(zmq.RECONNECT_IVL, 'int[1]')
def_opt(zmq.BACKLOG, 'int[1]')
def_opt(zmq.RECONNECT_IVL_MAX, 'int[1]')
def_opt(zmq.MAXMSGSIZE, 'int64_t[1]')
def_opt(zmq.SNDHWM, 'int[1]')
def_opt(zmq.RCVHWM, 'int[1]')
def_opt(zmq.MULTICAST_HOPS, 'int[1]')
def_opt(zmq.RCVTIMEO, 'int[1]')
def_opt(zmq.SNDTIMEO, 'int[1]')
def_opt(zmq.RCVLABEL, 'int[1]')
end
end
@ -146,6 +246,20 @@ end
uint64_t uint64_val;
int64_t int64_val;
#if VERSION_3_0
/* 3.0 backwards compatibility support for HWM. */
if(${opt} == ZMQ_HWM) {
int_val = luaL_checklong(L, ${val::idx});
val = &int_val;
val_len = sizeof(int_val);
${err} = zmq_setsockopt(${this}, ZMQ_SNDHWM, val, val_len);
if(-1 != ${err}) {
${err} = zmq_setsockopt(${this}, ZMQ_RCVHWM, val, val_len);
}
goto finished;
}
#endif
if(${opt} > MAX_OPTS) {
return luaL_argerror(L, ${opt::idx}, "Invalid socket option.");
}
@ -178,7 +292,7 @@ end
val = &int64_val;
val_len = sizeof(int64_val);
break;
case OPT_TYPE_STR:
case OPT_TYPE_BLOB:
val = luaL_checklstring(L, ${val::idx}, &(val_len));
break;
default:
@ -187,20 +301,35 @@ end
break;
}
${err} = zmq_setsockopt(${this}, ${opt}, val, val_len);
finished:
]],
ffi_source[[
local ctype = option_types[${opt}]
local tval
local tval_len = 0
if ctype then
if ctype == 'string' then
tval = tostring(${val})
tval_len = #${val}
tval_len = #tval
else
tval = option_tmps[${opt}]
tval = option_tmps[ctype]
tval[0] = ${val}
tval_len = option_len[${opt}]
end
${err} = C.zmq_setsockopt(${this}, ${opt}, tval, tval_len)
elseif ${opt} == zmq.HWM then
-- 3.0 backwards compatibility support for HWM.
ctype = option_types[zmq.SNDHWM]
tval = option_tmps[ctype]
tval[0] = ${val}
tval_len = option_len[zmq.SNDHWM]
${err} = C.zmq_setsockopt(${this}, zmq.SNDHWM, tval, tval_len)
if (-1 ~= ${err}) then
${err} = C.zmq_setsockopt(${this}, zmq.RCVHWM, tval, tval_len)
end
else
error("Invalid socket option.")
end
]],
},
ffi_source[[
@ -272,7 +401,7 @@ local tmp_val_len = ffi.new('size_t[1]', 4)
return 1;
}
break;
case OPT_TYPE_STR:
case OPT_TYPE_BLOB:
val_len = STR_MAX;
${err} = zmq_getsockopt(${this}, ${opt}, str_val, &val_len);
if(0 == ${err}) {
@ -291,21 +420,15 @@ local tmp_val_len = ffi.new('size_t[1]', 4)
ffi_source[[
local ctype = option_types[${opt}]
local val
local val_len = tmp_val_len
if ctype == 'string' then
val_len[0] = 255
val = ffi.new('uint8_t[?]', val_len[0])
ffi.fill(val, val_len[0])
else
val = option_tmps[${opt}]
val[0] = 0
val_len[0] = option_len[${opt}]
local val_len
if ctype then
val = option_tmps[ctype]
val_len = option_len[${opt}]
${err} = C.zmq_getsockopt(${this}, ${opt}, val, tmp_val_len)
end
${err} = C.zmq_getsockopt(${this}, ${opt}, val, val_len)
if ${err} == 0 then
if ctype == 'string' then
val_len = val_len[0]
return ffi.string(val, val_len)
return ffi.string(val, tmp_val_len[0])
else
return tonumber(val[0])
end
@ -313,21 +436,29 @@ local tmp_val_len = ffi.new('size_t[1]', 4)
]],
},
ffi_source[[
-- temp. values for 'events' function.
local events_tmp = ffi.new('uint32_t[1]', 0)
local events_tmp_size = ffi.sizeof('uint32_t')
local events_tmp_len = ffi.new('size_t[1]', events_tmp_size)
local ZMQ_EVENTS = _M.EVENTS
-- temp. values for 'events' function.
local events_ctype = option_types[ZMQ_EVENTS]
local events_tmp = option_tmps[events_ctype]
local events_tmp_len = option_len[ZMQ_EVENTS]
local events_tmp_size = ffi.sizeof(events_tmp)
]],
method "events" {
var_out{ "uint32_t", "events" },
var_out{ "ZMQ_Error", "err" },
c_source[[
#if VERSION_2_1
#if VERSION_3_0
int val;
size_t val_len = sizeof(val);
${err} = zmq_getsockopt(${this}, ZMQ_EVENTS, &val, &val_len);
${events} = val;
#else
# if VERSION_2_1
size_t val_len = sizeof(${events});
${err} = zmq_getsockopt(${this}, ZMQ_EVENTS, &(${events}), &val_len);
#else
# else
luaL_error(L, "'events' method only supported in 0MQ version >= 2.1");
# endif
#endif
]],
ffi_source[[
@ -340,7 +471,7 @@ local ZMQ_EVENTS = _M.EVENTS
-- zmq_send
--
method "send_msg" {
c_method_call "ZMQ_Error" "zmq_send" { "zmq_msg_t *", "msg", "int", "flags?" },
c_export_method_call "ZMQ_Error" "zmq_sendmsg" { "zmq_msg_t *", "msg", "int", "flags?" },
},
-- create helper function for `zmq_send`
c_source[[
@ -353,7 +484,7 @@ ZMQ_Error simple_zmq_send(ZMQ_Socket *sock, const char *data, size_t data_len, i
/* fill message */
memcpy(zmq_msg_data(&msg), data, data_len);
/* send message */
err = zmq_send(sock, &msg, flags);
err = zmq_sendmsg(sock, &msg, flags);
/* close message */
zmq_msg_close(&msg);
}
@ -368,7 +499,7 @@ ZMQ_Error simple_zmq_send(ZMQ_Socket *sock, const char *data, size_t data_len, i
-- zmq_recv
--
method "recv_msg" {
c_method_call "ZMQ_Error" "zmq_recv" { "zmq_msg_t *", "msg", "int", "flags?" },
c_export_method_call "ZMQ_Error" "zmq_recvmsg" { "zmq_msg_t *", "msg", "int", "flags?" },
},
ffi_source[[
local tmp_msg = ffi.new('zmq_msg_t')
@ -383,7 +514,7 @@ local tmp_msg = ffi.new('zmq_msg_t')
${err} = zmq_msg_init(&msg);
if(0 == ${err}) {
/* receive message */
${err} = zmq_recv(${this}, &msg, ${flags});
${err} = zmq_recvmsg(${this}, &msg, ${flags});
if(0 == ${err}) {
${data} = zmq_msg_data(&msg);
${data_len} = zmq_msg_size(&msg);
@ -402,7 +533,7 @@ local tmp_msg = ffi.new('zmq_msg_t')
end
-- receive message
${err} = C.zmq_recv(${this}, msg, ${flags})
${err} = zmq_recvmsg(${this}, msg, ${flags})
if 0 == ${err} then
local data = ffi.string(C.zmq_msg_data(msg), C.zmq_msg_size(msg))
-- close message

119
zmq.nobj.lua
Unescape Escape View File

@ -31,68 +31,101 @@ luajit_ffi_load_cmodule = true,
sys_include "string.h",
include "zmq.h",
c_source "typedefs" [[
#ifndef ZMQ_DONTWAIT
# define ZMQ_DONTWAIT ZMQ_NOBLOCK
#endif
#if ZMQ_VERSION_MAJOR == 2
# define ZMQ_POLL_MSEC 1000 // zmq_poll is usec
#elif ZMQ_VERSION_MAJOR == 3
# define ZMQ_POLL_MSEC 1 // zmq_poll is msec
# ifndef ZMQ_HWM
# define ZMQ_HWM 1 // backwards compatibility
# endif
#endif
#ifndef ZMQ_DEALER
# define ZMQ_DEALER ZMQ_XREQ
#endif
#ifndef ZMQ_ROUTER
# define ZMQ_ROUTER ZMQ_XREP
#endif
]],
--
-- Module constants
--
constants {
MAX_VSM_SIZE = 30,
export_definitions {
MAX_VSM_SIZE = "ZMQ_MAX_VSM_SIZE",
-- message types
DELIMITER = 31,
VSM = 32,
DELIMITER = "ZMQ_DELIMITER",
VSM = "ZMQ_VSM",
-- message flags
MSG_MORE = 1,
MSG_SHARED = 128,
MSG_MORE = "ZMQ_MSG_MORE",
MSG_SHARED = "ZMQ_MSG_SHARED",
-- socket types
PAIR = 0,
PUB = 1,
SUB = 2,
REQ = 3,
REP = 4,
DEALER = 5,
ROUTER = 6,
PULL = 7,
PUSH = 8,
PAIR = "ZMQ_PAIR",
PUB = "ZMQ_PUB",
SUB = "ZMQ_SUB",
REQ = "ZMQ_REQ",
REP = "ZMQ_REP",
DEALER = "ZMQ_DEALER",
ROUTER = "ZMQ_ROUTER",
PULL = "ZMQ_PULL",
PUSH = "ZMQ_PUSH",
-- deprecated
XREQ = 5,
XREP = 6,
XREQ = "ZMQ_DEALER",
XREP = "ZMQ_ROUTER",
-- socket options
HWM = 1,
SWAP = 3,
AFFINITY = 4,
IDENTITY = 5,
SUBSCRIBE = 6,
UNSUBSCRIBE = 7,
RATE = 8,
RECOVERY_IVL = 9,
MCAST_LOOP = 10,
SNDBUF = 11,
RCVBUF = 12,
RCVMORE = 13,
FD = 14,
EVENTS = 15,
TYPE = 16,
LINGER = 17,
RECONNECT_IVL = 18,
BACKLOG = 19,
HWM = "ZMQ_HWM",
SWAP = "ZMQ_SWAP",
AFFINITY = "ZMQ_AFFINITY",
IDENTITY = "ZMQ_IDENTITY",
SUBSCRIBE = "ZMQ_SUBSCRIBE",
UNSUBSCRIBE = "ZMQ_UNSUBSCRIBE",
RATE = "ZMQ_RATE",
RECOVERY_IVL = "ZMQ_RECOVERY_IVL",
MCAST_LOOP = "ZMQ_MCAST_LOOP",
SNDBUF = "ZMQ_SNDBUF",
RCVBUF = "ZMQ_RCVBUF",
RCVMORE = "ZMQ_RCVMORE",
FD = "ZMQ_FD",
EVENTS = "ZMQ_EVENTS",
TYPE = "ZMQ_TYPE",
LINGER = "ZMQ_LINGER",
RECONNECT_IVL = "ZMQ_RECONNECT_IVL",
RECONNECT_IVL_MSEC= "ZMQ_RECONNECT_IVL_MSEC",
BACKLOG = "ZMQ_BACKLOG",
RECONNECT_IVL_MAX = "ZMQ_RECONNECT_IVL_MAX",
MAXMSGSIZE = "ZMQ_MAXMSGSIZE",
SNDHWM = "ZMQ_SNDHWM",
RCVHWM = "ZMQ_RCVHWM",
MULTICAST_HOPS = "ZMQ_MULTICAST_HOPS",
RCVTIMEO = "ZMQ_RCVTIMEO",
SNDTIMEO = "ZMQ_SNDTIMEO",
RCVLABEL = "ZMQ_RCVLABEL",
-- send/recv flags
NOBLOCK = 1,
SNDMORE = 2,
NOBLOCK = "ZMQ_NOBLOCK",
SNDMORE = "ZMQ_SNDMORE",
SNDLABEL = "ZMQ_SNDLABEL",
-- poll events
POLLIN = 1,
POLLOUT = 2,
POLLERR = 4,
POLLIN = "ZMQ_POLLIN",
POLLOUT = "ZMQ_POLLOUT",
POLLERR = "ZMQ_POLLERR",
-- poll milliseconds.
POLL_MSEC = "ZMQ_POLL_MSEC",
-- devices
STREAMER = 1,
FORWARDER = 2,
QUEUE = 3,
STREAMER = "ZMQ_STREAMER",
FORWARDER = "ZMQ_FORWARDER",
QUEUE = "ZMQ_QUEUE",
},

Write Preview
Loading…
Cancel
Save