lash-0.5.4/lashd/conn_mgr.c

/*
 *   LASH
 *
 *   Copyright (C) 2002 Robert Ham <rah@bash.sh>
 *
 *   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.
 *
 *   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, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#define _GNU_SOURCE /* addrinfo */

#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#include <string.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>

#include <lash/lash.h>
#include <lash/internal_headers.h>

#include "conn_mgr.h"
#include "conn.h"
#include "server.h"

void *conn_mgr_recv_run(void *data);
void *conn_mgr_send_run(void *data);

void
conn_mgr_free(conn_mgr_t * conn_mgr)
{
	conn_t *conn;
	lash_list_t *list;

	close(conn_mgr->listen_socket);

	for (list = conn_mgr->open_connections; list; list = lash_list_next(list)) {
		conn = (conn_t *) list->data;

		LASH_DEBUGARGS("closing open connection %ld ('%s')", conn->id,
					   conn_get_str_id(conn));
		conn_destroy(conn);
	}

	lash_list_free(conn_mgr->open_connections);
	conn_mgr->open_connections = NULL;
	for (list = conn_mgr->connections; list; list = list->next) {
		conn = (conn_t *) list->data;

		LASH_DEBUGARGS("closing connection %ld ('%s')", conn->id,
					   conn_get_str_id(conn));
		conn_destroy(conn);
	}
	lash_list_free(conn_mgr->connections);
	conn_mgr->connections = NULL;

	conn_mgr->listen_socket = 0;
	FD_ZERO(&conn_mgr->sockets);
	conn_mgr->fd_max = 0;
	conn_mgr->client_events = NULL;

	pthread_mutex_destroy(&conn_mgr->connections_lock);
	pthread_mutex_destroy(&conn_mgr->client_event_lock);
	pthread_cond_destroy(&conn_mgr->client_event_cond);
}

conn_mgr_t *
conn_mgr_new(server_t * server)
{
	conn_mgr_t *mgr;
	int err;

	mgr = lash_malloc0(sizeof(conn_mgr_t));
	pthread_mutex_init(&mgr->connections_lock, NULL);
	pthread_mutex_init(&mgr->client_event_lock, NULL);
	pthread_cond_init(&mgr->client_event_cond, NULL);
	FD_ZERO(&mgr->sockets);

	mgr->server = server;

	err = conn_mgr_start(mgr);
	if (err) {
		if (mgr)
			conn_mgr_destroy(mgr);
		return NULL;
	}

	return mgr;
}

void
conn_mgr_destroy(conn_mgr_t * conn_mgr)
{
	LASH_PRINT_DEBUG("stopping");
	conn_mgr_stop(conn_mgr);
	LASH_PRINT_DEBUG("stopped");

	conn_mgr_free(conn_mgr);
	free(conn_mgr);
}

static int
conn_mgr_set_socket_opts(int sock)
{
	int err;
	int reuse;

	/*
	 * reuse ports.  this is so we can bind again quickly after shutting down.
	 */
	reuse = 1;
	err = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
	if (err == -1) {
		fprintf(stderr, "%s: could not set SO_REUSEADDR on socket: %s\n",
				__FUNCTION__, strerror(errno));
	}

	/*
	 * make sure socket is closed on exec()
	 */
	err = fcntl(sock, F_SETFD, FD_CLOEXEC);
	if (err == -1) {
		fprintf(stderr,
				"%s: could not set close-on-exec on listen socket: %s\n",
				__FUNCTION__, strerror(errno));
		return -1;
	}

	return 0;
}

