Q-Sys Redis Client

redis.lua

Redis = (function(host, port)

  local R = { _ = {} };
  
  -- Socket management
  R._.Host = host;
  R._.Port = port;
  R._.Socket = TcpSocket.New();
  R._.Socket.ReconnectTimeout = 0.1;
  R._.Socket.EventHandler = function(sock, evt)
    print(evt);
  end;
  R.Reconnect = function()
    if(R._.Socket.IsConnected) then R._.Socket:Disconnect(); end;
    R._.Socket:Connect(host, port);
  end
  
  -- Command queue & timeout timer
  R.Timeout = 1000; -- one second
  R._.TimeoutTimer = Timer.New();
  R._.CmdQueue = {};
  function enqueue(msg)
    table.insert(R._.CmdQueue, msg);          -- add the command to the queue
    if(not R._.Waiting) then dequeue(); end;  -- send immediately if we can
  end;
  function dequeue()
    local msg = table.remove(R._.CmdQueue, 1);
    if(not msg) then return; end;             -- bail out if the queue is empty
    R._.Waiting = msg;                        -- store command details for data handler
    R._.Socket:Write(msg.s);                  -- send command out to the socket
    R._.TimeoutTimer:Start(R.Timeout / 1000); -- start timeout timer in milliseconds
  end;
  R._.TimeoutTimer.EventHandler = function()
    R._.TimeoutTimer:Stop();
    R.Reconnect();
    error('redis - timeout waiting for response to ' .. (R._.Waiting.cmd) .. ' (' .. R.Timeout .. 'ms)');
  end
  
  -- Multi-mode response tokenizer
  function s_read(sock, amt)
    return (sock.BufferLength > amt) and sock:Read(amt) or nil;
  end
  R._.Deserializer = { read = 'crlf' };
  R._.Socket.Data = function(sock)
    local d = R._.Deserializer;
    -- modes: linemode & binary mode.
    while(d.read ~= 'crlf' and sock.BufferLength > d.read+2 or sock:Search('\r\n')) do
      if(d.read == 'crlf') then
        handleToken(sock:ReadLine(TcpSocket.EOL.CrLf), 'line');
      else
        handleToken(sock:Read(d.read), 'bin');
        sock:Read(2);; -- discard a CrLf
        d.read = 'crlf'; -- go back to line mode
      end;
    end
  end;
  
  -- Result inflater
  R._.ObjectConstructor = { tree = {}, obj = {} };
  function handleToken(val, typ)
    local d = R._.ObjectConstructor;
    local el, more, err = (function()
      if(typ == 'line') then
        return deserialize(val)
      else
        return val, 0
      end
    end)();
    
    -- Track the depth into the resultant object, and the number of items at that level
    local target = d.tree[#d.tree] or { obj = d.obj, len = 1 };
    local new_leaf = {obj = el or target.obj, len = more};
    
    -- Construct the result object
    table.insert(target.obj, el);
    target.len = target.len - 1;
    
    -- Walk back up the tree as far as we can
    table.insert(d.tree, new_leaf);
    while(#d.tree > 0 and d.tree[#d.tree].len == 0) do
      table.remove(d.tree);
    end;
    
    -- If we're back at level 0, we've parsed all the objects and can return.
    if(#d.tree == 0) then
      R._.TimeoutTimer:Stop();
      R._.Waiting.cb(d.obj[1], err);
      R._.Waiting = nil;
      d.obj = {};
      dequeue();
    end;
    
  end;
  
  -- Serialization
  function serializeCommand(parts)
    local serialized = '*'         -- commands are sent as an array
            .. (#parts) .. '\r\n'; -- number of parts = array length
    for _,p in ipairs(parts) do        -- serialize each part
      serialized = serialized  .. '$'  -- as a bulk string type
                    .. (#p) ..'\r\n'   -- with a given length
                    .. p .. '\r\n'     -- the string itself
    end
    return serialized;
  end
  
  function deserialize(line)
    local resp_type, data = line:match('(.)(.+)');
    if(resp_type == '+') then     -- it's a string
      return data, 0;
    elseif(resp_type == '$') then -- it's a bulk string
      if(data == '-1') then
        return nil, 0;
      else
        R._.Deserializer.read = tonumber(data);
        return nil,1;
      end;
    elseif(resp_type == '-') then -- it's an error
      return nil, 0, data;
    elseif(resp_type == ':') then -- it's an integer
      return tonumber(data), 0;
    elseif(resp_type == '*') then -- it's an array!
      if(data == '-1') then
        return nil, 0;
      else
        return {}, tonumber(data);
      end;
    else
      error('redis: unrecognized type ' .. resp_type);
    end;
  end;
  
  
  function sendCommand(cmd, args, cb)
  
    -- Serialize command
    cmd = cmd:upper();
    local parts = {cmd};
    for i=1,#args do table.insert(parts, tostring(args[i])); end;
    local s = serializeCommand(parts);
    
    -- Enqueue for sending
    enqueue({s=s, cb=cb, cmd=cmd});
    
  end;
  
  -- Lazy creation of methods for the Redis API
  setmetatable(R, {
    __index = function(o,k)
      k = k:gsub('_',' '):upper();
      local method = function(...)
        local args = {...};
        local cb = table.remove(args);
        sendCommand(k:upper(), args, cb);
      end;
      R[k] = method; -- cache the function
      return method;
    end
  });
  
  R.Reconnect();
  return R;

end);

local redis = Redis('localhost', 6379);

--[[
EXAMPLE USAGE:

  redis.auth('mypassword', function(result, err)
    if(result ~= 'OK') then error(err); end;
  end)

  redis.get('foo', function(result, err) 
    if(err) then error(err); end;
    print('RESULT: ', result);
  end);

]]