Added support for connecting the hub to an upstream master hub (slave mode).

.gitignore

@@ -14,6 +14,7 @@ uhub-admin
 adcrush
 uhub
 build-stamp
+build.ninja
 debian/files
 debian/uhub.debhelper.log
 debian/uhub.postinst.debhelper
@@ -22,3 +23,4 @@ debian/uhub.prerm.debhelper
 debian/uhub.substvars
 uhub-passwd
 src/version.h
+

CMakeLists.txt

@@ -36,6 +36,10 @@ if (SSL_SUPPORT)
 	endif()
 endif()
 
+if (LINK_SUPPORT)
+	add_definitions(-DLINK_SUPPORT)
+endif()
+
 if (SYSTEMD_SUPPORT)
         INCLUDE(FindPkgConfig)
         pkg_search_module(SD_DAEMON REQUIRED libsystemd-daemon)
@@ -200,7 +204,7 @@ if (RELEASE)
 	add_definitions(-DNDEBUG)
 else()
 	set(CMAKE_BUILD_TYPE Debug)
-	# add_definitions(-DDEBUG)
+	add_definitions(-DDEBUG)
 endif()
 
 if (UNIX)

autotest/test.c

@@ -761,9 +761,6 @@ int main(int argc, char** argv)
 	exotic_add_test(&handle, &exotic_test_timer_add_5_events_1, "timer_add_5_events_1");
 	exotic_add_test(&handle, &exotic_test_timer_check_5_events_1, "timer_check_5_events_1");
 	exotic_add_test(&handle, &exotic_test_timer_process_5_events_1, "timer_process_5_events_1");
-	exotic_add_test(&handle, &exotic_test_timer_clear_1, "timer_clear_1");
-	exotic_add_test(&handle, &exotic_test_timer_bulk_1, "timer_bulk_1");
-	exotic_add_test(&handle, &exotic_test_timer_bulk_2, "timer_bulk_2");
 	exotic_add_test(&handle, &exotic_test_tokenizer_basic_0, "tokenizer_basic_0");
 	exotic_add_test(&handle, &exotic_test_tokenizer_basic_1, "tokenizer_basic_1");
 	exotic_add_test(&handle, &exotic_test_tokenizer_basic_1a, "tokenizer_basic_1a");
