[thirdparty/strongswan.git] / src / libcharon / plugins / vici / vici_dispatcher.c

/*
 * Copyright (C) 2014 Martin Willi
 * Copyright (C) 2014 revosec AG
 *
 * 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 2 of the License, or (at your
 * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
 *
 * 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.
 */

#include "vici_dispatcher.h"
#include "vici_socket.h"

#include <bio/bio_reader.h>
#include <bio/bio_writer.h>
#include <threading/mutex.h>
#include <threading/condvar.h>
#include <threading/thread.h>
#include <collections/array.h>
#include <collections/hashtable.h>

typedef struct private_vici_dispatcher_t private_vici_dispatcher_t;

/**
 * Private data of an vici_dispatcher_t object.
 */
struct private_vici_dispatcher_t {

	/**
	 * Public vici_dispatcher_t interface.
	 */
	vici_dispatcher_t public;

	/**
	 * Socket to send/receive messages
	 */
	vici_socket_t *socket;

	/**
	 * List of registered commands (char* => command_t*)
	 */
	hashtable_t *cmds;

	/**
	 * List of known events, and registered clients (char* => event_t*)
	 */
	hashtable_t *events;

	/**
	 * Mutex to lock hashtables
	 */
	mutex_t *mutex;

	/**
	 * Condvar to signal command termination
	 */
	condvar_t *cond;
};

/**
 * Registered command
 */
typedef struct {
	/** command name */
	char *name;
	/** callback for command */
	vici_command_cb_t cb;
	/** user data to pass to callback */
	void *user;
	/** command currently in use? */
	u_int uses;
} command_t;

/**
 * Registered event
 */
typedef struct {
	/** event name */
	char *name;
	/** registered clients, as u_int */
	array_t *clients;
	/** event currently in use? */
	u_int uses;
} event_t;

/**
 * Send a operation code, optionally with name and message
 */
static void send_op(private_vici_dispatcher_t *this, u_int id,
					vici_operation_t op, char *name, vici_message_t *message)
{
	bio_writer_t *writer;
	u_int len;

	len = sizeof(u_int8_t);
	if (name)
	{
		len += sizeof(u_int8_t) + strlen(name);
	}
	if (message)
	{
		len += message->get_encoding(message).len;
	}
	writer = bio_writer_create(len);
	writer->write_uint8(writer, op);
	if (name)
	{
		writer->write_data8(writer, chunk_from_str(name));
	}
	if (message)
	{
		writer->write_data(writer, message->get_encoding(message));
	}
	this->socket->send(this->socket, id, writer->extract_buf(writer));
	writer->destroy(writer);
}

/**
 * Register client for event
 */
static void register_event(private_vici_dispatcher_t *this, char *name,
						   u_int id)
{
	event_t *event;

	this->mutex->lock(this->mutex);
	event = this->events->get(this->events, name);
	if (event)
	{
		array_insert(event->clients, ARRAY_TAIL, &id);
	}
	this->mutex->unlock(this->mutex);

	if (event)
	{
		DBG2(DBG_CFG, "vici client %u registered for: %s", id, name);
		send_op(this, id, VICI_EVENT_CONFIRM, NULL, NULL);
	}
	else
	{
		DBG1(DBG_CFG, "vici client %u invalid registration: %s", id, name);
		send_op(this, id, VICI_EVENT_UNKNOWN, NULL, NULL);
	}
}

/**
 * Unregister client for event
 */
static void unregister_event(private_vici_dispatcher_t *this, char *name,
							 u_int id)
{
	enumerator_t *enumerator;
	event_t *event;
	u_int *current;
	bool found = FALSE;

	this->mutex->lock(this->mutex);
	event = this->events->get(this->events, name);
	if (event)
	{
		enumerator = array_create_enumerator(event->clients);
		while (enumerator->enumerate(enumerator, &current))
		{
			if (*current == id)
			{
				array_remove_at(event->clients, enumerator);
				found = TRUE;
				break;
			}
		}
		enumerator->destroy(enumerator);
	}
	this->mutex->unlock(this->mutex);

	DBG2(DBG_CFG, "vici client %u unregistered for: %s", id, name);

	if (found)
	{
		send_op(this, id, VICI_EVENT_CONFIRM, NULL, NULL);
	}
	else
	{
		send_op(this, id, VICI_EVENT_UNKNOWN, NULL, NULL);
	}
}

