Changeset

--- a/plugins/mod_storage_sql2.lua	Thu May 21 12:11:08 2015 +0200
+++ b/plugins/mod_storage_sql2.lua	Wed Jun 24 23:25:42 2015 +0100
@@ -1,5 +1,6 @@
 
 local json = require "util.json";
+local sql = require "util.sql";
 local xml_parse = require "util.xml".parse;
 local uuid = require "util.uuid";
 local resolve_relative_path = require "util.paths".resolve_relative_path;
@@ -20,106 +21,9 @@
 	end, result, nil;
 end
 
-local mod_sql = module:require("sql");
-local params = module:get_option("sql");
-
-local engine; -- TODO create engine
-
-local function create_table()
-	local Table,Column,Index = mod_sql.Table,mod_sql.Column,mod_sql.Index;
-
-	local ProsodyTable = Table {
-		name="prosody";
-		Column { name="host", type="TEXT", nullable=false };
-		Column { name="user", type="TEXT", nullable=false };
-		Column { name="store", type="TEXT", nullable=false };
-		Column { name="key", type="TEXT", nullable=false };
-		Column { name="type", type="TEXT", nullable=false };
-		Column { name="value", type="MEDIUMTEXT", nullable=false };
-		Index { name="prosody_index", "host", "user", "store", "key" };
-	};
-	engine:transaction(function()
-		ProsodyTable:create(engine);
-	end);
-
-	local ProsodyArchiveTable = Table {
-		name="prosodyarchive";
-		Column { name="sort_id", type="INTEGER", primary_key=true, auto_increment=true };
-		Column { name="host", type="TEXT", nullable=false };
-		Column { name="user", type="TEXT", nullable=false };
-		Column { name="store", type="TEXT", nullable=false };
-		Column { name="key", type="TEXT", nullable=false }; -- item id
-		Column { name="when", type="INTEGER", nullable=false }; -- timestamp
-		Column { name="with", type="TEXT", nullable=false }; -- related id
-		Column { name="type", type="TEXT", nullable=false };
-		Column { name="value", type="MEDIUMTEXT", nullable=false };
-		Index { name="prosodyarchive_index", unique = true, "host", "user", "store", "key" };
-	};
-	engine:transaction(function()
-		ProsodyArchiveTable:create(engine);
-	end);
-end
+local default_params = { driver = "SQLite3" };
 
-local function upgrade_table()
-	if params.driver == "MySQL" then
-		local success,err = engine:transaction(function()
-			local result = engine:execute("SHOW COLUMNS FROM prosody WHERE Field='value' and Type='text'");
-			if result:rowcount() > 0 then
-				module:log("info", "Upgrading database schema...");
-				engine:execute("ALTER TABLE prosody MODIFY COLUMN `value` MEDIUMTEXT");
-				module:log("info", "Database table automatically upgraded");
-			end
-			return true;
-		end);
-		if not success then
-			module:log("error", "Failed to check/upgrade database schema (%s), please see "
-				.."http://prosody.im/doc/mysql for help",
-				err or "unknown error");
-			return false;
-		end
-		-- COMPAT w/pre-0.9: Upgrade tables to UTF-8 if not already
-		local check_encoding_query = "SELECT `COLUMN_NAME`,`COLUMN_TYPE` FROM `information_schema`.`columns` WHERE `TABLE_NAME`='prosody' AND ( `CHARACTER_SET_NAME`!='utf8' OR `COLLATION_NAME`!='utf8_bin' );";
-		success,err = engine:transaction(function()
-			local result = engine:execute(check_encoding_query);
-			local n_bad_columns = result:rowcount();
-			if n_bad_columns > 0 then
-				module:log("warn", "Found %d columns in prosody table requiring encoding change, updating now...", n_bad_columns);
-				local fix_column_query1 = "ALTER TABLE `prosody` CHANGE `%s` `%s` BLOB;";
-				local fix_column_query2 = "ALTER TABLE `prosody` CHANGE `%s` `%s` %s CHARACTER SET 'utf8' COLLATE 'utf8_bin';";
-				for row in result:rows() do
-					local column_name, column_type = unpack(row);
-					engine:execute(fix_column_query1:format(column_name, column_name));
-					engine:execute(fix_column_query2:format(column_name, column_name, column_type));
-				end
-				module:log("info", "Database encoding upgrade complete!");
-			end
-		end);
-		success,err = engine:transaction(function() return engine:execute(check_encoding_query); end);
-		if not success then
-			module:log("error", "Failed to check/upgrade database encoding: %s", err or "unknown error");
-		end
-	end
-end
-
-do -- process options to get a db connection
-	params = params or { driver = "SQLite3" };
-
-	if params.driver == "SQLite3" then
-		params.database = resolve_relative_path(prosody.paths.data or ".", params.database or "prosody.sqlite");
-	end
-
-	assert(params.driver and params.database, "Both the SQL driver and the database need to be specified");
-
-	--local dburi = db2uri(params);
-	engine = mod_sql:create_engine(params);
-
-	if module:get_option("sql_manage_tables", true) then
-		-- Automatically create table, ignore failure (table probably already exists)
-		create_table();
-		-- Encoding mess
-		upgrade_table();
-	end
-end
+local engine;
 
 local function serialize(value)
 	local t = type(value);