int
conn_mgr_bind_socket(int *sock, struct addrinfo *addr)
{
	int err;

	*sock = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
	if (*sock == -1) {
		LASH_DEBUGARGS
			("could not create socket with params domain=%d, type=%d, protocol=%d: %s",
			 addr->ai_family, addr->ai_socktype, addr->ai_protocol,
			 strerror(errno));
		return -1;
	}

	conn_mgr_set_socket_opts(*sock);

	err = bind(*sock, addr->ai_addr, addr->ai_addrlen);
	if (err) {
		LASH_DEBUGARGS("could not bind socket: %s", strerror(errno));

		err = close(*sock);
		if (err) {
			fprintf(stderr, "%s: error closing unconnected socket: %s",
					__FUNCTION__, strerror(errno));
		}

		return -1;
	}

	return 0;
}

int
conn_mgr_start(conn_mgr_t * conn_mgr)
{
	struct addrinfo hints;
	struct addrinfo *addrs;
	struct addrinfo *addr;
	int bound = 0;
	int err;

	memset(&hints, 0, sizeof(hints));
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_flags = AI_PASSIVE;

	err = getaddrinfo(NULL, LASH_PORT, &hints, &addrs);
	if (err) {
		fprintf(stderr, "%s: could not look up service name: %s\n",
				__FUNCTION__, gai_strerror(err));
		return -1;
	}

	/* try ipv6 first */
	if (!no_v6)
		for (addr = addrs; addr; addr = addr->ai_next) {
			if (addr->ai_family == AF_INET6) {
				err = conn_mgr_bind_socket(&conn_mgr->listen_socket, addr);
				if (err)
					continue;
				else {
					bound = 1;
					break;
				}
			}
		}

	if (!bound)
		for (addr = addrs; addr; addr = addr->ai_next) {
			if (no_v6 && addr->ai_family == AF_INET6)
				continue;

			err = conn_mgr_bind_socket(&conn_mgr->listen_socket, addr);
			if (err)
				continue;
			else {
				bound = 1;
				break;
			}
		}

	freeaddrinfo(addrs);

	if (!bound) {
		fprintf(stderr, "%s: could not create listen socket\n", __FUNCTION__);
		return -1;
	}

	/* start the listening */
	err = listen(conn_mgr->listen_socket, 20);
	if (err == -1) {
		fprintf(stderr, "%s: error setting socket to listen: %s\n",
				__FUNCTION__, strerror(errno));
		return -1;
	}
	printf("Listening for connections\n");

	FD_SET(conn_mgr->listen_socket, &conn_mgr->sockets);
	conn_mgr->fd_max = conn_mgr->listen_socket;

	/* start up the threads */
	LASH_PRINT_DEBUG("starting recv thread");
	err =
		pthread_create(&conn_mgr->recv_thread, NULL, conn_mgr_recv_run,
					   conn_mgr);
	if (err) {
		fprintf(stderr, "%s: could not start recv thread\n", __FUNCTION__);
		abort();
	}
	LASH_PRINT_DEBUG("recv thread started; starting send thread");
	pthread_create(&conn_mgr->send_thread, NULL, conn_mgr_send_run, conn_mgr);
	LASH_PRINT_DEBUG("send thread started");

	return 0;
}

void
conn_mgr_stop(conn_mgr_t * conn_mgr)
{
	conn_mgr->quit = 1;
	pthread_cond_signal(&conn_mgr->client_event_cond);
	if (conn_mgr->send_thread)
		pthread_join(conn_mgr->send_thread, NULL);
	if (conn_mgr->recv_thread)
		pthread_join(conn_mgr->recv_thread, NULL);
}

void
conn_mgr_send_client_event(conn_mgr_t * conn_mgr, server_event_t * event)
{
	pthread_mutex_lock(&conn_mgr->client_event_lock);
	conn_mgr->client_events =
		lash_list_append(conn_mgr->client_events, event);
	pthread_mutex_unlock(&conn_mgr->client_event_lock);
	pthread_cond_signal(&conn_mgr->client_event_cond);
}

void
conn_mgr_send_client_lash_event(conn_mgr_t * conn_mgr, unsigned long conn_id,
								lash_event_t * lash_event)
{
	server_event_t *server_event;

	if (!lash_event)
		return;

	server_event = server_event_new();
	server_event_set_conn_id(server_event, conn_id);
	server_event_set_lash_event(server_event, lash_event);

	conn_mgr_send_client_event(conn_mgr, server_event);
}