/**
 * Data to release on thread cancellation
 */
typedef struct {
	private_vici_dispatcher_t *this;
	command_t *cmd;
	vici_message_t *request;
} release_data_t;

/**
 * Release command after execution/cancellation
 */
CALLBACK(release_command, void,
	release_data_t *release)
{
	release->request->destroy(release->request);

	release->this->mutex->lock(release->this->mutex);
	if (--release->cmd->uses == 0)
	{
		release->this->cond->broadcast(release->this->cond);
	}
	release->this->mutex->unlock(release->this->mutex);

	free(release);
}

/**
 * Process a request message
 */
void process_request(private_vici_dispatcher_t *this, char *name, u_int id,
					 chunk_t data)
{
	vici_message_t *response = NULL;
	release_data_t *release;
	command_t *cmd;

	this->mutex->lock(this->mutex);
	cmd = this->cmds->get(this->cmds, name);
	if (cmd)
	{
		cmd->uses++;
	}
	this->mutex->unlock(this->mutex);

	if (cmd)
	{
		INIT(release,
			.this = this,
			.cmd = cmd,
		);

		DBG2(DBG_CFG, "vici client %u requests: %s", id, name);

		thread_cleanup_push(release_command, release);

		release->request = vici_message_create_from_data(data, FALSE);
		response = release->cmd->cb(cmd->user, cmd->name, id, release->request);

		thread_cleanup_pop(TRUE);

		if (response)
		{
			send_op(this, id, VICI_CMD_RESPONSE, NULL, response);
			response->destroy(response);
		}
	}
	else
	{
		DBG1(DBG_CFG, "vici client %u invalid request: %s", id, name);
		send_op(this, id, VICI_CMD_UNKNOWN, NULL, NULL);
	}
}

CALLBACK(inbound, void,
	private_vici_dispatcher_t *this, u_int id, chunk_t data)
{
	bio_reader_t *reader;
	chunk_t chunk;
	u_int8_t type;
	char name[257];

	reader = bio_reader_create(data);
	if (reader->read_uint8(reader, &type))
	{
		switch (type)
		{
			case VICI_EVENT_REGISTER:
				if (reader->read_data8(reader, &chunk) &&
					vici_stringify(chunk, name, sizeof(name)))
				{
					register_event(this, name, id);
				}
				else
				{
					DBG1(DBG_CFG, "invalid vici register message");
				}
				break;
			case VICI_EVENT_UNREGISTER:
				if (reader->read_data8(reader, &chunk) &&
					vici_stringify(chunk, name, sizeof(name)))
				{
					unregister_event(this, name, id);
				}
				else
				{
					DBG1(DBG_CFG, "invalid vici unregister message");
				}
				break;
			case VICI_CMD_REQUEST:
				if (reader->read_data8(reader, &chunk) &&
					vici_stringify(chunk, name, sizeof(name)))
				{
					thread_cleanup_push((void*)reader->destroy, reader);
					process_request(this, name, id, reader->peek(reader));
					thread_cleanup_pop(FALSE);
				}
				else
				{
					DBG1(DBG_CFG, "invalid vici request message");
				}
				break;
			case VICI_CMD_RESPONSE:
			case VICI_EVENT_CONFIRM:
			case VICI_EVENT_UNKNOWN:
			case VICI_EVENT:
			default:
				DBG1(DBG_CFG, "unsupported vici operation: %u", type);
				break;
		}
	}
	else
	{
		DBG1(DBG_CFG, "invalid vici message");
	}
	reader->destroy(reader);
}

CALLBACK(connect_, void,
	private_vici_dispatcher_t *this, u_int id)
{
	DBG2(DBG_CFG, "vici client %u connected", id);
}