@@ -192,13 +96,18 @@
 	return true;
 end
 
+--- Key/value store API (default store type)
+
 local keyval_store = {};
 keyval_store.__index = keyval_store;
 function keyval_store:get(username)
-	user,store = username,self.store;
+	user, store = username, self.store;
 	local ok, result = engine:transaction(keyval_store_get);
-	if not ok then return ok, result; end
-	return result;
+	if not ok then
+		module:log("error", "Unable to read from database %s store for %s: %s", store, username or "<host>", result);
+		return nil, result;
+	end
+	return result;	
 end
 function keyval_store:set(username, data)
 	user,store = username,self.store;
@@ -214,6 +123,8 @@
 	return iterator(result);
 end
 
+--- Archive store API
+
 local archive_store = {}
 archive_store.caps = {
 	total = true;
@@ -345,6 +256,10 @@
 	archive = archive_store;
 };
 
+--- Implement storage driver API
+
+-- FIXME: Some of these operations need to operate on the archive store(s) too
+
 local driver = {};
 
 function driver:open(store, typ)
@@ -356,13 +271,13 @@
 end
 
 function driver:stores(username)
-	local sql = "SELECT DISTINCT `store` FROM `prosody` WHERE `host`=? AND `user`" ..
+	local query = "SELECT DISTINCT `store` FROM `prosody` WHERE `host`=? AND `user`" ..
 		(username == true and "!=?" or "=?");
 	if username == true or not username then
 		username = "";
 	end
 	local ok, result = engine:transaction(function()
-		return engine:select(sql, host, username);
+		return engine:select(query, host, username);
 	end);
 	if not ok then return ok, result end
 	return iterator(result);
@@ -371,10 +286,121 @@
 function driver:purge(username)
 	return engine:transaction(function()
 		local stmt,err = engine:delete("DELETE FROM `prosody` WHERE `host`=? AND `user`=?", host, username);
-		return true,err;
+		return true, err;
+	end);
+end
+
+--- Initialization
+
+
+local function create_table(name)
+	local Table, Column, Index = sql.Table, sql.Column, sql.Index;
+
+	local ProsodyTable = Table {
+		name= name or "prosody";
+		Column { name="host", type="TEXT", nullable=false };
+		Column { name="user", type="TEXT", nullable=false };
+		Column { name="store", type="TEXT", nullable=false };
+		Column { name="key", type="TEXT", nullable=false };
+		Column { name="type", type="TEXT", nullable=false };
+		Column { name="value", type="MEDIUMTEXT", nullable=false };
+		Index { name="prosody_index", "host", "user", "store", "key" };
+	};
+	engine:transaction(function()
+		ProsodyTable:create(engine);
+	end);
+
+	local ProsodyArchiveTable = Table {
+		name="prosodyarchive";
+		Column { name="sort_id", type="INTEGER", primary_key=true, auto_increment=true };
+		Column { name="host", type="TEXT", nullable=false };
+		Column { name="user", type="TEXT", nullable=false };
+		Column { name="store", type="TEXT", nullable=false };
+		Column { name="key", type="TEXT", nullable=false }; -- item id
+		Column { name="when", type="INTEGER", nullable=false }; -- timestamp
+		Column { name="with", type="TEXT", nullable=false }; -- related id
+		Column { name="type", type="TEXT", nullable=false };
+		Column { name="value", type="MEDIUMTEXT", nullable=false };
+		Index { name="prosodyarchive_index", unique = true, "host", "user", "store", "key" };
+	};
+	engine:transaction(function()
+		ProsodyArchiveTable:create(engine);
 	end);
 end
 
