Higher level MQ functions, with push/pull object serialisation. Delicate.

2 files changed, 276 insertions, 0 deletions

diff --git a/luxio/mq.lua b/luxio/mq.lua
new file mode 100644
index 0000000..b42adc6
--- /dev/null
+++ b/luxio/mq.lua
@@ -0,0 +1,172 @@
+-- Light Unix I/O for Lua
+-- Copyright 2012 Rob Kendrick <rjek+luxio@rjek.com>
+--
+-- Distributed under the same terms as Lua itself (MIT).
+--
+-- High-level functions for using POSIX message queues.  Delicate.
+
+local l = require "luxio"
+local sio = require "luxio.simple"
+local serialise = require "luxio.serialise"
+local lua_version = tonumber(_VERSION:match "Lua (%d.%d)")
+
+local function err(txt, errno)
+   return nil, ("%s: %s"):format(txt, l.strerror(errno)), errno
+end
+
+local mq_check_mt
+
+local function mq_meta_gc(o)
+   if not o.closed then
+      l.mq_close(o.desc)
+   end
+end
+
+local mq_mt = {
+   send = function(o, msg, prio)
+      mq_check_mt(o)
+
+      local r, errno = l.mq_send(o.desc, msg, prio)
+      if r < 0 then
+	 return err("mq_send", errno)
+      end
+
+      return true
+   end,
+
+   receive = function(o)
+      mq_check_mt(o)
+
+      local r, errno, msg, prio = l.mq_receive(o.desc)
+      if r < 0 then
+	 return err("mq_receive", errno)
+      end
+
+      return msg, prio
+   end,
+
+   close = function(o)
+      mq_check_mt(o)
+      local r, errno = l.mq_close(o.desc)
+      if r < 0 then
+	 return err("mq_close", errno)
+      end
+      o.closed = true
+
+      return true
+   end,
+
+   push = function(o, data, prio)
+      mq_check_mt(o)
+      local payload = serialise(data)
+      local r, errno, attr = l.mq_getattr(o.desc)
+      
+      if r < 0 then
+	 return err("mq_getattr", errno)
+      end
+
+      if #payload > attr.mq_msgsize then
+	 return nil, "payload serialisation too large"
+      end
+
+      r, errno = l.mq_send(o.desc, payload, prio or 0)
+      if r < 0 then
+	 return err("mq_send", errno)
+      end
+      
+      return true
+   end,
+
+   pull = function(o)
+      mq_check_mt(o)
+      local r, errno, data, prio = l.mq_receive(o.desc)
+      if r < 0 then
+	 return err("mq_receive", errno)
+      end
+
+      local f, err = loadstring("return " .. data)
+      if f == nil then
+	 return nil, "payload error: " .. err
+      end
+
+      local ok, result = pcall(f)
+      if not ok then
+	 return nil, "payload error: ".. result
+      end
+      
+      return result
+   end,
+
+   __tostring = function(x)
+      return ("message queue: %s %s (%d)%s"):format(
+	 x.rw_mode, x.path, x.desc, x.closed and " closed" or "")
+   end,
+
+   __gc = mq_meta_gc
+
+}
+
+mq_mt.__index = mq_mt
+
+if lua_version == 5.1 then
+   mq_wrap_mt = function(t)
+      t.proxy = newproxy(true)
+      getmetatable(t.proxy).__gc = function()
+	 mq_meta_gc(t)
+      end
+      setmetatable(t, mq_mt)
+   end
+else
+   mq_wrap_mt = function(t) 
+      setmetatable(t, mq_mt)
+   end
+end
+
+
+mq_check_mt = function(x)
+   if getmetatable(x) ~= mq_mt then
+      error "object passed is not an MQ object"
+   end
+   if x.closed then
+      error "MQ passed has been closed"
+   end
+end
+
+local rw_mode_map = {
+   r = l.O_RDONLY,
+   w = l.O_WRONLY,
+   rw = l.O_RDWR
+}
+
+local function open(path, rw, mode)
+   -- Opens a message queue.  rw is one of "r", "w", or "rw".
+   -- If mode is specified, it will also try to create it.
+
+   local rw_mode = rw_mode_map[rw] or error "unknown read/write mode"
+   local mq, errno
+
+   if mode then
+      mode = sio.tomode(mode) 
+      mq, errno = l.mq_open(path, l.bit.bor(rw_mode, l.O_CREAT), mode)
+   else
+      mq, errno = l.mq_open(path, rw_mode)
+   end
+
+   if mq < 0 then
+      return err("mq_open", errno)
+   end
+
+   local t = { path = path, desc = mq, rw_mode = rw }
+   mq_wrap_mt(t)
+   return t
+end
+
+local function unlink(path)
+   local r, errno = l.mq_unlink(path)
+   if r < 0 then
+      return err("mq_unlink", errno)
+   end
+   return true
+end
+
+return { open = open, unlink = unlink }
\ No newline at end of file
diff --git a/luxio/serialise.lua b/luxio/serialise.lua
new file mode 100644
index 0000000..0e9152b
--- /dev/null
+++ b/luxio/serialise.lua
@@ -0,0 +1,104 @@
+local type = type
+local format = string.format
+local tostring = tostring
+local error = error
+local concat = table.concat
+local dump = string.dump
+
+local function a(t, x)
+   t[#t+1] = x
+end
+
+local keywords = {
+   ["and"] = true,
+   ["break"] = true,
+   ["do"] = true,
+   ["else"] = true,
+   ["elseif"] = true,
+   ["end"] = true,
+   ["false"] = true,
+   ["for"] = true,
+   ["function"] = true,
+   ["if"] = true,
+   ["in"] = true,
+   ["local"] = true,
+   ["nil"] = true,
+   ["not"] = true,
+   ["or"] = true,
+   ["repeat"] = true,
+   ["return"] = true,
+   ["then"] = true,
+   ["true"] = true,
+   ["until"] = true,
+   ["while"] = true
+}
+
+local serialise
+
+serialise = function(x)
+   local t = type(x)
+   if t == "string" then
+      return format("%q", x)
+   elseif t == "function" then
+      return format("loadstring(%q)", dump(x))
+   elseif t == "userdata" or t == "lightuserdata" then
+      error "Cannot serialise userdata or lightuserdata"
+   elseif t ~= "table" then
+      return tostring(x)
+   end
+
+   local r = { "{" }
+
+   -- first emit any sequence
+   local k = 1
+   repeat
+      local v = x[k]
+      if v ~= nil then
+	 a(r, serialise(v))
+	 a(r, ",")
+      end
+
+      k = k + 1
+   until v == nil
+
+   -- emit other numerical keys
+   for i, j in pairs(x) do
+      if type(i) == "number" and i >= k then
+	 if r[#r] ~= "," then a(r, ",") end
+	 a(r, format("[%d]=%s", i, serialise(j)))
+	 a(r, ",")
+      end
+   end
+
+   -- emit non-numeric keys
+   for i, j in pairs(x) do
+      local key
+      local ti = type(i)
+      if ti ~= "number" then
+	 if ti == "string" then
+	    key = format("%q", i)
+	    if not keywords[i] and i == format("%q", i):sub(2, -2) then
+	       key = i
+	    else
+	       key = "[" .. key .. "]"
+	    end
+	 elseif ti == "function" then
+	    key = "[loadstring(" .. format("%q", dump(i)) .. ")]"
+	 elseif ti == "table" then
+	    key = "[" .. serialise(i) .. "]"
+	 else
+	    error "unhandled key type"
+	 end
+	       
+	 a(r, format("%s=%s", key, serialise(j)))
+	 a(r, ",")
+      end
+   end
+
+   if r[#r] == "," then r[#r] = nil end
+   a(r, "}")
+
+   return concat(r)
+end
+
+return serialise