CALLBACK(disconnect, void,
	private_vici_dispatcher_t *this, u_int id)
{
	enumerator_t *events, *ids;
	event_t *event;
	u_int *current;

	/* deregister all clients */
	this->mutex->lock(this->mutex);
	events = this->events->create_enumerator(this->events);
	while (events->enumerate(events, NULL, &event))
	{
		ids = array_create_enumerator(event->clients);
		while (ids->enumerate(ids, &current))
		{
			if (id == *current)
			{
				array_remove_at(event->clients, ids);
			}
		}
		ids->destroy(ids);
	}
	events->destroy(events);
	this->mutex->unlock(this->mutex);

	DBG2(DBG_CFG, "vici client %u disconnected", id);
}

METHOD(vici_dispatcher_t, manage_command, void,
	private_vici_dispatcher_t *this, char *name,
	vici_command_cb_t cb, void *user)
{
	command_t *cmd;

	this->mutex->lock(this->mutex);
	if (cb)
	{
		INIT(cmd,
			.name = strdup(name),
			.cb = cb,
			.user = user,
		);
		cmd = this->cmds->put(this->cmds, cmd->name, cmd);
	}
	else
	{
		cmd = this->cmds->remove(this->cmds, name);
	}
	if (cmd)
	{
		while (cmd->uses)
		{
			this->cond->wait(this->cond, this->mutex);
		}
		free(cmd->name);
		free(cmd);
	}
	this->mutex->unlock(this->mutex);
}

METHOD(vici_dispatcher_t, manage_event, void,
	private_vici_dispatcher_t *this, char *name, bool reg)
{
	event_t *event;

	this->mutex->lock(this->mutex);
	if (reg)
	{
		INIT(event,
			.name = strdup(name),
			.clients = array_create(sizeof(u_int), 0),
		);
		event = this->events->put(this->events, event->name, event);
	}
	else
	{
		event = this->events->remove(this->events, name);
	}
	if (event)
	{
		while (event->uses)
		{
			this->cond->wait(this->cond, this->mutex);
		}
		array_destroy(event->clients);
		free(event->name);
		free(event);
	}
	this->mutex->unlock(this->mutex);
}

METHOD(vici_dispatcher_t, raise_event, void,
	private_vici_dispatcher_t *this, char *name, u_int id,
	vici_message_t *message)
{
	enumerator_t *enumerator;
	event_t *event;
	u_int *current;

	this->mutex->lock(this->mutex);
	event = this->events->get(this->events, name);
	if (event)
	{
		event->uses++;
	}
	this->mutex->unlock(this->mutex);

	enumerator = array_create_enumerator(event->clients);
	while (enumerator->enumerate(enumerator, &current))
	{
		if (id == 0 || id == *current)
		{
			send_op(this, *current, VICI_EVENT, name, message);
		}
	}
	enumerator->destroy(enumerator);

	this->mutex->lock(this->mutex);
	if (--event->uses == 0)
	{
		this->cond->broadcast(this->cond);
	}
	this->mutex->unlock(this->mutex);

	message->destroy(message);
}

METHOD(vici_dispatcher_t, destroy, void,
	private_vici_dispatcher_t *this)
{
	DESTROY_IF(this->socket);
	this->mutex->destroy(this->mutex);
	this->cond->destroy(this->cond);
	this->cmds->destroy(this->cmds);
	this->events->destroy(this->events);
	free(this);
}

/**
 * See header
 */