void
conn_mgr_send_client_lash_config(conn_mgr_t * conn_mgr, unsigned int conn_id,
								 lash_config_t * lash_config)
{
	server_event_t *server_event;

	if (!lash_config)
		return;

	server_event = server_event_new();
	server_event_set_conn_id(server_event, conn_id);
	server_event_set_lash_config(server_event, lash_config);

	conn_mgr_send_client_event(conn_mgr, server_event);
}

void
conn_mgr_send_client_lash_comm_event(conn_mgr_t * conn_mgr,
									 unsigned int conn_id,
									 lash_comm_event_t * event)
{
	server_event_t *server_event;

	if (!event)
		return;

	server_event = server_event_new();
	server_event_set_conn_id(server_event, conn_id);
	server_event_set_lash_comm_event(server_event, event);

	conn_mgr_send_client_event(conn_mgr, server_event);
}

void
conn_mgr_send_server_lash_event(conn_mgr_t * conn_mgr, unsigned long conn_id,
								lash_event_t * lash_event)
{
	server_event_t *server_event;

	if (!lash_event)
		return;

	server_event = server_event_new();
	server_event_set_conn_id(server_event, conn_id);
	server_event_set_lash_event(server_event, lash_event);

	server_send_event(conn_mgr->server, server_event);
}

void
conn_mgr_send_server_lash_config(conn_mgr_t * conn_mgr, unsigned int conn_id,
								 lash_config_t * lash_config)
{
	server_event_t *server_event;

	if (!lash_config)
		return;

	server_event = server_event_new();
	server_event_set_conn_id(server_event, conn_id);
	server_event_set_lash_config(server_event, lash_config);

	server_send_event(conn_mgr->server, server_event);
}

void
conn_mgr_send_server_lash_comm_event(conn_mgr_t * conn_mgr,
									 unsigned int conn_id,
									 lash_comm_event_t * event)
{
	server_event_t *server_event;

	if (!event)
		return;

	server_event = server_event_new();
	server_event_set_conn_id(server_event, conn_id);
	server_event_set_lash_comm_event(server_event, event);

	server_send_event(conn_mgr->server, server_event);
}

/***********************************
 *********** recv thread ***********
 ***********************************/

void
conn_mgr_accept_connection(conn_mgr_t * conn_mgr)
{
	conn_t *conn;

	/*struct sockaddr_storage ss;
	size_t ss_len;*/
	int err;

	conn = conn_new();

	/* accept the connection */
	conn->socket = accept(conn_mgr->listen_socket, NULL, NULL);
	/*(struct sockaddr *) &ss,
	 * &ss_len); */

	if (conn->socket == -1) {
		fprintf(stderr,
				"%s: error accepting socket connection from '%s': %s\n",
				__FUNCTION__, conn_get_str_id(conn), strerror(errno));
		conn_destroy(conn);
		return;
	}
	LASH_DEBUGARGS("accepted connection from '%s' with id %ld",
				   conn_get_str_id(conn), conn->id);

	/*
	 * make sure socket is closed on exec()
	 */
	err = fcntl(conn_mgr->listen_socket, F_SETFD, FD_CLOEXEC);
	if (err == -1) {
		fprintf(stderr,
				"%s: could not set close-on-exec on connection %ld ('%s'): %s\n",
				__FUNCTION__, conn->id, conn_get_str_id(conn),
				strerror(errno));
		exit(1);
	}

	/* socket stuff */
	FD_SET(conn->socket, &conn_mgr->sockets);
	if (conn->socket > conn_mgr->fd_max)
		conn_mgr->fd_max = conn->socket;

	/* add the connection to the open connection list */
	conn_mgr->open_connections =
		lash_list_append(conn_mgr->open_connections, conn);
}

