Added FFI support to Poller object.

src/poller.lua

@@ -65,9 +65,10 @@ function poller_mt:poll(timeout)
 			break
 		end
 		local cb = callbacks[sock]
-		if cb then
-			cb(sock, revents)
+		if not cb then
+			error("Missing callback for sock:" .. tostring(sock))
 		end
+		cb(sock, revents)
 	end
 	return true
 end

src/poller.nobj.lua

@@ -335,6 +335,15 @@ int zmq_poll(zmq_pollitem_t *items, int nitems, long timeout);
 	} else {
 		${this}->next = -1;
 	}
+]],
+		ffi_source[[
+	-- poll for events
+	${err} = poller_poll(${this}, ${timeout})
+	if(${err} > 0) then
+		${this}.next = 0
+	else
+		${this}.next = -1
+	end
 ]],
 	},
 	method "next_revents" {
@@ -372,6 +381,38 @@ int zmq_poll(zmq_pollitem_t *items, int nitems, long timeout);
 		lua_pushnil(L);
 		${this}->next = -1;
 	}
+]],
+		ffi_source[[
+	local sock
+	local idx = ${this}.next
+	if (idx < 0) then return nil, -1 end
+	local count = ${this}.count
+	-- find next item with pending events.
+	while (idx < count and ${this}.items[idx].revents == 0) do
+		idx = idx + 1
+		if (idx >= count) then
+			idx = -1
+			break
+		end
+	end
+	-- did we find a pending event?
+	if(idx >= 0) then
+		-- push the event's sock/fd.
+		if(${this}.items[idx].socket ~= NULL) then
+			sock = obj_type_ZMQ_Socket_push(${this}.items[idx].socket, 0)
+		else
+			sock = tonumber(${this}.items[idx].fd)
+		end
+		${revents} = ${this}.items[idx].revents
+		-- is this the last event.
+		idx = idx + 1
+		if (idx >= count) then
+			idx = -1
+		end
+		${this}.next = idx
+		return sock, ${revents}
+	end
+	${this}.next = idx
 ]],
 	},
 	method "count" {

src/pre_generated-zmq.nobj.c

@@ -1495,6 +1495,69 @@ static const char zmq_ffi_lua_code[] = "-- try loading luajit's ffi\n"
 "\n"
 "local poller_remove_item = ffi.new(\"poller_remove_item_func\", _priv[\"poller_remove_item\"])\n"
 "\n"
+"-- method: poll\n"
+"function ZMQ_Poller_meth.poll(self, timeout)\n"
+"  local this = obj_type_ZMQ_Poller_check(self)\n"
+"  \n"
+"  local err\n"
+"	-- poll for events\n"
+"	err = poller_poll(this, timeout)\n"
+"	if(err > 0) then\n"
+"		this.next = 0\n"
+"	else\n"
+"		this.next = -1\n"
+"	end\n"
+"\n"
+"  -- check for error.\n"
+"  local err_err\n"
+"  if (-1 == err) then\n"
+"    err = false\n"
+"    err_err =   error_code__ZMQ_Error__push(err)\n"
+"  else\n"
+"    err = true\n"
+"  end\n"
+"  return err, err_err\n"
+"end\n"
+"\n"
+"-- method: next_revents\n"
+"function ZMQ_Poller_meth.next_revents(self)\n"
+"  local this = obj_type_ZMQ_Poller_check(self)\n"
+"  local revents\n"
+"	local sock\n"
+"	local idx = this.next\n"
+"	if (idx < 0) then return nil, -1 end\n"
+"	local count = this.count\n"
+"	-- find next item with pending events.\n"
+"	while (idx < count and this.items[idx].revents == 0) do\n"
+"		idx = idx + 1\n"
+"		if (idx >= count) then\n"
+"			idx = -1\n"
+"			break\n"
+"		end\n"
+"	end\n"
+"	-- did we find a pending event?\n"
+"	if(idx >= 0) then\n"
+"		-- push the event's sock/fd.\n"
+"		if(this.items[idx].socket ~= NULL) then\n"
+"			sock = obj_type_ZMQ_Socket_push(this.items[idx].socket, 0)\n"
+"		else\n"
+"			sock = tonumber(this.items[idx].fd)\n"
+"		end\n"
+"		revents = this.items[idx].revents\n"
+"		-- is this the last event.\n"
+"		idx = idx + 1\n"
+"		if (idx >= count) then\n"
+"			idx = -1\n"
+"		end\n"
+"		this.next = idx\n"
+"		return sock, revents\n"
+"	end\n"
+"	this.next = idx\n"
+"\n"
+"  revents = revents\n"
+"  return revents\n"
+"end\n"
+"\n"
 "-- method: count\n"
 "function ZMQ_Poller_meth.count(self)\n"
 "  local this = obj_type_ZMQ_Poller_check(self)\n"