vici_dispatcher_t *vici_dispatcher_create(char *uri)
{
	private_vici_dispatcher_t *this;

	INIT(this,
		.public = {
			.manage_command = _manage_command,
			.manage_event = _manage_event,
			.raise_event = _raise_event,
			.destroy = _destroy,
		},
		.cmds = hashtable_create(hashtable_hash_str, hashtable_equals_str, 1),
		.events = hashtable_create(hashtable_hash_str, hashtable_equals_str, 1),
		.mutex = mutex_create(MUTEX_TYPE_DEFAULT),
		.cond = condvar_create(CONDVAR_TYPE_DEFAULT),
	);

	this->socket = vici_socket_create(uri, inbound, connect_, disconnect, this);
	if (!this->socket)
	{
		destroy(this);
		return NULL;
	}

	return &this->public;
}
Commit	Line	Data
8383d626 MW	1	/*
	2	* Copyright (C) 2014 Martin Willi
	3	* Copyright (C) 2014 revosec AG
	4	*
	5	* This program is free software; you can redistribute it and/or modify it
	6	* under the terms of the GNU General Public License as published by the
	7	* Free Software Foundation; either version 2 of the License, or (at your
	8	* option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>.
	9	*
	10	* This program is distributed in the hope that it will be useful, but
	11	* WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
	12	* or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
	13	* for more details.
	14	*/
	15
	16	#include "vici_dispatcher.h"
	17	#include "vici_socket.h"
	18
	19	#include <bio/bio_reader.h>
	20	#include <bio/bio_writer.h>
	21	#include <threading/mutex.h>
	22	#include <threading/condvar.h>
ecc4b510	23	#include <threading/thread.h>
8383d626 MW	24	#include <collections/array.h>
	25	#include <collections/hashtable.h>
	26
	27	typedef struct private_vici_dispatcher_t private_vici_dispatcher_t;
	28
	29	/**
	30	* Private data of an vici_dispatcher_t object.
	31	*/
	32	struct private_vici_dispatcher_t {
	33
	34	/**
	35	* Public vici_dispatcher_t interface.
	36	*/
	37	vici_dispatcher_t public;
	38
	39	/**
	40	* Socket to send/receive messages
	41	*/
	42	vici_socket_t *socket;
	43
	44	/**
	45	* List of registered commands (char* => command_t*)
	46	*/
	47	hashtable_t *cmds;
	48
	49	/**
	50	* List of known events, and registered clients (char* => event_t*)
	51	*/
	52	hashtable_t *events;
	53
	54	/**
	55	* Mutex to lock hashtables
	56	*/
	57	mutex_t *mutex;
	58
	59	/**
	60	* Condvar to signal command termination
	61	*/
	62	condvar_t *cond;
	63	};
	64
	65	/**
	66	* Registered command
	67	*/
	68	typedef struct {
	69	/** command name */
	70	char *name;
	71	/** callback for command */
	72	vici_command_cb_t cb;
	73	/** user data to pass to callback */
	74	void *user;
	75	/** command currently in use? */
	76	u_int uses;
	77	} command_t;
	78
	79	/**
	80	* Registered event
	81	*/
	82	typedef struct {
	83	/** event name */
	84	char *name;
	85	/** registered clients, as u_int */
	86	array_t *clients;
	87	/** event currently in use? */
88	u_int uses;
89	} event_t;
90
91	/**
92	* Send a operation code, optionally with name and message
93	*/
94	static void send_op(private_vici_dispatcher_t *this, u_int id,
95	vici_operation_t op, char name, vici_message_t message)
96	{
97	bio_writer_t *writer;
98	u_int len;
99
100	len = sizeof(u_int8_t);
101	if (name)
102	{
103	len += sizeof(u_int8_t) + strlen(name);
104	}
105	if (message)
106	{
107	len += message->get_encoding(message).len;
108	}
109	writer = bio_writer_create(len);
110	writer->write_uint8(writer, op);
111	if (name)
112	{
113	writer->write_data8(writer, chunk_from_str(name));
114	}
115	if (message)
116	{
117	writer->write_data(writer, message->get_encoding(message));
118	}
119	this->socket->send(this->socket, id, writer->extract_buf(writer));
120	writer->destroy(writer);
121	}
122
123	/**
124	* Register client for event
125	*/
126	static void register_event(private_vici_dispatcher_t this, char name,
127	u_int id)
128	{
129	event_t *event;
130
131	this->mutex->lock(this->mutex);
132	event = this->events->get(this->events, name);
133	if (event)
134	{
135	array_insert(event->clients, ARRAY_TAIL, &id);
136	}
137	this->mutex->unlock(this->mutex);
138
139	if (event)
140	{
2676ffdb	141	DBG2(DBG_CFG, "vici client %u registered for: %s", id, name);
8383d626 MW	142	send_op(this, id, VICI_EVENT_CONFIRM, NULL, NULL);
	143	}
	144	else
	145	{
37aa250c	146	DBG1(DBG_CFG, "vici client %u invalid registration: %s", id, name);
8383d626 MW	147	send_op(this, id, VICI_EVENT_UNKNOWN, NULL, NULL);
	148	}
	149	}
	150
	151	/**
	152	* Unregister client for event
	153	*/
	154	static void unregister_event(private_vici_dispatcher_t this, char name,
	155	u_int id)
	156	{
	157	enumerator_t *enumerator;
	158	event_t *event;
	159	u_int *current;
	160	bool found = FALSE;
	161
	162	this->mutex->lock(this->mutex);
	163	event = this->events->get(this->events, name);
	164	if (event)
	165	{
	166	enumerator = array_create_enumerator(event->clients);
	167	while (enumerator->enumerate(enumerator, &current))
	168	{
	169	if (*current == id)
	170	{
	171	array_remove_at(event->clients, enumerator);
	172	found = TRUE;
	173	break;
	174	}
	175	}
	176	enumerator->destroy(enumerator);
	177	}
	178	this->mutex->unlock(this->mutex);
	179
2676ffdb	180	DBG2(DBG_CFG, "vici client %u unregistered for: %s", id, name);
37aa250c	181
8383d626 MW	182	if (found)
	183	{
	184	send_op(this, id, VICI_EVENT_CONFIRM, NULL, NULL);
	185	}
	186	else
	187	{
	188	send_op(this, id, VICI_EVENT_UNKNOWN, NULL, NULL);
	189	}
	190	}
	191