void
conn_mgr_connect_client(conn_mgr_t * conn_mgr, conn_t * conn)
{
	lash_comm_event_t *event;
	server_event_t *server_event;
	int err;
	lash_connect_params_t *params;
	char id[37];

	/* read the Connect event */
	event = lash_comm_event_new();

	LASH_PRINT_DEBUG("recieving Connect");
	err = lash_comm_recv_event(conn->socket, event);
	if (err == -1 ||
		err == -2 ||
		err == -3 ||
		lash_comm_event_get_type(event) != LASH_Comm_Event_Connect ||
		!LASH_PROTOCOL_IS_VALID(event->event_data.connect->protocol_version))
	{
		/* there was an error */

		if (err == -1)
			fprintf(stderr,
					"%s: there was a recieve error from connection '%s': disconnecting client\n",
					__FUNCTION__, conn_get_str_id(conn));

		else if (err == -2)
			fprintf(stderr,
					"%s: connection '%s' closed before sending Connect event\n",
					__FUNCTION__, conn_get_str_id(conn));

		else if (err == -3)
			fprintf(stderr,
					"%s: connection '%s' is using the wrong low-level protocol version\n",
					__FUNCTION__, conn_get_str_id(conn));

		else if (lash_comm_event_get_type(event) != LASH_Comm_Event_Connect)
			fprintf(stderr,
					"%s: connection '%s' sent an event (of type %d) that wasn't Connect before it was connected; removing\n",
					__FUNCTION__, conn_get_str_id(conn),
					lash_comm_event_get_type(event));

		else if (!LASH_PROTOCOL_IS_VALID
				 (event->event_data.connect->protocol_version)) {
			lash_comm_event_t mismatch_event;

			fprintf(stderr,
					"%s: connection '%s' is using protocol %s; disconnecting\n",
					__FUNCTION__, conn_get_str_id(conn),
					lash_protocol_string(event->event_data.connect->
										 protocol_version));

			lash_comm_event_set_protocol_mismatch(&mismatch_event,
												  LASH_PROTOCOL_VERSION);
			lash_comm_send_event(conn->socket, &mismatch_event);
		}

		lash_comm_event_destroy(event);
		FD_CLR(conn->socket, &conn_mgr->sockets);
		conn_mgr->open_connections =
			lash_list_remove(conn_mgr->open_connections, conn);
		conn_destroy(conn);
		return;
	}

	params = event->event_data.connect;
	uuid_unparse(params->id, id);

	LASH_PRINT_DEBUG("connecting new client with:");
	LASH_DEBUGARGS("  connection id:  %ld", conn->id);
	LASH_DEBUGARGS("  source address: %s", conn_get_str_id(conn));
	LASH_DEBUGARGS("  protocol:       %s",
				   lash_protocol_string(event->event_data.connect->
										protocol_version));
	LASH_DEBUGARGS("  class:          '%s'", params->class);
	LASH_DEBUGARGS("  id:             '%s'", id);
	LASH_DEBUGARGS("  working dir:    '%s'", params->working_dir);
	LASH_DEBUGARGS("  requested proj: '%s'", params->project);
	LASH_DEBUGARGS("  flags:          %d", params->flags);
	LASH_DEBUGARGS("  %d args:", params->argc);
	for (err = 0; err < params->argc; err++) {
		LASH_DEBUGARGS("   arg %d: '%s'", err, params->argv[err]);
	}

	/* tell the server there's a new client connection */
	server_event = server_event_new();
	server_event_set_conn_id(server_event, conn->id);
	server_event_set_lash_connect_params(server_event,
										 event->event_data.connect);

	server_send_event(conn_mgr->server, server_event);

	event->event_data.connect = NULL;
	lash_comm_event_destroy(event);

	/* remove the connection from the open list */
	conn_mgr->open_connections =
		lash_list_remove(conn_mgr->open_connections, conn);

	/* add the connection to the connection list */
	pthread_mutex_lock(&conn_mgr->connections_lock);
	conn_mgr->connections = lash_list_append(conn_mgr->connections, conn);
	pthread_mutex_unlock(&conn_mgr->connections_lock);

	conn_set_recv_stamp(conn);
}