-module:provides("storage", driver);
+local function upgrade_table(params, apply_changes)
+	local changes = false;
+	if params.driver == "MySQL" then
+		local success,err = engine:transaction(function()
+			local result = engine:execute("SHOW COLUMNS FROM prosody WHERE Field='value' and Type='text'");
+			if result:rowcount() > 0 then
+				changes = true;
+				if apply_changes then
+					module:log("info", "Upgrading database schema...");
+					engine:execute("ALTER TABLE prosody MODIFY COLUMN `value` MEDIUMTEXT");
+					module:log("info", "Database table automatically upgraded");
+				end
+			end
+			return true;
+		end);
+		if not success then
+			module:log("error", "Failed to check/upgrade database schema (%s), please see "
+				.."http://prosody.im/doc/mysql for help",
+				err or "unknown error");
+			return false;
+		end
 
+		-- COMPAT w/pre-0.10: Upgrade table to UTF-8 if not already
+		local check_encoding_query = "SELECT `COLUMN_NAME`,`COLUMN_TYPE` FROM `information_schema`.`columns` WHERE `TABLE_NAME`='prosody' AND ( `CHARACTER_SET_NAME`!='utf8' OR `COLLATION_NAME`!='utf8_bin' );";
+		success,err = engine:transaction(function()
+			local result = engine:execute(check_encoding_query);
+			local n_bad_columns = result:rowcount();
+			if n_bad_columns > 0 then
+				changes = true;
+				if apply_changes then
+					module:log("warn", "Found %d columns in prosody table requiring encoding change, updating now...", n_bad_columns);
+					local fix_column_query1 = "ALTER TABLE `prosody` CHANGE `%s` `%s` BLOB;";
+					local fix_column_query2 = "ALTER TABLE `prosody` CHANGE `%s` `%s` %s CHARACTER SET 'utf8' COLLATE 'utf8_bin';";
+					for row in result:rows() do
+						local column_name, column_type = unpack(row);
+						engine:execute(fix_column_query1:format(column_name, column_name));
+						engine:execute(fix_column_query2:format(column_name, column_name, column_type));
+					end
+					module:log("info", "Database encoding upgrade complete!");
+				end
+			end
+		end);
+		success,err = engine:transaction(function() return engine:execute(check_encoding_query); end);
+		if not success then
+			module:log("error", "Failed to check/upgrade database encoding: %s", err or "unknown error");
+			return false;
+		end
+	end
+end
 