ecc4b510 MW	192	/**
	193	* Data to release on thread cancellation
	194	*/
	195	typedef struct {
	196	private_vici_dispatcher_t *this;
	197	command_t *cmd;
	198	vici_message_t *request;
	199	} release_data_t;
	200
	201	/**
	202	* Release command after execution/cancellation
	203	*/
	204	CALLBACK(release_command, void,
	205	release_data_t *release)
	206	{
	207	release->request->destroy(release->request);
	208
	209	release->this->mutex->lock(release->this->mutex);
	210	if (--release->cmd->uses == 0)
	211	{
	212	release->this->cond->broadcast(release->this->cond);
	213	}
	214	release->this->mutex->unlock(release->this->mutex);
	215
	216	free(release);
	217	}
	218
8383d626 MW	219	/**
	220	* Process a request message
	221	*/
	222	void process_request(private_vici_dispatcher_t this, char name, u_int id,
	223	chunk_t data)
	224	{
ecc4b510 MW	225	vici_message_t *response = NULL;
ecc4b510 MW	226	release_data_t *release;
9bfa397e	227	command_t *cmd;
8383d626 MW	228
8383d626 MW	229	this->mutex->lock(this->mutex);
9bfa397e MW	230	cmd = this->cmds->get(this->cmds, name);
9bfa397e MW	231	if (cmd)
8383d626	232	{
9bfa397e	233	cmd->uses++;
8383d626 MW	234	}
	235	this->mutex->unlock(this->mutex);
	236
9bfa397e	237	if (cmd)
8383d626	238	{
9bfa397e MW	239	INIT(release,
	240	.this = this,
	241	.cmd = cmd,
	242	);
	243
2676ffdb	244	DBG2(DBG_CFG, "vici client %u requests: %s", id, name);
37aa250c	245
ecc4b510 MW	246	thread_cleanup_push(release_command, release);
	247
	248	release->request = vici_message_create_from_data(data, FALSE);
9bfa397e	249	response = release->cmd->cb(cmd->user, cmd->name, id, release->request);
ecc4b510 MW	250
ecc4b510 MW	251	thread_cleanup_pop(TRUE);
8383d626	252
ecc4b510	253	if (response)
8383d626	254	{
ecc4b510 MW	255	send_op(this, id, VICI_CMD_RESPONSE, NULL, response);
ecc4b510 MW	256	response->destroy(response);
8383d626	257	}
8383d626 MW	258	}
	259	else
	260	{
37aa250c	261	DBG1(DBG_CFG, "vici client %u invalid request: %s", id, name);
8383d626 MW	262	send_op(this, id, VICI_CMD_UNKNOWN, NULL, NULL);
	263	}
	264	}
	265
	266	CALLBACK(inbound, void,
	267	private_vici_dispatcher_t *this, u_int id, chunk_t data)
	268	{
	269	bio_reader_t *reader;
	270	chunk_t chunk;
	271	u_int8_t type;
	272	char name[257];
	273
	274	reader = bio_reader_create(data);
	275	if (reader->read_uint8(reader, &type))
	276	{
	277	switch (type)
	278	{
	279	case VICI_EVENT_REGISTER:
	280	if (reader->read_data8(reader, &chunk) &&
	281	vici_stringify(chunk, name, sizeof(name)))
	282	{
	283	register_event(this, name, id);
	284	}
	285	else
	286	{
	287	DBG1(DBG_CFG, "invalid vici register message");
	288	}
	289	break;
	290	case VICI_EVENT_UNREGISTER:
	291	if (reader->read_data8(reader, &chunk) &&
	292	vici_stringify(chunk, name, sizeof(name)))
	293	{
	294	unregister_event(this, name, id);
	295	}
	296	else
	297	{
	298	DBG1(DBG_CFG, "invalid vici unregister message");
	299	}
	300	break;
	301	case VICI_CMD_REQUEST:
	302	if (reader->read_data8(reader, &chunk) &&
	303	vici_stringify(chunk, name, sizeof(name)))
	304	{
ecc4b510	305	thread_cleanup_push((void*)reader->destroy, reader);
8383d626	306	process_request(this, name, id, reader->peek(reader));
ecc4b510	307	thread_cleanup_pop(FALSE);
8383d626 MW	308	}
	309	else
	310	{
	311	DBG1(DBG_CFG, "invalid vici request message");
	312	}
	313	break;
	314	case VICI_CMD_RESPONSE:
	315	case VICI_EVENT_CONFIRM:
	316	case VICI_EVENT_UNKNOWN:
	317	case VICI_EVENT:
	318	default:
	319	DBG1(DBG_CFG, "unsupported vici operation: %u", type);
	320	break;
	321	}
	322	}
	323	else
	324	{
	325	DBG1(DBG_CFG, "invalid vici message");
	326	}
	327	reader->destroy(reader);
	328	}
	329
	330	CALLBACK(connect_, void,
	331	private_vici_dispatcher_t *this, u_int id)
	332	{
2676ffdb	333	DBG2(DBG_CFG, "vici client %u connected", id);
8383d626 MW	334	}
	335
	336	CALLBACK(disconnect, void,
	337	private_vici_dispatcher_t *this, u_int id)
	338	{
	339	enumerator_t events, ids;
	340	event_t *event;
	341	u_int *current;
	342
	343	/* deregister all clients */
	344	this->mutex->lock(this->mutex);
	345	events = this->events->create_enumerator(this->events);
	346	while (events->enumerate(events, NULL, &event))
	347	{
	348	ids = array_create_enumerator(event->clients);
	349	while (ids->enumerate(ids, &current))
	350	{
	351	if (id == *current)
	352	{
	353	array_remove_at(event->clients, ids);
	354	}
	355	}
	356	ids->destroy(ids);
	357	}
	358	events->destroy(events);
	359	this->mutex->unlock(this->mutex);
37aa250c	360
2676ffdb	361	DBG2(DBG_CFG, "vici client %u disconnected", id);
8383d626 MW	362	}
	363
	364	METHOD(vici_dispatcher_t, manage_command, void,
	365	private_vici_dispatcher_t this, char name,
	366	vici_command_cb_t cb, void *user)
	367	{
	368	command_t *cmd;
	369
	370	this->mutex->lock(this->mutex);
	371	if (cb)
	372	{
	373	INIT(cmd,
	374	.name = strdup(name),
	375	.cb = cb,
	376	.user = user,
	377	);
	378	cmd = this->cmds->put(this->cmds, cmd->name, cmd);
	379	}
	380	else
	381	{
	382	cmd = this->cmds->remove(this->cmds, name);
	383	}
	384	if (cmd)
	385	{
	386	while (cmd->uses)
	387	{
	388	this->cond->wait(this->cond, this->mutex);
	389	}
	390	free(cmd->name);
	391	free(cmd);
	392	}
	393	this->mutex->unlock(this->mutex);
	394	}
	395
	396	METHOD(vici_dispatcher_t, manage_event, void,
	397	private_vici_dispatcher_t this, char name, bool reg)
	398	{
	399	event_t *event;
	400
	401	this->mutex->lock(this->mutex);
	402	if (reg)
	403	{
	404	INIT(event,
	405	.name = strdup(name),
	406	.clients = array_create(sizeof(u_int), 0),
	407	);
	408	event = this->events->put(this->events, event->name, event);
	409	}
	410	else
	411	{
	412	event = this->events->remove(this->events, name);
	413	}
	414	if (event)
	415	{
	416	while (event->uses)
	417	{
	418	this->cond->wait(this->cond, this->mutex);
	419	}
	420	array_destroy(event->clients);
	421	free(event->name);
	422	free(event);
	423	}
	424	this->mutex->unlock(this->mutex);
	425	}