void
conn_mgr_disconnect_client(conn_mgr_t * conn_mgr, conn_t * conn,
						   int notify_server)
{
	unsigned long conn_id;

	/* clear up what we know of the connection */
	conn_mgr->connections = lash_list_remove(conn_mgr->connections, conn);

	conn_id = conn->id;
	FD_CLR(conn->socket, &conn_mgr->sockets);

	conn_unlock(conn);
	conn_destroy(conn);

	if (notify_server) {
		server_event_t *server_event;

		server_event = server_event_new();
		server_event_set_type(server_event, Client_Disconnect);
		server_event_set_conn_id(server_event, conn_id);

		server_send_event(conn_mgr->server, server_event);
	}

	LASH_DEBUGARGS("disconnected client on connection %ld", conn_id);
}

void
conn_mgr_deal_with_event(conn_mgr_t * conn_mgr, conn_t * conn,
						 lash_comm_event_t * event)
{
	switch (lash_comm_event_get_type(event)) {
	case LASH_Comm_Event_Event:
		conn_mgr_send_server_lash_event(conn_mgr, conn->id,
										lash_comm_event_take_event(event));
		lash_comm_event_destroy(event);
		break;
	case LASH_Comm_Event_Config:
		conn_mgr_send_server_lash_config(conn_mgr, conn->id,
										 lash_comm_event_take_config(event));
		lash_comm_event_destroy(event);
		break;
	case LASH_Comm_Event_Ping:
		lash_comm_event_set_type(event, LASH_Comm_Event_Pong);
		conn_mgr_send_client_lash_comm_event(conn_mgr, conn->id, event);
		break;
	case LASH_Comm_Event_Pong:
		lash_comm_event_destroy(event);
		break;
	default:
		break;
	}
}

void
conn_mgr_recv_event(conn_mgr_t * conn_mgr, conn_t * conn)
{
	int err;
	lash_comm_event_t *event;

	/* recieve the event */
	event = lash_comm_event_new();
	err = lash_comm_recv_event(conn->socket, event);

	if (err == -1) {
		fprintf(stderr, "%s: error recieving event from connection '%s'\n",
				__FUNCTION__, conn_get_str_id(conn));
		return;
	}

	/* connection disconnect */
	if (err == -2 || lash_comm_event_get_type(event) == LASH_Comm_Event_Close) {
		LASH_DEBUGARGS("%s: connection '%s' disconnected",
					   conn_get_str_id(conn));

		if (lash_comm_event_get_type(event) == LASH_Comm_Event_Close)
			lash_comm_event_destroy(event);

		conn_mgr_disconnect_client(conn_mgr, conn, 1);

		/* done what we might have done */
		pthread_mutex_unlock(&conn_mgr->connections_lock);
		return;
	}

	/* give up the lock here, as we're definately not messing with it now */
	pthread_mutex_unlock(&conn_mgr->connections_lock);

	/* this function will destroy the event */
	conn_mgr_deal_with_event(conn_mgr, conn, event);

	/* reset ping */
	conn_set_recv_stamp(conn);

	/* unlock the connection */
	conn_unlock(conn);
}