@@ -954,6 +951,7 @@ int exotic_initialize(struct exotic_handle* handle, int argc, char** argv)
 void exotic_add_test(struct exotic_handle* handle, exo_test_t func, const char* name)
 {
 	struct exo_test_data* test;
+
 	if (!handle)
 	{
 		fprintf(stderr, "exotic_add_test: failed, no handle!\n");

autotest/test_timer.tcc

@@ -1,6 +1,6 @@
 #include <uhub.h>
 
-#define MAX_EVENTS 100
+#define MAX_EVENTS 15
 static struct timeout_queue* g_queue;
 static time_t g_now;
 static size_t g_max;
@@ -117,28 +117,3 @@ EXO_TEST(timer_process_5_events_1,{
 	g_now = 4;
 	return timeout_queue_process(g_queue, g_now) == g_triggered;
 });
-
-EXO_TEST(timer_clear_1,{
-	size_t n;
-	for (n = 0; n < MAX_EVENTS; n++)
-		timeout_queue_remove(g_queue, &g_events[n]);
-	return timeout_queue_get_next_timeout(g_queue, g_now) == g_max;
-});
-
-EXO_TEST(timer_bulk_1,{
-	size_t n;
-	g_now = 10;
-	for (n = 0; n < MAX_EVENTS; n++)
-	{
-		timeout_queue_insert(g_queue, &g_events[0], 0);
-	}
-	return timeout_queue_process(g_queue, g_now) == 10;
-});
-
-EXO_TEST(timer_bulk_2,{
-	g_now = 110;
-	return timeout_queue_process(g_queue, g_now) == 90;
-});
-
-
-

src/adc/adcconst.h

@@ -1,6 +1,6 @@
 /*
  * uhub - A tiny ADC p2p connection hub
- * Copyright (C) 2007-2012, Jan Vidar Krey
+ * Copyright (C) 2007-2013, Jan Vidar Krey
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -34,6 +34,7 @@ typedef uint32_t fourcc_t;
 
 /* default welcome protocol support message, as sent by this server */
 #define ADC_PROTO_SUPPORT "ADBASE ADTIGR ADPING ADUCMD"
+#define ADC_PROTO_LINK_SUPPORT "ADTIGR ADLINK"
 
 /* Server sent commands */
 #define ADC_CMD_ISID FOURCC('I','S','I','D')
@@ -101,6 +102,12 @@ typedef uint32_t fourcc_t;
 #define ADC_CMD_HCMD FOURCC('H','C','M','D')
 #define ADC_CMD_ICMD FOURCC('I','C','M','D')
 
+/* Link commands */
+#define ADC_CMD_LSUP FOURCC('L','S','U','P') /* Link support handshake */
+#define ADC_CMD_LINF FOURCC('L','I','N','F') /* Hub link info */
+#define ADC_CMD_LGPA FOURCC('L','G','P','A') /* Hub link get password */
+#define ADC_CMD_LPAS FOURCC('L','P','A','S') /* Hub link password */
+#define ADC_CMD_LSTA FOURCC('L','S','T','A') /* Hub link status */
 
 #define ADC_INF_FLAG_IPV4_ADDR          "I4" /* ipv4 address */
 #define ADC_INF_FLAG_IPV6_ADDR          "I6" /* ipv6 address */

src/adc/message.c

@@ -1,6 +1,6 @@
 /*
  * uhub - A tiny ADC p2p connection hub
- * Copyright (C) 2007-2010, Jan Vidar Krey
+ * Copyright (C) 2007-2013, Jan Vidar Krey
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -178,6 +178,7 @@ int adc_msg_get_arg_offset(struct adc_message* msg)
 		
 		case 'I':
 		case 'H':
+		case 'L':
 			return 4;
 			
 		case 'B':
@@ -383,8 +384,9 @@ struct adc_message* adc_msg_parse(const char* line, size_t length)
 			ok = 0;
 			break;
 
-		case 'I':
-		case 'H':
+		case 'I': /* Hub to client */
+		case 'H': /* Clien to hub */
+		case 'L': /* hub to hub Link */
 			ok = (length > 3);
 			break;
 
@@ -789,12 +791,15 @@ int adc_msg_add_argument(struct adc_message* cmd, const char* string)
 
 int adc_msg_add_argument_string(struct adc_message* cmd, const char* string)
 {
-	char* escaped = adc_msg_escape(string);
-	int ret = adc_msg_add_argument(cmd, escaped);
-	hub_free(escaped);
+	char* arg = adc_msg_escape(string);
+	int ret;
+	if (!arg) return -1;
+	ret = adc_msg_add_argument(cmd, arg);
+	hub_free(arg);
 	return ret;
 }
 
+
 char* adc_msg_get_argument(struct adc_message* cmd, int offset)
 {
 	char* start;
@@ -873,21 +878,23 @@ int adc_msg_get_argument_index(struct adc_message* cmd, const char prefix[2])
 	return -1;
 }
 
-int adc_msg_escape_length(const char* str)
+
+
+size_t adc_msg_escape_length(const char* str)
 {
-	int add = 0;
-	int n = 0;
+	size_t add = 0;
+	size_t n = 0;
 	for (; str[n]; n++)
 		if (str[n] == ' ' || str[n] == '\n' || str[n] == '\\') add++;
 	return n + add;
 }
 
 
-int adc_msg_unescape_length(const char* str)
+size_t adc_msg_unescape_length(const char* str)
 {
-	int add = 0;
-	int n = 0;
-	int escape = 0;
+	size_t add = 0;
+	size_t n = 0;
+	size_t escape = 0;
 	for (; str[n]; n++)
 	{
 		if (escape)
@@ -1003,3 +1010,20 @@ char* adc_msg_escape(const char* string)
 	return str;
 }
 
+enum msg_type adc_msg_get_type(const struct adc_message* msg)
+{
+	switch (msg->cache[0])
+	{
+		case 'B': return msg_type_client_broadcast;
+		case 'C': return msg_type_client_to_client;
+		case 'D': return msg_type_client_direct;
+		case 'E': return msg_type_client_echo;
+		case 'F': return msg_type_client_feature;
+		case 'H': return msg_type_client_to_hub;
+		case 'I': return msg_type_hub_to_client;
+		case 'L': return msg_type_link_to_link;
+		case 'U': return msg_type_hub_to_client_udp;
+	}
+	return msg_type_unknown;
+}
+

src/adc/message.h

@@ -1,6 +1,6 @@
 /*
  * uhub - A tiny ADC p2p connection hub
- * Copyright (C) 2007-2010, Jan Vidar Krey
+ * Copyright (C) 2007-2013, Jan Vidar Krey
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -43,6 +43,20 @@ enum msg_status_level
 	status_level_fatal = 2, /* Fatal error (disconnect) */
 };
 
+enum msg_type
+{
+	msg_type_unknown           = 0,
+	msg_type_client_broadcast  = 'B',
+	msg_type_client_to_client  = 'C',
+	msg_type_client_direct     = 'D',
+	msg_type_client_echo       = 'E',
+	msg_type_client_feature    = 'F',
+	msg_type_client_to_hub     = 'H',
+	msg_type_hub_to_client     = 'I',
+	msg_type_link_to_link      = 'L',
+	msg_type_hub_to_client_udp = 'U',
+};
+
 /**
  * Increase the reference counter for an ADC message struct.
  * NOTE: Always use the returned value, and not the passed value, as
@@ -172,7 +186,8 @@ extern int adc_msg_replace_named_argument(struct adc_message* cmd, const char pr
 extern int adc_msg_add_argument(struct adc_message* cmd, const char* string);
 
 /**
- * Add an argument string, the string will be automatcally escaped.
+ * Append a string argumnent.
+ * The string will automatcally be escaped.
  * @return  0 if successful, or -1 if an error occured (out of memory).
  */
 extern int adc_msg_add_argument_string(struct adc_message* cmd, const char* string);
@@ -215,13 +230,6 @@ extern char* adc_msg_unescape(const char* string);
  */
 extern int adc_msg_unescape_to_target(const char* string, char* target, size_t target_size);
 
-/**
- * Returns the length of the string once escaped with
- * adc_msg_escape().
- *
- * The string must be NULL terminated.
- */
-extern int adc_msg_escape_length(const char* str);
 
 /**
  * Convert a string to a ADC command escaped string.
@@ -229,6 +237,12 @@ extern int adc_msg_escape_length(const char* str);
  */
 extern char* adc_msg_escape(const char* string);
 
+/**
+ * Calculate the length str would be after escaping.
+ * Does not include any NULL terminator.
+ */
+size_t adc_msg_escape_length(const char* str);
+
 /**
  * This will ensure a newline is at the end of the command.
  */
@@ -247,5 +261,6 @@ void adc_msg_unterminate(struct adc_message* cmd);
  */
 int adc_msg_get_arg_offset(struct adc_message* msg);
 
+enum msg_type adc_msg_get_type(const struct adc_message* msg);
 
 #endif /* HAVE_UHUB_COMMAND_H */

src/core/commands.c

@@ -353,7 +353,6 @@ static int command_kick(struct command_base* cbase, struct hub_user* user, struc
 	}
 	else
 	{
-		hub_notify(cbase->hub, notify_info, "Kicking user \"%s\" (%s)", target->id.nick, user->id.nick);
 		cbuf_append_format(buf, "Kicking user \"%s\".", target->id.nick);
 		hub_disconnect_user(cbase->hub, target, quit_kicked);
 	}
@@ -363,7 +362,6 @@ static int command_kick(struct command_base* cbase, struct hub_user* user, struc
 static int command_reload(struct command_base* cbase, struct hub_user* user, struct hub_command* cmd)
 {
 	cbase->hub->status = hub_status_restart;
-	hub_notify(cbase->hub, notify_info, "Reloading configuration (%s)", user->id.nick);
 	return command_status(cbase, user, cmd, cbuf_create_const("Reloading configuration..."));
 }
 
@@ -386,7 +384,6 @@ static int command_unload(struct command_base* cbase, struct hub_user* user, str
 
 static int command_shutdown_hub(struct command_base* cbase, struct hub_user* user, struct hub_command* cmd)
 {
-	hub_notify(cbase->hub, notify_warn, "Shutting down hub (%s)", user->id.nick);
 	cbase->hub->status = hub_status_shutdown;
 	return command_status(cbase, user, cmd, cbuf_create_const("Hub shutting down..."));
 }

src/core/config.c

@@ -102,7 +102,7 @@ static int config_parse_line(char* line, int line_count, void* ptr_data)
 	data = strip_white_space(data);
 	data = strip_off_quotes(data);
 
-	if (!*key || !*data)
+	if (!*key /*|| !*data*/)
 	{
 		LOG_FATAL("Configuration parse error on line %d", line_count);
 		return -1;

src/core/config.xml

@@ -165,6 +165,36 @@
 		<since>0.3.2</since>
 	</option>
 
+	<option name="hub_link_enabled" type="boolean" default="0">
+		<short>Allow other hubs to link to this hub</short>
+		<description><![CDATA[
+			This allows multiple hubs to link to this hub so
+			that users on the different hubs appear as being on one hub.
+			This is useful for distributing or load balancing large hubs.
+		]]></description>
+		<since>0.5.0</since>
+		<ifdef>LINK_SUPPORT</ifdef>
+	</option>
+
+	<option name="hub_link_secret" type="string" default="">
+		<short>A secret token required to accept hub linking</short>
+		<description><![CDATA[
+			This should be a secret token needed to authorize hubs
+			linking into this one.
+		]]></description>
+		<since>0.5.0</since>
+		<ifdef>LINK_SUPPORT</ifdef>
+	</option>
+
+	<option name="hub_link_connect" type="string" default="">
+		<short>Connect this to hub to another hub</short>
+		<description><![CDATA[
+			The other hub must allow links to be established.
+			Example: uhub://host:port
+		]]></description>
+		<since>0.5.0</since>
+		<ifdef>LINK_SUPPORT</ifdef>
+	</option>
 
 	<option name="max_recv_buffer" type="int" default="4096" advanced="true" >
 		<check min="1024" max="1048576" />

src/core/gen_config.c

@@ -17,6 +17,15 @@ void config_defaults(struct hub_config* config)
 	config->hub_name = hub_strdup("uhub");
 	config->hub_description = hub_strdup("no description");
 	config->redirect_addr = hub_strdup("");
+#ifdef LINK_SUPPORT
+	config->hub_link_enabled = 0;
+#endif /* LINK_SUPPORT */
+#ifdef LINK_SUPPORT
+	config->hub_link_secret = hub_strdup("");
+#endif /* LINK_SUPPORT */
+#ifdef LINK_SUPPORT
+	config->hub_link_connect = hub_strdup("");
+#endif /* LINK_SUPPORT */
 	config->max_recv_buffer = 4096;
 	config->max_send_buffer = 131072;
 	config->max_send_buffer_soft = 98304;
@@ -245,6 +254,42 @@ static int apply_config(struct hub_config* config, char* key, char* data, int li
 		return 0;
 	}
 
+#ifdef LINK_SUPPORT
+	if (!strcmp(key, "hub_link_enabled"))
+	{
+		if (!apply_boolean(key, data, &config->hub_link_enabled))
+		{
+			LOG_ERROR("Configuration parse error on line %d", line_count);
+			return -1;
+		}
+		return 0;
+	}
+#endif /* LINK_SUPPORT */
+
+#ifdef LINK_SUPPORT
+	if (!strcmp(key, "hub_link_secret"))
+	{
+		if (!apply_string(key, data, &config->hub_link_secret, (char*) ""))
+		{
+			LOG_ERROR("Configuration parse error on line %d", line_count);
+			return -1;
+		}
+		return 0;
+	}
+#endif /* LINK_SUPPORT */
+
+#ifdef LINK_SUPPORT
+	if (!strcmp(key, "hub_link_connect"))
+	{
+		if (!apply_string(key, data, &config->hub_link_connect, (char*) ""))
+		{
+			LOG_ERROR("Configuration parse error on line %d", line_count);
+			return -1;
+		}
+		return 0;
+	}
+#endif /* LINK_SUPPORT */
+
 	if (!strcmp(key, "max_recv_buffer"))
 	{
 		min = 1024;
@@ -943,6 +988,14 @@ void free_config(struct hub_config* config)
 
 	hub_free(config->redirect_addr);
 
+#ifdef LINK_SUPPORT
+	hub_free(config->hub_link_secret);
+#endif /* LINK_SUPPORT */
+
+#ifdef LINK_SUPPORT
+	hub_free(config->hub_link_connect);
+#endif /* LINK_SUPPORT */
+
 	hub_free(config->tls_require_redirect_addr);
 
 	hub_free(config->tls_certificate);
@@ -1074,6 +1127,21 @@ void dump_config(struct hub_config* config, int ignore_defaults)
 	if (!ignore_defaults || strcmp(config->redirect_addr, "") != 0)
 		fprintf(stdout, "redirect_addr = \"%s\"\n", config->redirect_addr);
 
+#ifdef LINK_SUPPORT
+	if (!ignore_defaults || config->hub_link_enabled != 0)
+		fprintf(stdout, "hub_link_enabled = %s\n", config->hub_link_enabled ? "yes" : "no");
+#endif /* LINK_SUPPORT */
+
+#ifdef LINK_SUPPORT
+	if (!ignore_defaults || strcmp(config->hub_link_secret, "") != 0)
+		fprintf(stdout, "hub_link_secret = \"%s\"\n", config->hub_link_secret);
+#endif /* LINK_SUPPORT */
+
+#ifdef LINK_SUPPORT
+	if (!ignore_defaults || strcmp(config->hub_link_connect, "") != 0)
+		fprintf(stdout, "hub_link_connect = \"%s\"\n", config->hub_link_connect);
+#endif /* LINK_SUPPORT */
+
 	if (!ignore_defaults || config->max_recv_buffer != 4096)
 		fprintf(stdout, "max_recv_buffer = %d\n", config->max_recv_buffer);
 

src/core/gen_config.h

@@ -17,6 +17,15 @@ struct hub_config
 	char* hub_name;                        /*<<< Name of hub (default: "uhub") */
 	char* hub_description;                 /*<<< Short hub description, topic or subject. (default: "no description") */
 	char* redirect_addr;                   /*<<< A common hub redirect address. (default: "") */
+#ifdef LINK_SUPPORT
+	int   hub_link_enabled;                /*<<< Allow other hubs to link to this hub (default: 0) */
+#endif /* LINK_SUPPORT */
+#ifdef LINK_SUPPORT
+	char* hub_link_secret;                 /*<<< A secret token required to accept hub linking (default: "") */
+#endif /* LINK_SUPPORT */
+#ifdef LINK_SUPPORT
+	char* hub_link_connect;                /*<<< Connect this to hub to another hub (default: "") */
+#endif /* LINK_SUPPORT */
 	int   max_recv_buffer;                 /*<<< Max read buffer before parse, per user (default: 4096) */
 	int   max_send_buffer;                 /*<<< Max send buffer before disconnect, per user (default: 131072) */
 	int   max_send_buffer_soft;            /*<<< Max send buffer before message drops, per user (default: 98304) */

src/core/hub.c

@@ -653,7 +653,6 @@ static struct net_connection* start_listening_socket(const char* bind_addr, uint
 	if (ret == -1)
 	{
 		LOG_ERROR("hub_start_service(): Unable to bind to TCP local address. errno=%d, str=%s", net_error(), net_error_string(net_error()));
-		hub_notify(hub, notify_error, "Unable to bind to network address %s on port %d: %s (%d)", bind_addr, port, net_error_string(net_error()), net_error());
 		net_close(sd);
 		return 0;
 	}
@@ -672,60 +671,6 @@ static struct net_connection* start_listening_socket(const char* bind_addr, uint
 	return server;
 }
 
-
-int hub_is_running(struct hub_info* hub)
-{
-	return hub->status == hub_status_running || hub->status == hub_status_restart;
-}
-
-
-void hub_notify(struct hub_info* hub, enum notify_verbosity verbosity, const char* fmt, ...)
-{
-	struct cbuffer* buf;
-	struct adc_message* msg;
-	va_list args;
-	char temp[1024];
-
-	va_start(args, fmt);
-	vsnprintf(temp, sizeof(temp), fmt, args);
-	va_end(args);
-
-	buf = cbuf_create(strlen(temp) + 8);
-
-	switch (verbosity)
-	{
-		case notify_error:
-			cbuf_append(buf, "ERROR: ");
-			LOG_ERROR(temp);
-			break;
-		case notify_warn:
-			cbuf_append(buf, "WARN: ");
-			LOG_WARN(temp);
-			break;
-		case notify_info:
-			cbuf_append(buf, "INFO: ");
-			LOG_INFO(temp);
-			break;
-		case notify_debug:
-			cbuf_append(buf, "DEBUG: ");
-			LOG_DEBUG(temp);
-			break;
-	}
-
-	cbuf_append(buf, temp);
-
-	if (hub_is_running(hub))
-	{
-		msg = adc_msg_construct(ADC_CMD_IMSG, 5 + adc_msg_escape_length(cbuf_get(buf)) + 2);
-		adc_msg_add_argument_string(msg, cbuf_get(buf));
-		route_to_operators(hub, msg);
-		adc_msg_free(msg);
-	}
-
-	cbuf_destroy(buf);
-}
-
-
 struct server_alt_port_data
 {
 	struct hub_info* hub;
@@ -806,54 +751,6 @@ static void unload_ssl_certificates(struct hub_info* hub)
 }
 #endif /* SSL_SUPPORT */
 
-// #ifdef BOT_SUPPORT
-
-static void route_privmsg_to_operators(struct hub_user* bot, sid_t from, const char* escaped_msg, int action)
-{
-	struct hub_info* hub = bot->hub;
-	struct hub_user* user = (struct hub_user*) list_get_first(hub->users->list);
-	while (user)
-	{
-		if (from != user->id.sid && user_flag_get(user, flag_opnotify))
-		{
-			struct adc_message* msg = adc_msg_construct_source_dest(ADC_CMD_EMSG, from, user->id.sid, strlen(escaped_msg) + (action * 4) + 7);
-			adc_msg_add_argument(msg, escaped_msg);
-			adc_msg_add_named_argument(msg, ADC_MSG_FLAG_PRIVATE, sid_to_string(bot->id.sid));
-			if (action) adc_msg_add_named_argument(msg, ADC_MSG_FLAG_ACTION, "1");
-			route_to_user(hub, user, msg);
-			adc_msg_free(msg);
-		}
-		user = (struct hub_user*) list_get_next(hub->users->list);
-	}
-}
-
-/// This receives private messages and transmits them to the connected operators.
-static void hub_bot_op_notify_handle(struct hub_user* bot, struct adc_message* msg)
-{
-	char* chat;
-	LOG_TRACE("Invoked hub_bot_op_notify_handle()");
-	switch (msg->cmd)
-	{
-			case ADC_CMD_EMSG:
-			case ADC_CMD_DMSG:
-				chat = adc_msg_get_argument(msg, 0);
-				LOG_DEBUG("Hub chat: \"%s\"", chat);
-				route_privmsg_to_operators(bot, msg->source, chat, adc_msg_has_named_argument(msg, ADC_MSG_FLAG_ACTION) ? 1 : 0);
-				hub_free(chat);
-				break;
-			default:
-				/* ignore these messages! */
-				break;
-	}
-}
-
-static void hub_bot_op_notify_create(struct hub_info* hub)
-{
-	struct hub_user* opcom = user_create_bot(hub, "Operations", "Hub operators", hub_bot_op_notify_handle);
-	uman_add(hub->users, opcom);
-}
-// #endif
-
 struct hub_info* hub_start_service(struct hub_config* config)
 {
 	struct hub_info* hub = 0;
@@ -890,6 +787,13 @@ struct hub_info* hub_start_service(struct hub_config* config)
 	}
 #endif
 
+#ifdef LINK_SUPPORT
+	if (config->hub_link_enabled)
+	{
+		LOG_INFO("Hub linking support enabled");
+	}
+#endif
+
 	hub->config = config;
 	hub->users = NULL;
 
@@ -938,7 +842,14 @@ struct hub_info* hub_start_service(struct hub_config* config)
 	// Start the hub command sub-system
 	hub->commands = command_initialize(hub);
 
-	hub_bot_op_notify_create(hub);
+#ifdef LINK_SUPPORT
+	if (*config->hub_link_connect)
+	{
+		link_connect_uri(hub, config->hub_link_connect);
+	}
+#endif
+
+
 	return hub;
 }
 
@@ -1002,23 +913,14 @@ void hub_plugins_unload(struct hub_info* hub)
 
 void hub_set_variables(struct hub_info* hub, struct acl_handle* acl)
 {
-	char* tmp;
-	char* server = adc_msg_escape(PRODUCT_STRING); /* FIXME: OOM */
-
 	hub->acl = acl;
 	hub->command_info = adc_msg_construct(ADC_CMD_IINF, 15);
 	if (hub->command_info)
 	{
 		adc_msg_add_named_argument(hub->command_info, ADC_INF_FLAG_CLIENT_TYPE, ADC_CLIENT_TYPE_HUB);
-		adc_msg_add_named_argument(hub->command_info, ADC_INF_FLAG_USER_AGENT, server);
-
-		tmp = adc_msg_escape(hub->config->hub_name);
-		adc_msg_add_named_argument(hub->command_info, ADC_INF_FLAG_NICK, tmp);
-		hub_free(tmp);
-
-		tmp = adc_msg_escape(hub->config->hub_description);
-		adc_msg_add_named_argument(hub->command_info, ADC_INF_FLAG_DESCRIPTION, tmp);
-		hub_free(tmp);
+		adc_msg_add_named_argument_string(hub->command_info, ADC_INF_FLAG_USER_AGENT, PRODUCT_STRING);
+		adc_msg_add_named_argument_string(hub->command_info, ADC_INF_FLAG_NICK, hub->config->hub_name);
+		adc_msg_add_named_argument_string(hub->command_info, ADC_INF_FLAG_DESCRIPTION, hub->config->hub_description);
 	}
 
 	hub->command_support = adc_msg_construct(ADC_CMD_ISUP, 6 + strlen(ADC_PROTO_SUPPORT));
@@ -1027,16 +929,14 @@ void hub_set_variables(struct hub_info* hub, struct acl_handle* acl)
 		adc_msg_add_argument(hub->command_support, ADC_PROTO_SUPPORT);
 	}
 
-	hub->command_banner = adc_msg_construct(ADC_CMD_ISTA, 100 + strlen(server));
+	hub->command_banner = adc_msg_construct(ADC_CMD_ISTA, 100 + adc_msg_escape_length(PRODUCT_STRING));
 	if (hub->command_banner)
 	{
-		if (hub->config->show_banner_sys_info)
-			tmp = adc_msg_escape("Powered by " PRODUCT_STRING " on " OPSYS "/" CPUINFO);
-		else
-			tmp = adc_msg_escape("Powered by " PRODUCT_STRING);
 		adc_msg_add_argument(hub->command_banner, "000");
-		adc_msg_add_argument(hub->command_banner, tmp);
-		hub_free(tmp);
+		if (hub->config->show_banner_sys_info)
+			adc_msg_add_argument_string(hub->command_banner, "Powered by " PRODUCT_STRING " on " OPSYS "/" CPUINFO);
+		else
+			adc_msg_add_argument_string(hub->command_banner, "Powered by " PRODUCT_STRING);
 	}
 
 	if (hub_plugins_load(hub) < 0)
@@ -1047,7 +947,6 @@ void hub_set_variables(struct hub_info* hub, struct acl_handle* acl)
 	else
 
 	hub->status = (hub->config->hub_enabled ? hub_status_running : hub_status_disabled);
-	hub_free(server);
 }
 
 

src/core/hub.h

@@ -106,7 +106,7 @@ struct hub_info
 	struct adc_message* command_support; /* The hub's SUP command */
 	struct adc_message* command_banner;  /* The default welcome message */
 	time_t tm_started;
-	enum hub_state status;
+	int status;
 	char* recvbuf; /* Global receive buffer */
 	char* sendbuf; /* Global send buffer */
 
@@ -115,6 +115,10 @@ struct hub_info
 	struct command_base* commands;       /* Hub command handler */
 	struct uhub_plugins* plugins;        /* Plug-ins loaded for this hub instance. */
 
+#ifdef LINK_SUPPORT
+	struct linked_list* hub_links;       /* Other hubs linked to this hub */
+#endif
+
 #ifdef SSL_SUPPORT
 	struct ssl_context_handle* ctx;
 #endif /*  SSL_SUPPORT */
@@ -233,20 +237,6 @@ extern void hub_set_variables(struct hub_info* hub, struct acl_handle* acl);
  */
 extern void hub_free_variables(struct hub_info* hub);
 
-
-enum notify_verbosity
-{
-	notify_error = 0,
-	notify_warn  = 1,
-	notify_info  = 2,
-	notify_debug = 3,
-};
-
-/**
- * Send a notification message to operators.
- */
-extern void hub_notify(struct hub_info* hub, enum notify_verbosity verbosity, const char* fmt, ...);
-
 /**
  * Returns a string for the given status_message (See enum status_message).
  */
@@ -364,10 +354,6 @@ extern void hub_disconnect_user(struct hub_info* hub, struct hub_user* user, int
  */
 extern void hub_logout_log(struct hub_info* hub, struct hub_user* user);
 
-/**
- * Returns 1 if the hub is running, and 0 otherwise.
- */
-extern int hub_is_running(struct hub_info* hub);
 
 #endif /* HAVE_UHUB_HUB_H */
 

src/core/hubevent.c

@@ -31,10 +31,6 @@ void on_login_success(struct hub_info* hub, struct hub_user* u)
 	user_set_state(u, state_normal);
 	uman_add(hub->users, u);
 
-	// Make operators receive hub notifications by default.
-	if (user_is_protected(u))
-		user_flag_set(u, flag_opnotify);
-
 	/* Announce new user to all connected users */
 	if (user_is_logged_in(u))
 		route_info_message(hub, u);

src/core/ioqueue.c

@@ -49,6 +49,34 @@ void ioq_recv_destroy(struct ioq_recv* q)
 	}
 }
 
+#define IOQ_RECV_FLAGS_PREALLOC 1
+#define IOQ_RECV_FLAGS_FULL 2
+
+enum ioq_recv_status ioq_recv_read(struct ioq_recv* q, struct net_connection* con)
+{
+	static char buf[MAX_RECV_BUF];
+	size_t buf_size = ioq_recv_get(q, buf, MAX_RECV_BUF);
+	ssize_t size;
+
+	if (buf_size >= MAX_RECV_BUF)
+		return ioq_recv_full;
+
+	size = net_con_recv(con, buf + buf_size, MAX_RECV_BUF - buf_size);
+
+	if (size > 0)
+		buf_size += size;
+	if (size < 0)
+		return ioq_recv_error;
+	if (size == 0)
+		return ioq_recv_later;
+
+	ioq_recv_set(q, buf, buf_size);
+	return ioq_recv_ok;
+}
+
+
+
+
 size_t ioq_recv_get(struct ioq_recv* q, void* buf, size_t bufsize)
 {
 	uhub_assert(bufsize >= q->size);
@@ -88,6 +116,21 @@ size_t ioq_recv_set(struct ioq_recv* q, void* buf, size_t bufsize)
 }
 
 
+int ioq_recv_consume(struct ioq_recv* q, size_t bytes)
+{
+	size_t newsize;
+	void* ptr;
+
+	if (!q || bytes > q->size) return 0;
+
+	newsize = (q->size - bytes);
+	memmove(q->buf, q->buf + bytes, newsize);
+	ptr = hub_realloc(q->buf, newsize);
+	q->buf = ptr;
+	q->size = newsize;
+	return 1;
+}
+
 struct ioq_send* ioq_send_create()
 {
 	struct ioq_send* q = hub_malloc_zero(sizeof(struct ioq_send));

src/core/ioqueue.h

@@ -39,6 +39,7 @@ struct ioq_recv
 {
 	char* buf;
 	size_t size;
+	// int flags;
 };
 
 /**
@@ -102,5 +103,26 @@ extern size_t ioq_recv_set(struct ioq_recv*, void* buf, size_t bufsize);
 extern int ioq_recv_is_empty(struct ioq_recv* buf);
 
 
+enum ioq_recv_status
+{
+	ioq_recv_ok = 0,    // read data OK
+	ioq_recv_later = 1, // all OK, but call again later (no change)
+	ioq_recv_full  = 2, // all OK, but the buffer is full
+	ioq_recv_error = 3, // error (connection is not working)
+};
+
+/**
+ * Receive from connection into buffer.
+ */
+extern enum ioq_recv_status ioq_recv_read(struct ioq_recv* q, struct net_connection* con);
+
+/**
+ * Consume 'bytes' bytes.
+ * 'bytes' must be <= q->size
+ *
+ * @return 1 on success, or 0 on error (only if q == NULL or bytes is > q->size).
+ */
+extern int ioq_recv_consume(struct ioq_recv* q, size_t bytes);
+
 
 #endif /* HAVE_UHUB_IO_QUEUE_H */

src/core/link.c

@@ -0,0 +1,508 @@
+/*
+ * uhub - A tiny ADC p2p connection hub
+ * Copyright (C) 2007-2013, Jan Vidar Krey
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "uhub.h"
+
+#ifdef LINK_SUPPORT
+static int link_send_support(struct hub_link* link);
+
+static void link_net_event(struct net_connection* con, int event, void *arg)
+{
+	LOG_INFO("link_net_event(), event=%d", event);
+	struct hub_link* link = (struct hub_link*) arg;
+	struct hub_info* hub = link->hub;
+	int ret = 0;
+
+	if (event == NET_EVENT_TIMEOUT)
+	{
+		LOG_DEBUG("Hub link timeout!");
+	}
+
+	if (event & NET_EVENT_READ)
+	{
+		ret = link_handle_read(link);
+		if (ret < 0)
+		{
+			link_disconnect(link);
+			return;
+		}
+	}
+
+	if (event & NET_EVENT_WRITE)
+	{
+		ret = link_handle_write(link);
+		if (ret < 0)
+		{
+			link_disconnect(link);
+			return;
+		}
+	}
+}
+
+void link_disconnect(struct hub_link* link)
+{
+	if (link->connection)
+		net_con_close(link->connection);
+	link->connection = NULL;
+
+	ioq_send_destroy(link->send_queue);
+	ioq_recv_destroy(link->recv_queue);
+	link->send_queue = NULL;
+	link->recv_queue = NULL;
+
+	// FIXME: Notify hub and disconnect users!
+
+	hub_free(link);
+}
+
+static struct hub_link* link_create_internal(struct hub_info* hub)
+{
+	struct hub_link* link = NULL;
+
+	LOG_DEBUG("link_create_internal(), hub=%p");
+	link = (struct hub_link*) hub_malloc_zero(sizeof(struct hub_link));
+	if (link == NULL)
+		return NULL; /* OOM */
+
+	link->send_queue = ioq_send_create();
+	link->recv_queue = ioq_recv_create();
+
+	link->hub = hub;
+	link->state = state_protocol;
+	return link;
+}
+
+
+struct hub_link* link_create(struct hub_info* hub, struct net_connection* con, struct ip_addr_encap* addr)
+{
+	struct hub_link* link = link_create_internal(hub);
+	link->connection = con;
+	net_con_reinitialize(link->connection, link_net_event, link, NET_EVENT_READ);
+	link->mode = link_mode_server;
+	return link;
+}
+
+static void link_connect_callback(struct net_connect_handle* handle, enum net_connect_status status, struct net_connection* con, void* ptr)
+{
+	struct hub_link* link = (struct hub_link*) ptr;
+	link->connect_job = NULL;
+
+	LOG_DEBUG("link_connect_callback()");
+
+	switch (status)
+	{
+		case net_connect_status_ok:
+			link->connection = con;
+			net_con_reinitialize(link->connection, link_net_event, link, NET_EVENT_READ);
+			// FIXME: send handshake here
+			link_send_support(link);
+			break;
+
+		case net_connect_status_host_not_found:
+		case net_connect_status_no_address:
+		case net_connect_status_dns_error:
+		case net_connect_status_refused:
+		case net_connect_status_unreachable:
+		case net_connect_status_timeout:
+		case net_connect_status_socket_error:
+			// FIXME: Unable to connect - start timer and re-try connection establishment!
+			break;
+	}
+}
+
+struct link_address
+{
+	char host[256];
+	uint16_t port;
+};
+
+static int link_parse_address(const char* arg, struct link_address* addr)
+{
+	int port;
+	char* split;
+
+	memset(addr, 0, sizeof(struct link_address));
+
+	/* Split hostname and port (if possible) */
+	split = strrchr(arg, ':');
+	if (split == 0 || strlen(split) < 2 || strlen(split) > 6)
+		return 0;
+
+	/* Ensure port number is valid */
+	port = strtol(split+1, NULL, 10);
+	if (port <= 0 || port > 65535)
+		return 0;
+
+	memcpy(addr->host, arg, &split[0] - &arg[0]);
+	addr->port = port;
+	return 1;
+}
+
+
+struct hub_link* link_connect_uri(struct hub_info* hub, const char* address)
+{
+	struct link_address link_address;
+	if (!link_parse_address(address, &link_address))
+	{
+		LOG_INFO("Invalid master hub link address");
+		return NULL;
+	}
+
+	return link_connect(hub, link_address.host, link_address.port);
+}
+
+
+struct hub_link* link_connect(struct hub_info* hub, const char* address, uint16_t port)
+{
+	struct hub_link* link = link_create_internal(hub);
+
+	LOG_DEBUG("Connecting to master link at %s:%d...", address, port);
+
+	link->mode = link_mode_client;
+	link->connect_job = net_con_connect(address, port, link_connect_callback, link);
+	if (!link->connect_job)
+	{
+		// FIXME: Immediate failure!
+		LOG_DEBUG("Error connecting to master hub link.");
+		link_disconnect(link);
+		return NULL;
+	}
+
+	return link;
+}
+
+static int link_net_io_want_read(struct hub_link* link)
+{
+	net_con_update(link->connection, NET_EVENT_READ);
+}
+
+static int link_net_io_want_write(struct hub_link* link)
+{
+	net_con_update(link->connection, NET_EVENT_READ | NET_EVENT_WRITE);
+}
+
+
+int link_handle_write(struct hub_link* link)
+{
+	int ret = 0;
+	while (ioq_send_get_bytes(link->send_queue))
+	{
+		ret = ioq_send_send(link->send_queue, link->connection);
+		if (ret <= 0)
+			break;
+	}
+
+	if (ret < 0)
+		return -1; // FIXME! Extract socket error!
+
+	if (ioq_send_get_bytes(link->send_queue))
+		link_net_io_want_write(link);
+	else
+		link_net_io_want_read(link);
+	return 0;
+}
+
+
+int link_send_message(struct hub_link* link, struct adc_message* msg)
+{
+#ifdef DEBUG_SENDQ
+	char* data = strndup(msg->cache, msg->length-1);
+	LOG_PROTO("[link] send %p: \"%s\"", link, data);
+	free(data);
+#endif
+
+	if (!link->connection)
+		return -1;
+
+	uhub_assert(msg->cache && *msg->cache);
+
+	if (ioq_send_is_empty(link->send_queue) /*&& !user_flag_get(user, flag_pipeline)*/)
+	{
+		/* Perform oportunistic write */
+		ioq_send_add(link->send_queue, msg);
+		link_handle_write(link);
+	}
+	else
+	{
+// 		if (check_send_queue(hub, user, msg) >= 0)
+// 		{
+			ioq_send_add(link->send_queue, msg);
+// 			if (!user_flag_get(user, flag_pipeline))
+			link_net_io_want_write(link);
+	}
+	return 0;
+}
+
+static int link_send_support(struct hub_link* link)
+{
+	int ret;
+	struct adc_message* msg = adc_msg_construct(ADC_CMD_LSUP, 6 + strlen(ADC_PROTO_LINK_SUPPORT));
+	adc_msg_add_argument(msg, ADC_PROTO_LINK_SUPPORT);
+	ret = link_send_message(link, msg);
+	adc_msg_free(msg);
+	return ret;
+}
+
+static int link_send_welcome(struct hub_link* link)
+{
+	int ret;
+	struct adc_message* info = adc_msg_construct(ADC_CMD_LINF, 128);
+
+	if (!info)
+		return -1;
+
+	adc_msg_add_named_argument(info, ADC_INF_FLAG_CLIENT_TYPE, ADC_CLIENT_TYPE_HUB);
+	adc_msg_add_named_argument_string(info, ADC_INF_FLAG_USER_AGENT, PRODUCT_STRING);
+	adc_msg_add_named_argument_string(info, ADC_INF_FLAG_NICK, link->hub->config->hub_name);
+	adc_msg_add_named_argument_string(info, ADC_INF_FLAG_DESCRIPTION, link->hub->config->hub_description);
+
+	ret = link_send_message(link, info);
+
+	link->state = state_normal;
+}
+
+static int link_send_auth_response(struct hub_link* link, const char* challenge)
+{
+	int ret;
+	struct adc_message* msg = adc_msg_construct(ADC_CMD_LPAS, 128);
+
+	// FIXME: Solve challenge.
+
+	ret = link_send_message(link, msg);
+	adc_msg_free(msg);
+	return ret;
+}
+
+static int link_send_auth_request(struct hub_link* link)
+{
+	int ret;
+	struct adc_message* msg = adc_msg_construct(ADC_CMD_LGPA, 128);
+
+	// FIXME: Create challenge.
+	char buf[64];
+	uint64_t tiger_res[3];
+	static char tiger_buf[MAX_CID_LEN+1];
+
+	LOG_DEBUG("link_send_auth_request");
+
+	// FIXME: Generate a better nonce scheme.
+	snprintf(buf, 64, "%p%d", link, (int) net_con_get_sd(link->connection));
+
+	tiger((uint64_t*) buf, strlen(buf), (uint64_t*) tiger_res);
+	base32_encode((unsigned char*) tiger_res, TIGERSIZE, tiger_buf);
+	tiger_buf[MAX_CID_LEN] = 0;
+
+	// Add nonce to message
+	adc_msg_add_argument(msg, (const char*) tiger_buf);
+	ret = link_send_message(link, msg);
+	adc_msg_free(msg);
+	return ret;
+}
+
+static int link_handle_support(struct hub_link* link, struct adc_message* msg)
+{
+	int ret = 0;
+
+	LOG_DEBUG("link_handle_support");
+
+	if (link->mode == link_mode_server)
+	{
+		if (link->state == state_protocol)
+		{
+			ret = link_send_support(link);
+			if (ret == 0)
+				ret = link_send_auth_request(link);
+			link->state = state_verify;
+		}
+	}
+	return ret;
+}
+
+static int link_handle_auth_request(struct hub_link* link, struct adc_message* msg)
+{
+	char* challenge;
+	int ret = -1;
+
+	LOG_DEBUG("link_handle_auth_request");
+
+	if (link->state == state_verify)
+		return -1;
+
+	if (link->mode == link_mode_client)
+	{
+		challenge = adc_msg_get_argument(msg, 0);
+		ret = link_send_auth_response(link, challenge);
+		hub_free(challenge);
+	}
+	return ret;
+}
+
+static int link_handle_auth_response(struct hub_link* link, struct adc_message* msg)
+{
+
+	LOG_DEBUG("link_handle_auth_response. link_state=%d", (int) link->state);
+
+	if (link->state != state_verify)
+		return -1;
+
+	LOG_DEBUG("State is not verify!");
+
+	if (link->mode == link_mode_server)
+	{
+		// Check authentication data
+		// FIXME: Can involve plug-ins at this point.
+		return link_send_welcome(link);
+	}
+	else
+	{
+		LOG_DEBUG("Ignoring auth response - We're client mode!");
+	}
+
+	return -1;
+}
+
+static int link_handle_link_info(struct hub_link* link, struct adc_message* msg)
+{
+	LOG_DEBUG("link_handle_link_info");
+	return 0;
+}
+
+static int link_handle_status(struct hub_link* link, struct adc_message* msg)
+{
+	LOG_DEBUG("link_handle_status");
+	return -1;
+}
+
+static int link_handle_message(struct hub_link* link, const char* message, size_t length)
+{
+	int ret = 0;
+	struct adc_message* cmd = 0;
+
+	LOG_INFO("link_handle_message(): %s (%d)", message, (int) length);
+
+	// FIXME: is this needed?
+	if (link->state == state_cleanup || link->state == state_disconnected)
+		return -1;
+
+	cmd = adc_msg_parse(message, length);
+	if (!cmd)
+	{
+		LOG_DEBUG("Unable to parse hub-link message");
+		return -1;
+	}
+
+	// if (
+
+	switch (cmd->cmd)
+	{
+		case ADC_CMD_LSUP:
+			ret = link_handle_support(link, cmd);
+			break;
+
+		case ADC_CMD_LPAS:
+			ret = link_handle_auth_response(link, cmd);
+			break;
+
+		case ADC_CMD_LGPA:
+			ret = link_handle_auth_request(link, cmd);
+			break;
+
+		case ADC_CMD_LINF:
+			ret = link_handle_link_info(link, cmd);
+			break;
+
+		case ADC_CMD_LSTA:
+			ret = link_handle_status(link, cmd);
+			break;
+	}
+
+	adc_msg_free(cmd);
+	return ret;
+}
+
+
+static int link_read_message(struct hub_link* link)
+{
+	char* lastPos = 0;
+	char* pos = 0;
+	char* start = link->recv_queue->buf;
+	size_t remaining = link->recv_queue->size;
+
+	while ((pos = memchr(start, '\n', remaining)))
+	{
+		lastPos = pos+1;
+		pos[0] = '\0';
+
+		if (link->flags & 1)
+		{
+			 /* FIXME Unset maxbuf flag */
+			link->flags = 0;
+		}
+		else
+		{
+			if (link_handle_message(link, start, (pos - start)) == -1)
+			{
+				return -1;
+			}
+		}
+
+		pos[0] = '\n'; /* FIXME: not needed */
+		pos ++;
+		remaining -= (pos - start);
+		start = pos;
+	}
+
+	ioq_recv_consume(link->recv_queue, (start - link->recv_queue->buf));
+	return 0;
+}
+
+int link_handle_read(struct hub_link* link)
+{
+	int ret = 0;
+	while (1)
+	{
+		switch (ioq_recv_read(link->recv_queue, link->connection))
+		{
+			case ioq_recv_ok:
+				if (link_read_message(link) < 0)
+				{
+					// FIXME: propagate protocol error?
+					return -1;
+				}
+				// Parse messages then call again
+				break;
+				
+			case ioq_recv_later:
+				return 0;
+
+			case ioq_recv_full:
+				link->flags = 1; // FIXME: MAXBUF
+				ioq_recv_set(link->recv_queue, 0, 0);
+				break;
+
+			case ioq_recv_error:
+				return -1; // FIXME: it would be good to signal type of socket error
+		}
+	}
+
+	return 0;
+}
+
+
+#endif /* LINK_SUPPORT */

src/core/link.h

@@ -0,0 +1,70 @@
+/*
+ * uhub - A tiny ADC p2p connection hub
+ * Copyright (C) 2007-2013, Jan Vidar Krey
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#ifndef HAVE_UHUB_LINK_H
+#define HAVE_UHUB_LINK_H
+
+#ifdef LINK_SUPPORT
+
+struct hub_link
+{
+	char name[MAX_NICK_LEN+1];         /** The name of the linked hub */
+	char user_agent[MAX_UA_LEN+1];     /** The user agent of the linked hub */
+	char address[256];                 /** The official address of the linked hub */
+	enum link_mode { link_mode_client, link_mode_server } mode;
+	enum user_state state;
+	struct ioq_send* send_queue;
+	struct ioq_recv* recv_queue;
+	struct net_connection* connection; /** Connection data */
+	struct net_connect_handle* connect_job; /** Only used when establishing a connection in client mode */
+	struct hub_info* hub;
+	int flags;
+};
+
+/**
+ * Create a link from an accepted connection (act as a link server).
+ */
+extern struct hub_link* link_create(struct hub_info* hub, struct net_connection* con, struct ip_addr_encap* addr);
+
+/**
+ * Connect this hub to an upstream server (act as a link client).
+ */
+extern struct hub_link* link_connect(struct hub_info* hub, const char* address, uint16_t port);
+extern struct hub_link* link_connect_uri(struct hub_info* hub, const char* address);
+
+/**
+ * Disconnect a link connection.
+ */
+extern void link_disconnect(struct hub_link*);
+
+/**
+ * Read from link connection and process messages.
+ * @return 0 on success, and a negative value otherwise
+ */
+extern int link_handle_read(struct hub_link* link);
+
+/**
+ * Write queued messages to the link.
+ * @return 0 on success, and a negative value otherwise.
+ */
+extern int link_handle_write(struct hub_link* link);
+
+#endif // LINK_SUPPORT
+
+#endif /* HAVE_UHUB_LINK_H */

src/core/main.c

@@ -125,6 +125,10 @@ int main_loop()
 			hub_set_log_verbosity(arg_verbose);
 		}
 
+#ifdef DEBUG
+		LOG_INFO("Debug messages enabled");
+#endif
+	
 		if (read_config(arg_config, &configuration, !arg_have_config) == -1)
 			return -1;
 

src/core/plugincallback.c

@@ -104,7 +104,7 @@ static int cbfunc_command_add(struct plugin_handle* plugin, struct plugin_comman
 	cmdh->internal_handle = command;
 	list_append(data->commands, cmdh);
 	command_add(plugin_get_hub(plugin)->commands, command, (void*) plugin);
-	LOG_DEBUG("*** Add plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
+	printf("*** Add plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
 	return 0;
 }
 
@@ -113,7 +113,7 @@ static int cbfunc_command_del(struct plugin_handle* plugin, struct plugin_comman
 	struct plugin_callback_data* data = get_callback_data(plugin);
 	struct command_handle* command = (struct command_handle*) cmdh->internal_handle;
 
-	LOG_DEBUG("*** Del plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
+	printf("*** Del plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
 	list_remove(data->commands, cmdh);
 	command_del(plugin_get_hub(plugin)->commands, command);
 	hub_free(command);

src/core/probe.c

@@ -71,6 +71,17 @@ static void probe_net_event(struct net_connection* con, int events, void *arg)
 				probe_destroy(probe);
 				return;
 			}
+#ifdef LINK_SUPPORT
+			else if (probe->hub->config->hub_link_enabled && memcmp(probe_recvbuf, "LSUP", 4) == 0)
+			{
+				if (link_create(probe->hub, probe->connection, &probe->addr))
+				{
+					probe->connection = 0;
+				}
+				probe_destroy(probe);
+				return;
+			}
+#endif /* LINK_SUPPORT */
 #ifdef SSL_SUPPORT
 			else if (bytes >= 11 &&
 				probe_recvbuf[0] == 22 && 

src/core/route.c

@@ -99,24 +99,12 @@ int route_to_user(struct hub_info* hub, struct hub_user* user, struct adc_messag
 {
 #ifdef DEBUG_SENDQ
 	char* data = strndup(msg->cache, msg->length-1);
-	LOG_PROTO("send %s: \"%s\"", sid_to_string(user->id.sid), data);
+	LOG_PROTO("[user] send %s: \"%s\"", sid_to_string(user->id.sid), data);
 	free(data);
 #endif
 
 	if (!user->connection)
-	{
-		switch (user->type)
-		{
-			case user_type_client:
-				return 0; // No connection - we're about to drop this user.
-			case user_type_bot:
-			{
-				bot_recv_msg handler = (bot_recv_msg) user->ptr;
-				handler(user, msg);
 		return 0;
-			}
-		}
-	}
 
 	uhub_assert(msg->cache && *msg->cache);
 
@@ -161,19 +149,6 @@ int route_to_all(struct hub_info* hub, struct adc_message* command) /* iterate u
 	return 0;
 }
 
-int route_to_operators(struct hub_info* hub, struct adc_message* command) /* iterate users */
-{
-	struct hub_user* user = (struct hub_user*) list_get_first(hub->users->list);
-	while (user)
-	{
-		if (user_flag_get(user, flag_opnotify))
-			route_to_user(hub, user, command);
-		user = (struct hub_user*) list_get_next(hub->users->list);
-	}
-	return 0;
-}
-
-
 int route_to_subscribers(struct hub_info* hub, struct adc_message* command) /* iterate users */
 {
 	int do_send;

src/core/route.h

@@ -40,14 +40,6 @@ extern int route_to_user(struct hub_info* hub, struct hub_user*, struct adc_mess
  */
 extern int route_to_all(struct hub_info* hub, struct adc_message* command);
 
-/**
- * Send a message to all operators.
- * Technically it sends to all users with the flag_opnotify set
- * which is intended to be set for only operators - allowing operators to opt-out
- * of these messages.
- */
-extern int route_to_operators(struct hub_info* hub, struct adc_message* command);
-
 /**
  * Broadcast message to all users subscribing to the type of message.
  */

src/core/user.c

@@ -42,6 +42,7 @@ struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, s
 	LOG_TRACE("user_create(), hub=%p, con[sd=%d]", hub, net_con_get_sd(con));
 
 	user = (struct hub_user*) hub_malloc_zero(sizeof(struct hub_user));
+
 	if (user == NULL)
 		return NULL; /* OOM */
 
@@ -53,7 +54,6 @@ struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, s
 
 	memcpy(&user->id.addr, addr, sizeof(struct ip_addr_encap));
 	user_set_state(user, state_protocol);
-	user->type = user_type_client;
 
 	flood_control_reset(&user->flood_chat);
 	flood_control_reset(&user->flood_connect);
@@ -65,54 +65,6 @@ struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, s
 	return user;
 }
 
-/// generate a semi-stable CID: base32_encode(tiger({hub name} + {nick}))
-static const char* generate_bot_cid(struct hub_info* hub, const char* nick)
-{
-	static char result[MAX_CID_LEN+1];
-	char buf[(MAX_NICK_LEN*2)+1];
-	uint64_t tiger_res[3];
-
-	memset(buf, 0, sizeof(buf));
-	snprintf(buf, sizeof(buf), "%s%s", hub->config->hub_name, nick);
-	tiger((uint64_t*) buf, sizeof(buf), (uint64_t*) tiger_res);
-	base32_encode((unsigned char*) tiger_res, TIGERSIZE, result);
-	result[MAX_CID_LEN] = 0;
-	return result;
-}
-
-struct hub_user* user_create_bot(struct hub_info* hub, const char* nick, const char* description, bot_recv_msg msg_handler)
-{
-	struct hub_user* user = NULL;
-	LOG_TRACE("user_create_bot(), hub=%p, nick=\"%s\"", hub, nick);
-
-	user = (struct hub_user*) hub_malloc_zero(sizeof(struct hub_user));
-	if (user == NULL)
-		return NULL; /* OOM */
-
-	strcpy(user->id.nick, nick);
-	uman_get_free_sid(hub->users, user);
-
-	user_set_state(user, state_normal);
-	user->type = user_type_bot;
-	user->credentials = auth_cred_bot;
-
-	// The message handler
-	user->ptr = (void*) msg_handler;
-
-	user->info = adc_msg_construct(ADC_CMD_BINF, 15);
-	if (user->info)
-	{
-		adc_msg_add_argument(user->info, sid_to_string(user->id.sid));
-		adc_msg_add_named_argument(user->info, ADC_INF_FLAG_CLIENT_TYPE, ADC_CLIENT_TYPE_BOT);
-		adc_msg_add_named_argument_string(user->info, ADC_INF_FLAG_USER_AGENT, PRODUCT_STRING);
-		adc_msg_add_named_argument_string(user->info, ADC_INF_FLAG_NICK, nick);
-		adc_msg_add_named_argument_string(user->info, ADC_INF_FLAG_DESCRIPTION, description);
-		adc_msg_add_named_argument(user->info, ADC_INF_FLAG_CLIENT_ID, generate_bot_cid(hub, nick));
-	}
-
-	user->hub = hub;
-	return user;
-}
 
 void user_destroy(struct hub_user* user)
 {

src/core/user.h

@@ -47,7 +47,6 @@ enum user_flags
 	feature_link    = 0x00000100, /** LINK: Hub link (not supported) */
 	feature_adcs    = 0x00000200, /** ADCS: ADC over TLS/SSL */
 	feature_bas0    = 0x00000400, /** BAS0: Obsolete pre-ADC/1.0 protocol version */
-	flag_opnotify   = 0x00200000, /** User should receive operation notifications. NOTE: Only operators should have this flag! */
 	flag_flood      = 0x00400000, /** User has been notified about flooding. */
 	flag_muted      = 0x00800000, /** User is muted (cannot chat) */
 	flag_ignore     = 0x01000000, /** Ignore further reads */
@@ -105,30 +104,21 @@ struct hub_user_limits
 	size_t              hub_count_total;       /** The number of hubs connected to in total */
 };
 
-enum user_type
-{
-	user_type_client,   /** A user connected normally as an ADC client */
-	user_type_bot,      /** Not really a user, but a bot inside the hub */
-};
-
-typedef void (*bot_recv_msg)(struct hub_user*, struct adc_message* msg);
-
 struct hub_user
 {
 	struct hub_user_info    id;                 /** Contains nick name and CID */
 	enum auth_credentials   credentials;        /** see enum user_credentials */
 	enum user_state         state;              /** see enum user_state */
-	enum user_type          type;
 	uint32_t                flags;              /** see enum user_flags */
 	struct linked_list*     feature_cast;       /** Features supported by feature cast */
 	struct adc_message*     info;               /** ADC 'INF' message (broadcasted to everyone joining the hub) */
 	struct hub_info*        hub;                /** The hub instance this user belong to */
-	void* ptr;
 	struct ioq_recv*        recv_queue;
 	struct ioq_send*        send_queue;
 	struct net_connection*  connection;         /** Connection data */
 	struct hub_user_limits  limits;             /** Data used for limitation */
 	enum user_quit_reason   quit_reason;        /** Quit reason (see user_quit_reason) */
+
 	struct flood_control   flood_chat;
 	struct flood_control   flood_connect;
 	struct flood_control   flood_search;
@@ -148,7 +138,6 @@ struct hub_user
  * @return User object or NULL if not enough memory is available.
  */
 extern struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, struct ip_addr_encap* addr);
-extern struct hub_user* user_create_bot(struct hub_info* hub, const char* nick, const char* description, bot_recv_msg msg_handler);
 
 /**
  * Delete a user.

src/network/timeout.c

@@ -1,6 +1,6 @@
 /*
  * uhub - A tiny ADC p2p connection hub
- * Copyright (C) 2007-2013, Jan Vidar Krey
+ * Copyright (C) 2007-2010, Jan Vidar Krey
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -57,31 +57,15 @@ size_t timeout_queue_process(struct timeout_queue* t, time_t now)
 	size_t pos = (size_t) t->last;
 	size_t events = 0;
 	struct timeout_evt* evt = 0;
-
-	uhub_assert(t->last <= now);
 	t->last = now;
-
-	// We can optimize in case we need to wrap around
-	// the buffer, so we only do it once
-	if (MAX(pos, now) - MIN(pos, now) > t->max)
-	{
-		// FIXME: Double check this calculation
-		pos = (now - t->max);
-	}
-
 	for (; pos <= now; pos++)
 	{
-		evt = t->events[pos % t->max];
-		while (evt)
-		{
-			if (evt->timestamp < pos)
+		while ((evt = t->events[pos % t->max]))
 		{
 			timeout_queue_remove(t, evt);
 			evt->callback(evt);
 			events++;
 		}
-			evt = evt->next;
-		}
 	}
 	return events;
 }
@@ -101,22 +85,24 @@ size_t timeout_queue_get_next_timeout(struct timeout_queue* t, time_t now)
 
 void timeout_queue_insert(struct timeout_queue* t, struct timeout_evt* evt, size_t seconds)
 {
-	struct timeout_evt* it, *first;
+	struct timeout_evt* first;
 	size_t pos = ((t->last + seconds) % t->max);
 	evt->timestamp = t->last + seconds;
 	evt->next = 0;
 	
 	first = t->events[pos];
+	
 	if (first)
 	{
-		for (it = first; it->next; it = it->next) { }
-		it->next = evt;
-		evt->prev = it;
+		uhub_assert(first->timestamp == evt->timestamp);
+		first->prev->next = evt;
+		evt->prev = first->prev;
+		first->prev = evt;
 	}
 	else
 	{
 		t->events[pos] = evt;
-		evt->prev = evt; // point to self.
+		evt->prev = evt;
 	}
 	evt->next = 0;
 }

src/network/timeout.h

@@ -1,6 +1,6 @@
 /*
  * uhub - A tiny ADC p2p connection hub
- * Copyright (C) 2007-2013, Jan Vidar Krey
+ * Copyright (C) 2007-2010, Jan Vidar Krey
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by

src/uhub.h

@@ -94,6 +94,7 @@ extern "C" {
 #include "core/commands.h"
 #include "core/inf.h"
 #include "core/hubevent.h"
+#include "core/link.h"
 #include "core/plugincallback.h"
 #include "core/plugininvoke.h"
 #include "core/pluginloader.h"

tools/uhub-adc-redirector.py

@@ -1,43 +0,0 @@
-#!/usr/bin/env python
-"""
-A simple ADC redirector service.
-"""
-
-import SocketServer
-
-# The target hub we want to redirect clients to
-redirect_uri = "adcs://adcs.uhub.org:1511"
-
-# A message to be sent to users while they are being redirected.
-message = "This hub has been permanently moved."
-
-# The chat name of the message.
-bot_name = "Redirector"
-
-# The local address and port to bind the redirector to.
-bind_addr = "0.0.0.0"
-bind_port = 1411
-
-class AdcRedirector(SocketServer.BaseRequestHandler):
-
-	def escape(self, str):
-		modified = str.replace("\\", "\\\\").replace(" ", "\\s").replace("\n", "\\n")
-		return modified;
-
-	def handle(self):
-		supports = False;
-		while True:
-			data = self.request.recv(1024)
-			if (data.startswith("HSUP") and not supports):
-				self.request.sendall("ISUP ADBASE ADTIGR\nISID AAAX\nIINF CT32 NI%(botname)s VEuhub-adc-redirector/0.1\n" % { "address": redirect_uri, "botname": self.escape(bot_name), "message": self.escape(message) })
-				supports = True
-			elif (data.startswith("BINF") and supports):
-				self.request.sendall("IMSG %(message)s\nIQUI AAAX RD%(address)s\n" % {"message": self.escape(message), "address": redirect_uri })
-				break
-			else:
-				break
-
-if __name__ == "__main__":
-	server = SocketServer.TCPServer((bind_addr, bind_port), AdcRedirector)
-	server.allow_reuse_address = True
-	server.serve_forever()

tools/uhub-nmdc-redirector.py

@@ -1,30 +0,0 @@
-#!/usr/bin/env python
-"""
-A simple NMDC to ADC redirector service.
-"""
-
-import SocketServer
-
-# The target hub we want to redirect clients to
-redirect_uri = "adcs://adcs.uhub.org:1511"
-
-# A message to be sent to users while they are being redirected.
-message = "This hub has been permanently moved."
-
-# The chat name of the message.
-bot_name = "Redirector"
-
-# The local address and port to bind the redirector to.
-bind_addr = "0.0.0.0"
-bind_port = 1411
-
-class NmdcRedirector(SocketServer.BaseRequestHandler):
-
-	def setup(self):
-		self.request.sendall("<%(botname)s> %(message)s|$ForceMove %(address)s|" % { "address": redirect_uri, "botname": bot_name, "message": message })
-		return False
-
-if __name__ == "__main__":
-	server = SocketServer.TCPServer((bind_addr, bind_port), NmdcRedirector)
-	server.allow_reuse_address = True
-	server.serve_forever()