* Add lua bindings for upstream API.

8 files changed, 626 insertions, 38 deletions

diff --git a/src/cfg_file.h b/src/cfg_file.h
index b90171426..f07539295 100644
--- a/src/cfg_file.h
+++ b/src/cfg_file.h
@@ -336,6 +336,15 @@ struct config_file {
 
 
 /**
+ * Parse host[:port[:priority]] line
+ * @param ina host address
+ * @param port port
+ * @param priority priority
+ * @return TRUE if string was parsed
+ */
+gboolean parse_host_port_priority (const gchar *str, struct in_addr *ina, guint16 *port, guint *priority);
+
+/**
  * Parse host:port line
  * @param ina host address
  * @param port port
@@ -344,6 +353,14 @@ struct config_file {
 gboolean parse_host_port (const gchar *str, struct in_addr *ina, guint16 *port);
 
 /**
+ * Parse host:priority line
+ * @param ina host address
+ * @param priority priority
+ * @return TRUE if string was parsed
+ */
+gboolean parse_host_priority (const gchar *str, struct in_addr *ina, guint *priority);
+
+/**
  * Parse bind credits
  * @param cf config file to use
  * @param str line that presents bind line
diff --git a/src/cfg_utils.c b/src/cfg_utils.c
index f91882f98..c872583f2 100644
--- a/src/cfg_utils.c
+++ b/src/cfg_utils.c
@@ -41,11 +41,11 @@
 
 
 gboolean
-parse_host_port (const gchar *str, struct in_addr *ina, guint16 *port)
+parse_host_port_priority (const gchar *str, struct in_addr *ina, guint16 *port, guint *priority)
 {
-	gchar                           **tokens, *err_str;
+	gchar                          **tokens, *err_str, *cur_tok;
 	struct hostent                 *hent;
-	guint                           port_parsed, saved_errno = errno;
+	guint                           port_parsed, priority_parsed, saved_errno = errno;
 
 	tokens = g_strsplit_set (str, ":", 0);
 	if (!tokens || !tokens[0]) {
@@ -72,18 +72,38 @@ parse_host_port (const gchar *str, struct in_addr *ina, guint16 *port)
 	}
 	if (tokens[1] != NULL) {
 		/* Port part */
-		errno = 0;
-		port_parsed = strtoul (tokens[1], &err_str, 10);
-		if (*err_str != '\0' || errno != 0) {
-			msg_warn ("cannot parse port: %s, at symbol %c, error: %s", tokens[1], *err_str, strerror (errno));
-			goto err;
+		if (port != NULL) {
+			errno = 0;
+			port_parsed = strtoul (tokens[1], &err_str, 10);
+			if (*err_str != '\0' || errno != 0) {
+				msg_warn ("cannot parse port: %s, at symbol %c, error: %s", tokens[1], *err_str, strerror (errno));
+				goto err;
+			}
+			if (port_parsed > G_MAXUINT16) {
+				errno = ERANGE;
+				msg_warn ("cannot parse port: %s, error: %s", tokens[1], *err_str, strerror (errno));
+				goto err;
+			}
+			*port = port_parsed;
 		}
-		if (port_parsed > G_MAXUINT16) {
-			errno = ERANGE;
-			msg_warn ("cannot parse port: %s, error: %s", tokens[1], *err_str, strerror (errno));
-			goto err;
+		if (priority != NULL) {
+			if (port != NULL) {
+				cur_tok = tokens[2];
+			}
+			else {
+				cur_tok = tokens[1];
+			}
+			if (cur_tok != NULL) {
+				/* Priority part */
+				errno = 0;
+				priority_parsed = strtoul (cur_tok, &err_str, 10);
+				if (*err_str != '\0' || errno != 0) {
+					msg_warn ("cannot parse priority: %s, at symbol %c, error: %s", tokens[1], *err_str, strerror (errno));
+					goto err;
+				}
+				*priority = priority_parsed;
+			}
 		}
-		*port = port_parsed;
 	}
 	
 	/* Restore errno */
@@ -97,6 +117,18 @@ err:
 	return FALSE;
 }
 
+gboolean
+parse_host_port (const gchar *str, struct in_addr *ina, guint16 *port)
+{
+	return parse_host_port_priority (str, ina, port, NULL);
+}
+
+gboolean
+parse_host_priority (const gchar *str, struct in_addr *ina, guint *priority)
+{
+	return parse_host_port_priority (str, ina, NULL, priority);
+}
+
 gint
 parse_bind_line (struct config_file *cfg, struct worker_conf *cf, gchar *str)
 {
diff --git a/src/lua/CMakeLists.txt b/src/lua/CMakeLists.txt
index 70b3f7bc1..e46b5c885 100644
--- a/src/lua/CMakeLists.txt
+++ b/src/lua/CMakeLists.txt
@@ -9,7 +9,8 @@ SET(LUASRC			  lua_common.c
 					  lua_cdb.c
 					  lua_xmlrpc.c
 					  lua_http.c
-					  lua_redis.c)
+					  lua_redis.c
+					  lua_upstream.c)
 
 ADD_LIBRARY(rspamd_lua STATIC ${LUASRC})
 TARGET_LINK_LIBRARIES(rspamd_lua ${LUALIB})
diff --git a/src/lua/lua_common.c b/src/lua/lua_common.c
index 4d90048ca..8fbff979c 100644
--- a/src/lua/lua_common.c
+++ b/src/lua/lua_common.c
@@ -252,6 +252,7 @@ init_lua (struct config_file *cfg)
 	(void)luaopen_xmlrpc (L);
 	(void)luaopen_http (L);
 	(void)luaopen_redis (L);
+	(void)luaopen_upstream (L);
 	cfg->lua_state = L;
 	memory_pool_add_destructor (cfg->cfg_pool, (pool_destruct_func)lua_close, L);
 
diff --git a/src/lua/lua_common.h b/src/lua/lua_common.h
index 32604cbc4..adb87135e 100644
--- a/src/lua/lua_common.h
+++ b/src/lua/lua_common.h
@@ -17,7 +17,7 @@
 extern const luaL_reg null_reg[];
 extern GMutex *lua_mtx;
 
-#define RSPAMD_LUA_API_VERSION 9
+#define RSPAMD_LUA_API_VERSION 10
 
 /* Common utility functions */
 
@@ -61,6 +61,7 @@ gint luaopen_cdb (lua_State *L);
 gint luaopen_xmlrpc (lua_State * L);
 gint luaopen_http (lua_State * L);
 gint luaopen_redis (lua_State * L);
+gint luaopen_upstream (lua_State * L);
 void init_lua (struct config_file *cfg);
 gboolean init_lua_filters (struct config_file *cfg);
 
diff --git a/src/lua/lua_upstream.c b/src/lua/lua_upstream.c
new file mode 100644
index 000000000..5c72f7949
--- /dev/null
+++ b/src/lua/lua_upstream.c
@@ -0,0 +1,521 @@
+/* Copyright (c) 2010-2011, Vsevolod Stakhov
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *       * Redistributions of source code must retain the above copyright
+ *         notice, this list of conditions and the following disclaimer.
+ *       * Redistributions in binary form must reproduce the above copyright
+ *         notice, this list of conditions and the following disclaimer in the
+ *         documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED ''AS IS'' AND ANY
+ * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL AUTHOR BE LIABLE FOR ANY
+ * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "config.h"
+#include "lua_common.h"
+#include "upstream.h"
+#include "cfg_file.h"
+
+/* Upstream timeouts */
+#define DEFAULT_UPSTREAM_ERROR_TIME 10
+#define DEFAULT_UPSTREAM_DEAD_TIME 300
+#define DEFAULT_UPSTREAM_MAXERRORS 10
+
+/**
+ * This module implements upstreams manipulation from lua
+ */
+/* Upstream list functions */
+LUA_FUNCTION_DEF (upstream_list, create);
+LUA_FUNCTION_DEF (upstream_list, destroy);
+LUA_FUNCTION_DEF (upstream_list, get_upstream_by_hash);
+LUA_FUNCTION_DEF (upstream_list, get_upstream_round_robin);
+LUA_FUNCTION_DEF (upstream_list, get_upstream_master_slave);
+
+static const struct luaL_reg  upstream_list_m[] = {
+
+	LUA_INTERFACE_DEF (upstream_list, get_upstream_by_hash),
+	LUA_INTERFACE_DEF (upstream_list, get_upstream_round_robin),
+	LUA_INTERFACE_DEF (upstream_list, get_upstream_master_slave),
+	{"__tostring", lua_class_tostring},
+	{"__gc", lua_upstream_list_destroy},
+	{NULL, NULL}
+};
+static const struct luaL_reg  upstream_list_f[] = {
+	LUA_INTERFACE_DEF (upstream_list, create),
+	{NULL, NULL}
+};
+
+/* Upstream functions */
+LUA_FUNCTION_DEF (upstream, create);
+LUA_FUNCTION_DEF (upstream, destroy);
+LUA_FUNCTION_DEF (upstream, ok);
+LUA_FUNCTION_DEF (upstream, fail);
+LUA_FUNCTION_DEF (upstream, get_ip);
+LUA_FUNCTION_DEF (upstream, get_port);
+LUA_FUNCTION_DEF (upstream, get_ip_string);
+LUA_FUNCTION_DEF (upstream, get_priority);
+
+static const struct luaL_reg  upstream_m[] = {
+	LUA_INTERFACE_DEF (upstream, ok),
+	LUA_INTERFACE_DEF (upstream, fail),
+	LUA_INTERFACE_DEF (upstream, get_ip),
+	LUA_INTERFACE_DEF (upstream, get_ip_string),
+	LUA_INTERFACE_DEF (upstream, get_port),
+	LUA_INTERFACE_DEF (upstream, get_priority),
+	{"__tostring", lua_class_tostring},
+	{"__gc", lua_upstream_destroy},
+	{NULL, NULL}
+};
+static const struct luaL_reg  upstream_f[] = {
+	LUA_INTERFACE_DEF (upstream, create),
+	{NULL, NULL}
+};
+
+/* Upstream class */
+struct lua_upstream {
+	struct upstream up;
+	gchar *def;
+	guint16 port;
+	struct in_addr addr;
+};
+
+static struct lua_upstream	*
+lua_check_upstream (lua_State * L)
+{
+	void									*ud = luaL_checkudata (L, 1, "rspamd{upstream}");
+
+	luaL_argcheck (L, ud != NULL, 1, "'upstream' expected");
+	return ud ? *((struct lua_upstream **)ud) : NULL;
+}
+
+/**
+ * Create new upstream from its string definition like 'ip[:port[:priority]]' or 'host[:port[:priority]]'
+ * @param L
+ * @return upstream structure
+ */
+static gint
+lua_upstream_create (lua_State *L)
+{
+	struct lua_upstream						*new, **pnew;
+	const gchar								*def;
+
+	def = luaL_checkstring (L, 1);
+	if (def) {
+		new = g_slice_alloc0 (sizeof (struct lua_upstream));
+		new->def = g_strdup (def);
+		if (!parse_host_port_priority (new->def, &new->addr, &new->port, &new->up.priority)) {
+			g_free (new->def);
+			g_slice_free1 (sizeof (struct lua_upstream), new);
+			lua_pushnil (L);
+		}
+		else {
+			pnew = lua_newuserdata (L, sizeof (struct lua_upstream *));
+			lua_setclass (L, "rspamd{upstream}", -1);
+			*pnew = new;
+		}
+	}
+
+	return 1;
+}
+
+/**
+ * Destroy a single upstream object
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_destroy (lua_State *L)
+{
+	struct lua_upstream						*up = lua_check_upstream (L);
+
+	if (up) {
+		g_free (up->def);
+		g_slice_free1 (sizeof (struct lua_upstream), up);
+	}
+
+	return 0;
+}
+
+/**
+ * Get ip of upstream in numeric form (guint32)
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_get_ip (lua_State *L)
+{
+	struct lua_upstream						*up = lua_check_upstream (L);
+
+	if (up) {
+		lua_pushinteger (L, up->addr.s_addr);
+	}
+	else {
+		lua_pushnil (L);
+	}
+
+	return 1;
+}
+
+/**
+ * Get ip of upstream in string form
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_get_ip_string (lua_State *L)
+{
+	struct lua_upstream						*up = lua_check_upstream (L);
+
+	if (up) {
+		lua_pushstring (L, inet_ntoa (up->addr));
+	}
+	else {
+		lua_pushnil (L);
+	}
+
+	return 1;
+}
+
+/**
+ * Get port of upstream in numeric form
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_get_port (lua_State *L)
+{
+	struct lua_upstream						*up = lua_check_upstream (L);
+
+	if (up) {
+		lua_pushinteger (L, up->port);
+	}
+	else {
+		lua_pushnil (L);
+	}
+
+	return 1;
+}
+
+/**
+ * Get port of upstream in numeric form
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_get_priority (lua_State *L)
+{
+	struct lua_upstream						*up = lua_check_upstream (L);
+
+	if (up) {
+		lua_pushinteger (L, up->up.priority);
+	}
+	else {
+		lua_pushnil (L);
+	}
+
+	return 1;
+}
+
+/**
+ * Make upstream fail, the second argument is time, if absent the current time is used
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_fail (lua_State *L)
+{
+	struct lua_upstream						*up = lua_check_upstream (L);
+	time_t									 now;
+
+	if (up) {
+		if (lua_gettop (L) >= 2) {
+			now = luaL_checkinteger (L, 2);
+		}
+		else {
+			now = time (NULL);
+		}
+		upstream_fail (&up->up, now);
+	}
+
+	return 0;
+}
+
+/**
+ * Make upstream success, the second argument is time, if absent the current time is used
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_ok (lua_State *L)
+{
+	struct lua_upstream						*up = lua_check_upstream (L);
+	time_t									 now;
+
+	if (up) {
+		if (lua_gettop (L) >= 2) {
+			now = luaL_checkinteger (L, 2);
+		}
+		else {
+			now = time (NULL);
+		}
+		upstream_ok (&up->up, now);
+	}
+
+	return 0;
+}
+
+/* Upstream list class */
+struct lua_upstream_list {
+	struct lua_upstream *upstreams;
+	guint count;
+};
+
+static struct lua_upstream_list	*
+lua_check_upstream_list (lua_State * L)
+{
+	void									*ud = luaL_checkudata (L, 1, "rspamd{upstream_list}");
+
+	luaL_argcheck (L, ud != NULL, 1, "'upstream_list' expected");
+	return ud ? *((struct lua_upstream_list **)ud) : NULL;
+}
+
+/**
+ * Create new upstream list from its string definition like '<upstream>,<upstream>;<upstream>'
+ * @param L
+ * @return upstream list structure
+ */
+static gint
+lua_upstream_list_create (lua_State *L)
+{
+	struct lua_upstream_list				*new, **pnew;
+	struct lua_upstream						*cur;
+	const gchar								*def;
+	char									**tokens;
+	guint									 i, default_port = 0;
+
+	def = luaL_checkstring (L, 1);
+	if (def) {
+		if (lua_gettop (L) >= 2) {
+			default_port = luaL_checkinteger (L, 2);
+		}
+		new = g_slice_alloc0 (sizeof (struct lua_upstream_list));
+
+		tokens = g_strsplit_set (def, ",;", 0);
+		if (!tokens || !tokens[0]) {
+			goto err;
+		}
+		new->count = g_strv_length (tokens);
+		new->upstreams = g_slice_alloc0 (new->count * sizeof (struct lua_upstream));
+
+		for (i = 0; i < new->count; i ++) {
+			cur = &new->upstreams[i];
+			if (!parse_host_port_priority (tokens[i], &cur->addr, &cur->port, &cur->up.priority)) {
+				goto err;
+			}
+			if (cur->port == 0) {
+				cur->port = default_port;
+			}
+		}
+		pnew = lua_newuserdata (L, sizeof (struct upstream_list *));
+		lua_setclass (L, "rspamd{upstream_list}", -1);
+		*pnew = new;
+	}
+
+	return 1;
+err:
+	if (tokens) {
+		g_strfreev (tokens);
+	}
+	if (new->upstreams) {
+		g_slice_free1 (new->count * sizeof (struct lua_upstream), new->upstreams);
+	}
+	g_slice_free1 (sizeof (struct lua_upstream_list), new);
+	lua_pushnil (L);
+	return 1;
+}
+
+/**
+ * Destroy a single upstream list object
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_list_destroy (lua_State *L)
+{
+	struct lua_upstream_list					*upl = lua_check_upstream_list (L);
+
+	if (upl) {
+		if (upl->upstreams) {
+			g_slice_free1 (upl->count * sizeof (struct lua_upstream), upl->upstreams);
+		}
+		g_slice_free1 (sizeof (struct lua_upstream_list), upl);
+	}
+
+	return 0;
+}
+
+/**
+ * Get upstream by hash from key, params are: key and time (optional)
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_list_get_upstream_by_hash (lua_State *L)
+{
+	struct lua_upstream_list					*upl;
+	struct lua_upstream							*selected, **pselected;
+	time_t										 now;
+	const gchar									*key;
+
+	upl = lua_check_upstream_list (L);
+	if (upl) {
+		key = luaL_checkstring (L, 2);
+		if (key) {
+			if (lua_gettop (L) >= 3) {
+				now = luaL_checkinteger (L, 3);
+			}
+			else {
+				now = time (NULL);
+			}
+			selected = (struct lua_upstream *)get_upstream_by_hash (upl->upstreams, upl->count,
+					sizeof (struct lua_upstream), now,
+					DEFAULT_UPSTREAM_ERROR_TIME, DEFAULT_UPSTREAM_DEAD_TIME, DEFAULT_UPSTREAM_MAXERRORS,
+					key, 0);
+			if (selected) {
+				pselected = lua_newuserdata (L, sizeof (struct lua_upstream *));
+				lua_setclass (L, "rspamd{upstream}", -1);
+				*pselected = selected;
+			}
+			else {
+				lua_pushnil (L);
+			}
+		}
+		else {
+			lua_pushnil (L);
+		}
+	}
+	else {
+		lua_pushnil (L);
+	}
+
+	return 1;
+}
+
+/**
+ * Get upstream round robin (by current weight), params are: time (optional)
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_list_get_upstream_round_robin (lua_State *L)
+{
+	struct lua_upstream_list					*upl;
+	struct lua_upstream							*selected, **pselected;
+	time_t										 now;
+
+	upl = lua_check_upstream_list (L);
+	if (upl) {
+		if (lua_gettop (L) >= 2) {
+			now = luaL_checkinteger (L, 2);
+		}
+		else {
+			now = time (NULL);
+		}
+		selected = (struct lua_upstream *)get_upstream_round_robin (upl->upstreams, upl->count,
+				sizeof (struct lua_upstream), now,
+				DEFAULT_UPSTREAM_ERROR_TIME, DEFAULT_UPSTREAM_DEAD_TIME, DEFAULT_UPSTREAM_MAXERRORS);
+		if (selected) {
+			pselected = lua_newuserdata (L, sizeof (struct lua_upstream *));
+			lua_setclass (L, "rspamd{upstream}", -1);
+			*pselected = selected;
+		}
+		else {
+			lua_pushnil (L);
+		}
+	}
+	else {
+		lua_pushnil (L);
+	}
+
+	return 1;
+}
+
+/**
+ * Get upstream master slave order (by static priority), params are: time (optional)
+ * @param L
+ * @return
+ */
+static gint
+lua_upstream_list_get_upstream_master_slave (lua_State *L)
+{
+	struct lua_upstream_list					*upl;
+	struct lua_upstream							*selected, **pselected;
+	time_t										 now;
+
+	upl = lua_check_upstream_list (L);
+	if (upl) {
+		if (lua_gettop (L) >= 2) {
+			now = luaL_checkinteger (L, 2);
+		}
+		else {
+			now = time (NULL);
+		}
+		selected = (struct lua_upstream *)get_upstream_master_slave (upl->upstreams, upl->count,
+				sizeof (struct lua_upstream), now,
+				DEFAULT_UPSTREAM_ERROR_TIME, DEFAULT_UPSTREAM_DEAD_TIME, DEFAULT_UPSTREAM_MAXERRORS);
+		if (selected) {
+			pselected = lua_newuserdata (L, sizeof (struct lua_upstream *));
+			lua_setclass (L, "rspamd{upstream}", -1);
+			*pselected = selected;
+		}
+		else {
+			lua_pushnil (L);
+		}
+	}
+	else {
+		lua_pushnil (L);
+	}
+
+	return 1;
+}
+
+
+gint
+luaopen_upstream (lua_State * L)
+{
+	luaL_newmetatable (L, "rspamd{upstream_list}");
+	lua_pushstring (L, "__index");
+	lua_pushvalue (L, -2);
+	lua_settable (L, -3);
+
+	lua_pushstring (L, "class");
+	lua_pushstring (L, "rspamd{upstream_list}");
+	lua_rawset (L, -3);
+
+	luaL_openlib (L, NULL, upstream_list_m, 0);
+	luaL_openlib (L, "upstream_list", upstream_list_f, 0);
+
+	luaL_newmetatable (L, "rspamd{upstream}");
+	lua_pushstring (L, "__index");
+	lua_pushvalue (L, -2);
+	lua_settable (L, -3);
+
+	lua_pushstring (L, "class");
+	lua_pushstring (L, "rspamd{upstream}");
+	lua_rawset (L, -3);
+
+	luaL_openlib (L, NULL, upstream_m, 0);
+	luaL_openlib (L, "upstream", upstream_f, 0);
+
+	return 1;
+}
diff --git a/src/upstream.c b/src/upstream.c
index b3f282d59..398f80818 100644
--- a/src/upstream.c
+++ b/src/upstream.c
@@ -38,6 +38,7 @@ pthread_rwlock_t                upstream_mtx = PTHREAD_RWLOCK_INITIALIZER;
 #endif
 
 #define MAX_TRIES 20
+#define HASH_COMPAT
 
 /*
  * Poly: 0xedb88320
@@ -248,17 +249,26 @@ get_upstream_by_number (void *ups, size_t members, size_t msize, gint selected)
  * Get hash key for specified key (perl hash)
  */
 static                          guint32
-get_hash_for_key (guint32 hash, gchar *key, size_t keylen)
+get_hash_for_key (guint32 hash, const gchar *key, size_t keylen)
 {
 	guint32                         h, index;
 	const gchar                     *end = key + keylen;
 
 	h = ~hash;
 
-	while (key < end) {
-		index = (h ^ (u_char) * key) & 0x000000ffU;
-		h = (h >> 8) ^ crc32lookup[index];
-		++key;
+	if (end != key) {
+		while (key < end) {
+			index = (h ^ (u_char) * key) & 0x000000ffU;
+			h = (h >> 8) ^ crc32lookup[index];
+			++key;
+		}
+	}
+	else {
+		while (*key) {
+			index = (h ^ (u_char) * key) & 0x000000ffU;
+			h = (h >> 8) ^ crc32lookup[index];
+			++key;
+		}
 	}
 
 	return (~h);
@@ -268,7 +278,8 @@ get_hash_for_key (guint32 hash, gchar *key, size_t keylen)
  * Recheck all upstreams and return random active upstream
  */
 struct upstream                *
-get_random_upstream (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout, time_t revive_timeout, size_t max_errors)
+get_random_upstream (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout,
+		time_t revive_timeout, size_t max_errors)
 {
 	gint                            alive, selected;
 
@@ -282,7 +293,8 @@ get_random_upstream (void *ups, size_t members, size_t msize, time_t now, time_t
  * Return upstream by hash, that is calculated from active upstreams number
  */
 struct upstream                *
-get_upstream_by_hash (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout, time_t revive_timeout, size_t max_errors, gchar *key, size_t keylen)
+get_upstream_by_hash (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout,
+		time_t revive_timeout, size_t max_errors, const gchar *key, size_t keylen)
 {
 	gint                            alive, tries = 0, r;
 	guint32                         h = 0, ht;
@@ -332,7 +344,8 @@ get_upstream_by_hash (void *ups, size_t members, size_t msize, time_t now, time_
  * Recheck all upstreams and return upstream in round-robin order according to weight and priority
  */
 struct upstream                *
-get_upstream_round_robin (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout, time_t revive_timeout, size_t max_errors)
+get_upstream_round_robin (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout,
+		time_t revive_timeout, size_t max_errors)
 {
 	guint                           max_weight, i;
 	struct upstream                *cur, *selected = NULL;
@@ -381,7 +394,8 @@ get_upstream_round_robin (void *ups, size_t members, size_t msize, time_t now, t
  * Recheck all upstreams and return upstream in round-robin order according to only priority (master-slaves)
  */
 struct upstream                *
-get_upstream_master_slave (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout, time_t revive_timeout, size_t max_errors)
+get_upstream_master_slave (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout,
+		time_t revive_timeout, size_t max_errors)
 {
 	guint                           max_weight, i;
 	struct upstream                *cur, *selected = NULL;
@@ -459,7 +473,8 @@ upstream_ketama_add (struct upstream *up, gchar *up_key, size_t keylen, size_t k
  * Return upstream by hash and find nearest ketama point in some server
  */
 struct upstream                *
-get_upstream_by_hash_ketama (void *ups, size_t members, size_t msize, time_t now, time_t error_timeout, time_t revive_timeout, size_t max_errors, gchar *key, size_t keylen)
+get_upstream_by_hash_ketama (void *ups, size_t members, size_t msize,
+		time_t now, time_t error_timeout, time_t revive_timeout, size_t max_errors, const gchar *key, size_t keylen)
 {
 	guint                           alive, i;
 	guint32                         h = 0, step, middle, d, min_diff = UINT_MAX;
diff --git a/src/upstream.h b/src/upstream.h
index 101476bdf..da0a00013 100644
--- a/src/upstream.h
+++ b/src/upstream.h
@@ -4,7 +4,7 @@
 #include <sys/types.h>
 #include <stdint.h>
 
-/*
+/**
  * Structure of generic upstream
  */
 struct upstream {
@@ -17,7 +17,7 @@ struct upstream {
 	size_t ketama_points_size;			/**< Ketama array size			*/
 };
 
-/*
+/**
  * Upstream error logic
  * 1. During error time we count upstream_ok and upstream_fail
  * 2. If failcount is more then maxerrors then we mark upstream as unavailable for dead time
@@ -25,27 +25,27 @@ struct upstream {
  * 4. If all upstreams are dead, marks every upstream as alive
  */
 
-/*
+/**
  * Add an error to an upstream
  */
 void upstream_fail (struct upstream *up, time_t now);
 
-/*
+/**
  * Increase upstream successes count
  */
 void upstream_ok (struct upstream *up, time_t now);
 
-/*
+/**
  * Make all upstreams alive
  */
 void revive_all_upstreams (void *ups, size_t members, size_t msize);
 
-/*
+/**
  * Add ketama points for upstream
  */
 gint upstream_ketama_add (struct upstream *up, gchar *up_key, size_t keylen, size_t keypoints);
 
-/*
+/**
  * Get a random upstream from array of upstreams
  * @param ups array of structures that contains struct upstream as their first element
  * @param members number of elements in array
@@ -59,7 +59,7 @@ struct upstream* get_random_upstream   (void *ups, size_t members, size_t msize,
 										time_t now, time_t error_timeout, 
 										time_t revive_timeout, size_t max_errors);
 
-/*
+/**
  * Get upstream based on hash from array of upstreams
  * @param ups array of structures that contains struct upstream as their first element
  * @param members number of elements in array
@@ -74,9 +74,9 @@ struct upstream* get_random_upstream   (void *ups, size_t members, size_t msize,
 struct upstream* get_upstream_by_hash  (void *ups, size_t members, size_t msize, 
 										time_t now,  time_t error_timeout, 
 										time_t revive_timeout, size_t max_errors,
-										gchar *key, size_t keylen);
+										const gchar *key, size_t keylen);
 
-/*
+/**
  * Get an upstream from array of upstreams based on its current weight
  * @param ups array of structures that contains struct upstream as their first element
  * @param members number of elements in array
@@ -90,7 +90,7 @@ struct upstream* get_upstream_round_robin (void *ups, size_t members, size_t msi
 										time_t now, time_t error_timeout, 
 										time_t revive_timeout, size_t max_errors);
 
-/*
+/**
  * Get upstream based on hash from array of upstreams, this functions is using ketama algorithm
  * @param ups array of structures that contains struct upstream as their first element
  * @param members number of elements in array
@@ -104,9 +104,9 @@ struct upstream* get_upstream_round_robin (void *ups, size_t members, size_t msi
  */
 struct upstream* get_upstream_by_hash_ketama (void *ups, size_t members, size_t msize, time_t now, 
 										time_t error_timeout, time_t revive_timeout, size_t max_errors,
-										gchar *key, size_t keylen);
+										const gchar *key, size_t keylen);
 
-/*
+/**
  * Get an upstream from array of upstreams based on its current priority (not weight)
  * @param ups array of structures that contains struct upstream as their first element
  * @param members number of elements in array