+local function normalize_params(params)
+	assert(params.driver and params.database, "Configuration error: Both the SQL driver and the database need to be specified");
+	if params.driver == "SQLite3" then
+		params.database = resolve_relative_path(prosody.paths.data or ".", params.database or "prosody.sqlite");
+	end
+	return params;
+end
+
+function module.load()
+	if prosody.prosodyctl then return; end
+	local params = normalize_params(module:get_option("sql", default_params));
+	engine = sql:create_engine(params, function (engine)
+		if module:get_option("sql_manage_tables", true) then
+			-- Automatically create table, ignore failure (table probably already exists)
+			-- FIXME: we should check in information_schema, etc.
+			create_table();
+			-- Check whether the table needs upgrading
+			if not upgrade_table(params, true) then
+				module:log("error", "Old database format detected, and upgrade failed");
+				return false, "database upgrade needed";
+			end
+		end
+	end);
+
+	module:provides("storage", driver);
+end

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/tests/test_util_queue.lua	Wed Jun 24 23:25:42 2015 +0100
@@ -0,0 +1,68 @@
+local new = require "util.queue".new;
+
+local q = new(10);
+
+assert(q.size == 10);
+assert(q:count() == 0);
+
+assert(q:push("one"));
+assert(q:push("two"));
+assert(q:push("three"));
+
+for i = 4, 10 do
+	print("pushing "..i)
+	assert(q:push("hello"));
+	assert(q:count() == i, "count is not "..i.."("..q:count()..")");
+end
+assert(q:push("hello") == nil, "queue overfull!");
+assert(q:push("hello") == nil, "queue overfull!");
+assert(q:pop() == "one", "queue item incorrect");
+assert(q:pop() == "two", "queue item incorrect");
+assert(q:push("hello"));
+assert(q:push("hello"));
+assert(q:pop() == "three", "queue item incorrect");
+assert(q:push("hello"));
+assert(q:push("hello") == nil, "queue overfull!");
+assert(q:push("hello") == nil, "queue overfull!");
+
+assert(q:count() == 10, "queue count incorrect");
+
+for i = 1, 10 do
+	assert(q:pop() == "hello", "queue item incorrect");
+end
+
+assert(q:count() == 0, "queue count incorrect");
+
+assert(q:push(1));
+for i = 1, 1001 do
+	assert(q:pop() == i);
+	assert(q:count() == 0);
+	assert(q:push(i+1));
+	assert(q:count() == 1);
+end
+assert(q:pop() == 1002);
+assert(q:push(1));
+for i = 1, 1000000 do
+	q:pop();
+	q:push(i+1);
+end
+
+-- Test queues that purge old items when pushing to a full queue
+local q = new(10, true);
+
+for i = 1, 10 do
+	q:push(i);
+end
+
+assert(q:count() == 10);
+
+assert(q:push(11));
+assert(q:count() == 10);
+assert(q:pop() == 2); -- First item should have been purged
+
+for i = 12, 32 do
+	assert(q:push(i));
+end
+
+assert(q:count() == 10);
+assert(q:pop() == 23);

--- a/util/queue.lua	Thu May 21 12:11:08 2015 +0200
+++ b/util/queue.lua	Wed Jun 24 23:25:42 2015 +0100
@@ -11,7 +11,7 @@
 
 local have_utable, utable = pcall(require, "util.table"); -- For pre-allocation of table
 
-local function new(size)
+local function new(size, allow_wrapping)
 	-- Head is next insert, tail is next read
 	local head, tail = 1, 1;
 	local items = 0; -- Number of stored items
@@ -22,7 +22,12 @@
 		count = function (self) return items; end;
 		push = function (self, item)
 			if items >= size then
-				return nil, "queue full";
+				if allow_wrapping then
+					tail = (tail%size)+1; -- Advance to next oldest item
+					items = items - 1;
+				else
+					return nil, "queue full";
+				end
 			end
 			t[head] = item;
 			items = items + 1;

--- a/util/sql.lua	Thu May 21 12:11:08 2015 +0200
+++ b/util/sql.lua	Wed Jun 24 23:25:42 2015 +0100
@@ -25,34 +25,9 @@
 function is_index(x) return getmetatable(x)==index_mt; end
 function is_table(x) return getmetatable(x)==table_mt; end
 function is_query(x) return getmetatable(x)==query_mt; end
---function is_op(x) return getmetatable(x)==op_mt; end
---function expr(...) return setmetatable({...}, op_mt); end
 function Integer(n) return "Integer()" end
 function String(n) return "String()" end
 
---[[local ops = {
-	__add = function(a, b) return "("..a.."+"..b..")" end;
-	__sub = function(a, b) return "("..a.."-"..b..")" end;
-	__mul = function(a, b) return "("..a.."*"..b..")" end;
-	__div = function(a, b) return "("..a.."/"..b..")" end;
-	__mod = function(a, b) return "("..a.."%"..b..")" end;
-	__pow = function(a, b) return "POW("..a..","..b..")" end;
-	__unm = function(a) return "NOT("..a..")" end;
-	__len = function(a) return "COUNT("..a..")" end;
-	__eq = function(a, b) return "("..a.."=="..b..")" end;
-	__lt = function(a, b) return "("..a.."<"..b..")" end;
-	__le = function(a, b) return "("..a.."<="..b..")" end;
-};
-
-local functions = {
-
-};
-
-local cmap = {
-	[Integer] = Integer();
-	[String] = String();
-};]]
-
 function Column(definition)
 	return setmetatable(definition, column_mt);
 end