void
conn_mgr_read_socket(conn_mgr_t * conn_mgr, int socket)
{
	lash_list_t *list;
	conn_t *conn;

	/* check if it's a new connection */
	if (socket == conn_mgr->listen_socket) {
		conn_mgr_accept_connection(conn_mgr);
		return;
	}

	/* see if it's an open connection */
	list = conn_mgr->open_connections;
	while (list) {
		conn = (conn_t *) list->data;
		if (conn->socket == socket) {
			conn_mgr_connect_client(conn_mgr, conn);
			return;
		}

		list = list->next;
	}

	/* else find it in the existing connections */
	pthread_mutex_lock(&conn_mgr->connections_lock);
	list = conn_mgr->connections;
	while (list) {
		conn = (conn_t *) list->data;
		if (conn->socket == socket) {
			conn_lock(conn);

			/* must unlock connection! */
			conn_mgr_recv_event(conn_mgr, conn);
			return;
		}

		list = list->next;
	}

	fprintf(stderr,
			"%s: could not find a connection with socket %d!  this should, like, never happen.  you shouldn't be reading this.  if you are, bad things have happened.  your computer will likely melt after about 20 minutes.",
			__FUNCTION__, socket);
	abort();
}

void
conn_mgr_check_timeouts(conn_mgr_t * conn_mgr)
{
	lash_list_t *list, *next;
	conn_t *conn;
	time_t now;

	now = time(NULL);
	if (now == (time_t) - 1) {
		fprintf(stderr, "%s: could not get time, aborting!: %s\n",
				__FUNCTION__, strerror(errno));
		abort();
	}

	pthread_mutex_lock(&conn_mgr->connections_lock);
	list = conn_mgr->connections;
	while (list) {
		conn = (conn_t *) list->data;
		next = list->next;

		conn_lock(conn);

		if (conn_get_pinged(conn)) {
			if (conn_ping_timed_out(conn, now)) {
				fprintf(stderr,
						"%s: connection '%ld' has not responded to ping for %ld seconds, disconnecting it\n",
						__FUNCTION__, conn->id, CONN_TIMEOUT);
				conn_mgr_disconnect_client(conn_mgr, conn, 1);
			} else
				conn_unlock(conn);
		} else {
			if (conn_recv_timed_out(conn, now)) {
				/* send a ping */
				lash_comm_event_t *event;

/*              LASH_DEBUGARGS ("pinging connection '%ld'", conn_get_id (conn)); */

				event = lash_comm_event_new();
				lash_comm_event_set_type(event, LASH_Comm_Event_Ping);
				conn_mgr_send_client_lash_comm_event(conn_mgr, conn->id,
													 event);
				conn_set_ping_stamp(conn);
			}

			conn_unlock(conn);
		}

		list = next;
	}
	pthread_mutex_unlock(&conn_mgr->connections_lock);
}

void *
conn_mgr_recv_run(void *data)
{
	conn_mgr_t *conn_mgr;
	fd_set sockets;
	int err, i;
	struct timeval select_timeout;

	conn_mgr = (conn_mgr_t *) data;

	while (!conn_mgr->quit) {
		sockets = conn_mgr->sockets;
		select_timeout.tv_sec = 1;
		select_timeout.tv_usec = 0;

		err =
			select(conn_mgr->fd_max + 1, &sockets, NULL, NULL,
				   &select_timeout);
		if (err == -1) {
			if (errno == EINTR)
				continue;

			fprintf(stderr, "%s: error calling select(): %s\n", __FUNCTION__,
					strerror(errno));

			if (errno == EBADF)
				continue;
			else
				return NULL;
		}

		if (conn_mgr->quit)
			break;

		for (i = 0; i <= conn_mgr->fd_max; i++) {
			if (FD_ISSET(i, &sockets)) {
				conn_mgr_read_socket(conn_mgr, i);
			}
		}

		conn_mgr_check_timeouts(conn_mgr);
	}

	LASH_PRINT_DEBUG("finished");
	return NULL;
}

/****************************************
 ************ send thread ***************
 ****************************************/

void
conn_mgr_send_lash_comm_event_to_client(conn_t * conn,
										lash_comm_event_t * lash_comm_event)
{
	int err;

	err = lash_comm_send_event(conn->socket, lash_comm_event);

	if (err == -1) {
		fprintf(stderr, "%s: could not send event to client\n", __FUNCTION__);
	}

	lash_comm_event_destroy(lash_comm_event);
}