426
427	METHOD(vici_dispatcher_t, raise_event, void,
b40a12a9 MW	428	private_vici_dispatcher_t this, char name, u_int id,
b40a12a9 MW	429	vici_message_t *message)
8383d626 MW	430	{
	431	enumerator_t *enumerator;
	432	event_t *event;
b40a12a9	433	u_int *current;
8383d626 MW	434
	435	this->mutex->lock(this->mutex);
	436	event = this->events->get(this->events, name);
	437	if (event)
	438	{
	439	event->uses++;
	440	}
	441	this->mutex->unlock(this->mutex);
	442
	443	enumerator = array_create_enumerator(event->clients);
b40a12a9	444	while (enumerator->enumerate(enumerator, &current))
8383d626	445	{
b40a12a9 MW	446	if (id == 0 \|\| id == *current)
	447	{
	448	send_op(this, *current, VICI_EVENT, name, message);
	449	}
8383d626 MW	450	}
	451	enumerator->destroy(enumerator);
	452
	453	this->mutex->lock(this->mutex);
	454	if (--event->uses == 0)
	455	{
	456	this->cond->broadcast(this->cond);
	457	}
	458	this->mutex->unlock(this->mutex);
	459
	460	message->destroy(message);
	461	}
	462
	463	METHOD(vici_dispatcher_t, destroy, void,
	464	private_vici_dispatcher_t *this)
	465	{
	466	DESTROY_IF(this->socket);
	467	this->mutex->destroy(this->mutex);
	468	this->cond->destroy(this->cond);
	469	this->cmds->destroy(this->cmds);
	470	this->events->destroy(this->events);
	471	free(this);
	472	}
	473
	474	/**
	475	* See header
	476	*/
	477	vici_dispatcher_t vici_dispatcher_create(char uri)
	478	{
	479	private_vici_dispatcher_t *this;
	480
	481	INIT(this,
	482	.public = {
	483	.manage_command = _manage_command,
	484	.manage_event = _manage_event,
	485	.raise_event = _raise_event,
	486	.destroy = _destroy,
	487	},
	488	.cmds = hashtable_create(hashtable_hash_str, hashtable_equals_str, 1),
	489	.events = hashtable_create(hashtable_hash_str, hashtable_equals_str, 1),
	490	.mutex = mutex_create(MUTEX_TYPE_DEFAULT),
	491	.cond = condvar_create(CONDVAR_TYPE_DEFAULT),
	492	);
	493
	494	this->socket = vici_socket_create(uri, inbound, connect_, disconnect, this);
	495	if (!this->socket)
	496	{
	497	destroy(this);
	498	return NULL;
	499	}
	500
	501	return &this->public;
	502	}