@@ -94,7 +69,6 @@
 	return s..' }';
 --	return 'Index{ name="'..self.name..'", type="'..self.type..'" }'
 end
---
 
 local function urldecode(s) return s and (s:gsub("%%(%x%x)", function (c) return s_char(tonumber(c,16)); end)); end
 local function parse_url(url)
@@ -121,26 +95,6 @@
 	};
 end
 
---[[local session = {};
-
-function session.query(...)
-	local rets = {...};
-	local query = setmetatable({ __rets = rets, __filters }, query_mt);
-	return query;
-end
---
-
-local function db2uri(params)
-	return build_url{
-		scheme = params.driver,
-		user = params.username,
-		password = params.password,
-		host = params.host,
-		port = params.port,
-		path = params.database,
-	};
-end]]
-
 local engine = {};
 function engine:connect()
 	if self.conn then return true; end
@@ -209,8 +163,8 @@
 engine.update = engine.execute_update;
 function engine:_transaction(func, ...)
 	if not self.conn then
-		local a,b = self:connect();
-		if not a then return a,b; end
+		local ok, err = self:connect();
+		if not ok then return ok, err; end
 	end
 	--assert(not self.__transaction, "Recursive transactions not allowed");
 	local args, n_args = {...}, select("#", ...);
@@ -230,15 +184,15 @@
 	end
 end
 function engine:transaction(...)
-	local a,b = self:_transaction(...);
-	if not a then
+	local ok, ret = self:_transaction(...);
+	if not ok then
 		local conn = self.conn;
 		if not conn or not conn:ping() then
 			self.conn = nil;
-			a,b = self:_transaction(...);
+			ok, ret = self:_transaction(...);
 		end
 	end
-	return a,b;
+	return ok, ret;
 end
 function engine:_create_index(index)
 	local sql = "CREATE INDEX `"..index.name.."` ON `"..index.table.."` (";
@@ -319,7 +273,7 @@
 end
 local engine_mt = { __index = engine };
 
-local function db2uri(params)
+function db2uri(params)
 	return build_url{
 		scheme = params.driver,
 		user = params.username,
@@ -329,55 +283,9 @@
 		path = params.database,
 	};
 end
-local engine_cache = {}; -- TODO make weak valued
-function create_engine(self, params)
-	local url = db2uri(params);
-	if not engine_cache[url] then
-		local engine = setmetatable({ url = url, params = params }, engine_mt);
-		engine_cache[url] = engine;
-	end
-	return engine_cache[url];
+
+function create_engine(self, params, onconnect)
+	return setmetatable({ url = db2uri(params), params = params, onconnect = onconnect }, engine_mt);
 end
 
-
---[[Users = Table {
-	name="users";
-	Column { name="user_id", type=String(), primary_key=true };
-};
-print(Users)
-print(Users.c.user_id)]]
-
---local engine = create_engine('postgresql://scott:tiger@localhost:5432/mydatabase');
---[[local engine = create_engine{ driver = "SQLite3", database = "./alchemy.sqlite" };
-
-local i = 0;
-for row in assert(engine:execute("select * from sqlite_master")):rows(true) do
-	i = i+1;
-	print(i);
-	for k,v in pairs(row) do
-		print("",k,v);
-	end
-end
-print("---")
-
-Prosody = Table {
-	name="prosody";
-	Column { name="host", type="TEXT", nullable=false };
-	Column { name="user", type="TEXT", nullable=false };
-	Column { name="store", type="TEXT", nullable=false };
-	Column { name="key", type="TEXT", nullable=false };
-	Column { name="type", type="TEXT", nullable=false };
-	Column { name="value", type="TEXT", nullable=false };
-	Index { name="prosody_index", "host", "user", "store", "key" };
-};
---print(Prosody);
-assert(engine:transaction(function()
-	assert(Prosody:create(engine));
-end));
-
-for row in assert(engine:execute("select user from prosody")):rows(true) do
-	print("username:", row['username'])
-end
---result.close();]]
-
 return _M;