void
conn_mgr_send_disconnect_client(conn_mgr_t * conn_mgr, unsigned long conn_id)
{
	lash_list_t *list;
	conn_t *conn;

	pthread_mutex_lock(&conn_mgr->connections_lock);
	for (list = conn_mgr->connections; list; list = list->next) {
		conn = (conn_t *) list->data;

		if (conn->id == conn_id) {
			/* remove it */
			conn_lock(conn);
			conn_mgr_disconnect_client(conn_mgr, conn, 0);
			pthread_mutex_unlock(&conn_mgr->connections_lock);
			return;
		}
	}

	pthread_mutex_unlock(&conn_mgr->connections_lock);
	fprintf(stderr,
			"%s: request from server to remove unknown connection %ld\n",
			__FUNCTION__, conn_id);
}

void
conn_mgr_send_server_event_to_client(conn_mgr_t * conn_mgr,
									 server_event_t * server_event)
{
	lash_comm_event_t *lash_comm_event = NULL;
	unsigned long conn_id;
	lash_list_t *list;
	conn_t *conn;

	conn_id = server_event->conn_id;

	/* extract the comm event */
	switch (server_event->type) {
	case Client_Event:
		lash_comm_event = lash_comm_event_new();
		lash_comm_event_set_event(lash_comm_event,
								  server_event_take_lash_event(server_event));
		break;

	case Client_Config:
		lash_comm_event = lash_comm_event_new();
		lash_comm_event_set_config(lash_comm_event,
								   server_event_take_lash_config
								   (server_event));
		break;

	case Client_Comm_Event:
		lash_comm_event = server_event_take_lash_comm_event(server_event);
		break;

	case Client_Disconnect:
		conn_mgr_send_disconnect_client(conn_mgr, server_event->conn_id);
		server_event_destroy(server_event);
		return;

	case Client_Connect:
	default:
		fprintf(stderr,
				"%s: recieved unknown send request of type %d from server\n",
				__FUNCTION__, server_event->type);
		server_event_destroy(server_event);
		return;
	}
	server_event_destroy(server_event);

	/* attempt to find the connection */
	pthread_mutex_lock(&conn_mgr->connections_lock);
	list = conn_mgr->connections;
	while (list) {
		conn = (conn_t *) list->data;

		if (conn->id == conn_id) {
			/* send it */
			conn_lock(conn);
			pthread_mutex_unlock(&conn_mgr->connections_lock);
			conn_mgr_send_lash_comm_event_to_client(conn, lash_comm_event);
			conn_unlock(conn);
			return;
		}

		list = list->next;
	}

	pthread_mutex_unlock(&conn_mgr->connections_lock);

	fprintf(stderr, "%s: could not send event to unknown connection id %ld\n",
			__FUNCTION__, conn_id);
}

void *
conn_mgr_send_run(void *data)
{
	conn_mgr_t *conn_mgr;
	lash_list_t *list;
	server_event_t *server_event;

	LASH_PRINT_DEBUG("send thread starting");

	conn_mgr = (conn_mgr_t *) data;

	while (!conn_mgr->quit) {
		pthread_mutex_lock(&conn_mgr->client_event_lock);
		list = conn_mgr->client_events;
		if (list)
			conn_mgr->client_events = NULL;
		else {
			pthread_cond_wait(&conn_mgr->client_event_cond,
							  &conn_mgr->client_event_lock);
			list = conn_mgr->client_events;
			conn_mgr->client_events = NULL;
		}

		pthread_mutex_unlock(&conn_mgr->client_event_lock);

		if (conn_mgr->quit) {
			break;
		}

		while (list) {
			server_event = (server_event_t *) list->data;

			conn_mgr_send_server_event_to_client(conn_mgr, server_event);

			list = list->next;
		}
	}

	LASH_PRINT_DEBUG("finished");
	return NULL;
}

/* EOF */