Attempt to send messages via the Operations bot as private messages.

This is still pretty much work in progress.

autotest/test.c

@@ -761,6 +761,9 @@ 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");
@@ -951,7 +954,6 @@ 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 15
+#define MAX_EVENTS 100
 static struct timeout_queue* g_queue;
 static time_t g_now;
 static size_t g_max;
@@ -117,3 +117,28 @@ 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/message.c

@@ -787,6 +787,13 @@ int adc_msg_add_argument(struct adc_message* cmd, const char* string)
 	return 0;
 }
 
+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);
+	return ret;
+}
 
 char* adc_msg_get_argument(struct adc_message* cmd, int offset)
 {
@@ -866,8 +873,6 @@ int adc_msg_get_argument_index(struct adc_message* cmd, const char prefix[2])
 	return -1;
 }
 
-
-
 int adc_msg_escape_length(const char* str)
 {
 	int add = 0;

src/adc/message.h

@@ -171,6 +171,12 @@ 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.
+ * @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);
+
 /**
  * Append a named argument
  *
@@ -209,6 +215,13 @@ 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.
@@ -234,4 +247,5 @@ void adc_msg_unterminate(struct adc_message* cmd);
  */
 int adc_msg_get_arg_offset(struct adc_message* msg);
 
+
 #endif /* HAVE_UHUB_COMMAND_H */

src/core/commands.c

@@ -353,6 +353,7 @@ 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);
 	}
@@ -362,6 +363,7 @@ 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..."));
 }
 
@@ -384,6 +386,7 @@ 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/hub.c

@@ -653,6 +653,7 @@ 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;
 	}
@@ -671,6 +672,60 @@ 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;
@@ -751,6 +806,54 @@ 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;
@@ -834,6 +937,8 @@ 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);
 	return hub;
 }
 

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;
-	int status;
+	enum hub_state status;
 	char* recvbuf; /* Global receive buffer */
 	char* sendbuf; /* Global send buffer */
 
@@ -233,6 +233,20 @@ 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).
  */
@@ -350,6 +364,10 @@ 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,6 +31,10 @@ 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/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);
-	printf("*** Add plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
+	LOG_DEBUG("*** 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;
 
-	printf("*** Del plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
+	LOG_DEBUG("*** 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/route.c

@@ -104,7 +104,19 @@ int route_to_user(struct hub_info* hub, struct hub_user* user, struct adc_messag
 #endif
 
 	if (!user->connection)
-		return 0;
+	{
+		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);
 
@@ -149,6 +161,19 @@ 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,6 +40,14 @@ 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,7 +42,6 @@ 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 */
 
@@ -54,6 +53,7 @@ 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,6 +65,54 @@ 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,6 +47,7 @@ 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 */
@@ -104,21 +105,30 @@ 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;
@@ -138,6 +148,7 @@ 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-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
@@ -57,14 +57,30 @@ 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++)
 	{
-		while ((evt = t->events[pos % t->max]))
+		evt = t->events[pos % t->max];
+		while (evt)
 		{
-			timeout_queue_remove(t, evt);
-			evt->callback(evt);
-			events++;
+			if (evt->timestamp < pos)
+			{
+				timeout_queue_remove(t, evt);
+				evt->callback(evt);
+				events++;
+			}
+			evt = evt->next;
 		}
 	}
 	return events;
@@ -85,24 +101,22 @@ 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* first;
+	struct timeout_evt* it, *first;
 	size_t pos = ((t->last + seconds) % t->max);
 	evt->timestamp = t->last + seconds;
 	evt->next = 0;
 
 	first = t->events[pos];
-	
 	if (first)
 	{
-		uhub_assert(first->timestamp == evt->timestamp);
-		first->prev->next = evt;
-		evt->prev = first->prev;
-		first->prev = evt;
+		for (it = first; it->next; it = it->next) { }
+		it->next = evt;
+		evt->prev = it;
 	}
 	else
 	{
 		t->events[pos] = evt;
-		evt->prev = evt;
+		evt->prev = evt; // point to self.
 	}
 	evt->next = 0;
 }

src/network/timeout.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