/*-
 * $Id: rr-tcpconnection.c,v 1.71 2003/01/06 22:59:32 jonas Exp $
 *
 * See the file LICENSE for redistribution information. 
 * If you have not received a copy of the license, please contact CodeFactory
 * by email at info@codefactory.se, or on the web at http://www.codefactory.se/
 * You may also write to: CodeFactory AB, SE-903 47, Umeå, Sweden.
 *
 * Copyright (c) 2002 Jonas Borgström <jonas@codefactory.se>
 * Copyright (c) 2002 Daniel Lundin   <daniel@codefactory.se>
 * Copyright (c) 2002 CodeFactory AB.  All rights reserved.
 */

#include <librr/rr.h>

#ifdef G_PLATFORM_WIN32
#include "rr-win32.h"
#else
#include <unistd.h>
#include <netdb.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#endif

#include <stdio.h>
#include <string.h>

#define BUFFER_SIZE 32768

#define MAX_THREADS 10

typedef enum { 
	EVENT_IN  = 1 << 0,
	EVENT_OUT = 1 << 1,
	EVENT_ERR = 1 << 2
} Events;

static GObjectClass *parent_class = NULL;
static void finalize (GObject *object);

static gboolean enable_input   (RRConnection *conn);
static gboolean disable_input  (RRConnection *conn);

static gboolean enable_output  (RRConnection *conn);
static gboolean disable_output (RRConnection *conn);

static gboolean in_event (GIOChannel *source, GIOCondition condition, 
			  gpointer data);
static gboolean out_event (GIOChannel *source, GIOCondition condition, 
			   gpointer data);
static gboolean real_disconnect (RRTCPConnection *tcpc, Events ignore, 
				 GError **error);
static gboolean disconnect (RRConnection *connection, GError **error);

/* same as g_source_remove except that it uses rr_get_main_context */
static gboolean
source_remove (guint tag)
{
	GSource *source;
  
	g_return_val_if_fail (tag > 0, FALSE);

	source = g_main_context_find_source_by_id (rr_get_main_context (), tag);
	if (source)
		g_source_destroy (source);
	
	return source != NULL;
}

static guint
add_watch_full (GIOChannel *channel, gint priority, GIOCondition condition, 
		GIOFunc func, gpointer user_data, GDestroyNotify notify)
{
	GSource *source;
	guint id;

	source = g_io_create_watch (channel, condition);
	if (priority != G_PRIORITY_DEFAULT)
		g_source_set_priority (source, priority);
	g_source_set_callback (source, (GSourceFunc)func, user_data, notify);
	id = g_source_attach (source, rr_get_main_context ());
	g_source_unref (source);

	return id;
}

static void
rr_tcp_connection_init (GObject *object)
{
	RRTCPConnection *tcpc = (RRTCPConnection *)object;

	tcpc->iochannel = NULL;
	tcpc->ibuffer = g_new (gchar, BUFFER_SIZE);
	tcpc->obuffer = g_new (gchar, BUFFER_SIZE);
	tcpc->ibuffer_offset = 0;

	g_static_mutex_init (&tcpc->event_lock);

	tcpc->active_mutex  = g_mutex_new ();
	tcpc->active_cond   = g_cond_new  ();

	tcpc->filter = rr_tcp_filter_new ();

	rr_filterstack_push (RR_CONNECTION (tcpc)->filter_stack,
			     RR_FILTER (tcpc->filter));
}

static void
rr_tcp_connection_class_init (GObjectClass *klass)
{
	RRConnectionClass *conn_class = (RRConnectionClass *)klass;

	conn_class->enable_input   = enable_input;
	conn_class->disable_input  = disable_input;

	conn_class->enable_output  = enable_output;
	conn_class->disable_output = disable_output;

	conn_class->disconnect = disconnect;

	klass->finalize = finalize;

	parent_class = g_type_class_peek_parent (klass);

	/* Register the "SEQ" frame type */
	rr_framefactory_register_type ("SEQ", RR_TYPE_FRAME_SEQ);
}

GType 
rr_tcp_connection_get_type (void)
{
	static GType rr_type = 0;

	if (!rr_type) {
		static GTypeInfo type_info = {
			sizeof (RRTCPConnectionClass),
			NULL,
			NULL,
			(GClassInitFunc) rr_tcp_connection_class_init,
			NULL,
			NULL,
			sizeof (RRTCPConnection),
			16,
			(GInstanceInitFunc) rr_tcp_connection_init
		};
		rr_type = g_type_register_static (RR_TYPE_CONNECTION, 
						  "RRTCPConnection", 
						  &type_info, 0);

#ifdef G_PLATFORM_WIN32
	rr_win32_init_winsock ();
#endif
	}
	return rr_type;
}

/**
 * rr_tcp_connection_new_unconnected:
 * @profreg: A #RRProfileRegistry of profiles this connection supports.
 * 
 * This function creates a new tcp connection object. 
 * Note: the profreg will be g_object unref:ed.
 * 
 * Return value: the new #RRTCPConnection.
 **/
RRTCPConnection *
rr_tcp_connection_new_unconnected (RRProfileRegistry *profreg)
{
	RRTCPConnection *tcpc;
	RRConnection *conn;

	tcpc = g_object_new (RR_TYPE_TCP_CONNECTION, NULL);
	conn = RR_CONNECTION (tcpc);

	if (profreg) {
		rr_connection_set_profile_registry (conn, profreg);
		g_object_unref (G_OBJECT (profreg));
	}

	rr_connection_add_channel (conn, RR_CHANNEL (conn->manager));

	return tcpc;
}

static void
event_join (RRTCPConnection *tcpc, Events event)
{
	g_mutex_lock (tcpc->active_mutex);
	switch (event) {
	case EVENT_IN:
		while (tcpc->in_active > 0)
			g_cond_wait (tcpc->active_cond, tcpc->active_mutex);
		break;
	case EVENT_OUT:
		while (tcpc->out_active > 0)
			g_cond_wait (tcpc->active_cond, tcpc->active_mutex);
		break;
	case EVENT_ERR:
		while (tcpc->err_active > 0)
			g_cond_wait (tcpc->active_cond, tcpc->active_mutex);
		break;
	default:
		g_assert_not_reached ();
	}
	g_mutex_unlock (tcpc->active_mutex);
}

static void
set_active (RRTCPConnection *tcpc, Events event)
{
	g_mutex_lock (tcpc->active_mutex);
	switch (event) {
	case EVENT_IN:
		tcpc->in_active++;
		break;
	case EVENT_OUT:
		tcpc->out_active++;
		break;
	case EVENT_ERR:
		tcpc->err_active++;
		break;
	default:
		g_assert_not_reached ();
	}
	g_cond_signal (tcpc->active_cond);
	g_mutex_unlock (tcpc->active_mutex);
}

static void
set_inactive (RRTCPConnection *tcpc, Events event)
{
	g_mutex_lock (tcpc->active_mutex);
	switch (event) {
	case EVENT_IN:
		tcpc->in_active--;
		break;
	case EVENT_OUT:
		tcpc->out_active--;
		break;
	case EVENT_ERR:
		tcpc->err_active--;
		break;
	default:
		g_assert_not_reached ();
	}
	g_cond_signal (tcpc->active_cond);
	g_mutex_unlock (tcpc->active_mutex);
}

static void
in_removed (gpointer data)
{
	RRTCPConnection *tcpc = RR_TCP_CONNECTION (data);

	set_inactive (tcpc, EVENT_IN);
}

static void
out_removed (gpointer data)
{
	RRTCPConnection *tcpc = RR_TCP_CONNECTION (data);

	set_inactive (tcpc, EVENT_OUT);
}

static void
err_removed (gpointer data)
{
	RRTCPConnection *tcpc = RR_TCP_CONNECTION (data);

	set_inactive (tcpc, EVENT_ERR);
}

static gboolean
enable_input (RRConnection *conn)
{
	RRTCPConnection *tcpc;

	if (conn->connected == FALSE)
		return FALSE;

	tcpc = RR_TCP_CONNECTION (conn);

	g_static_mutex_lock (&tcpc->event_lock);

	if (tcpc->in_event == 0) {
		set_active (tcpc, EVENT_IN);
		tcpc->in_event = add_watch_full (tcpc->iochannel, 0, G_IO_IN, 
						 in_event, tcpc, in_removed);
		if (tcpc->in_event <= 0) {
			g_static_mutex_unlock (&tcpc->event_lock);
			return FALSE;
		}
	}
	g_static_mutex_unlock (&tcpc->event_lock);
	return TRUE;
}

static gboolean
disable_input (RRConnection *conn)
{
	RRTCPConnection *tcpc;
	gboolean ret = TRUE;

	if (conn->connected == FALSE)
		return FALSE;

	tcpc = RR_TCP_CONNECTION (conn);

	g_static_mutex_lock (&tcpc->event_lock);
	if (tcpc->in_event) {
		ret = source_remove (tcpc->in_event);
		tcpc->in_event = 0;
	}
	g_static_mutex_unlock (&tcpc->event_lock);
	return ret;
}

static gboolean
disable_output (RRConnection *conn)
{
	RRTCPConnection *tcpc;
	gboolean ret = TRUE;

	if (conn->connected == FALSE)
		return FALSE;

	tcpc = RR_TCP_CONNECTION (conn);

	g_static_mutex_lock (&tcpc->event_lock);
	if (tcpc->out_event) {
		ret = source_remove (tcpc->out_event);
		tcpc->out_event = 0;
	}
	g_static_mutex_unlock (&tcpc->event_lock);
	return ret;
}

static gboolean
enable_output (RRConnection *conn)
{
	RRTCPConnection *tcpc;

	if (conn->connected == FALSE)
		return FALSE;

	tcpc = RR_TCP_CONNECTION (conn);
	g_static_mutex_lock (&tcpc->event_lock);

	if (tcpc->out_event == 0 && 
	    rr_connection_pending_transmissions_p (conn)) {

		set_active (tcpc, EVENT_OUT);
		tcpc->out_event = add_watch_full (tcpc->iochannel, 0, G_IO_OUT,
						  out_event, tcpc, out_removed);
		if (tcpc->out_event <= 0) {
			g_static_mutex_unlock (&tcpc->event_lock);
			return FALSE;
		}
	}
	g_static_mutex_unlock (&tcpc->event_lock);
	return TRUE;
}

static void
handle_seq_frame (RRTCPConnection *tcpc, RRFrameSeq *frame)
{
	RRChannel *channel;
	gint size;

	channel = rr_connection_get_channel_locked (RR_CONNECTION (tcpc),
						    frame->channel_id);
	/* Just ignore bogus SEQ frames */
	if (channel == NULL) {

		g_object_unref (G_OBJECT (frame));
		return;
	}

	size = frame->size + frame->seqno - channel->seq_out;


	rr_debug3 ("connection::handle_seq_frame chan=%d, window size set to %d", 
		   frame->channel_id, size);
	channel->window_out = size;
	channel->starved = FALSE;

	rr_channel_unlock (channel);

	rr_connection_enable_output (RR_CONNECTION (tcpc));

	g_object_unref (G_OBJECT (frame));
}

static void
send_seq_frame (RRTCPConnection *tcpc, RRChannel *channel)
{
	RRFrameSeq *frame;

	g_return_if_fail (RR_IS_TCP_CONNECTION (tcpc));
	g_return_if_fail (RR_IS_CHANNEL (channel));

	if (channel->window_in < (0.33 * channel->window_size)) {

		frame = rr_frame_seq_new (channel->id, channel->seq_in, 
					  channel->window_size);
		
		channel->window_in = channel->window_size;
		
		rr_connection_send_frame (RR_CONNECTION (tcpc), 
					  RR_FRAME (frame), NULL);
	}
}

static gboolean
frame_divider (RRTCPConnection *tcpc, gchar *buffer, gint data_len,
	       GError **error)
{
	RRConnection *conn;
	gint offset = 0, len = data_len;

	g_return_val_if_fail (RR_IS_TCP_CONNECTION (tcpc), FALSE);
	g_return_val_if_fail (buffer != NULL, FALSE);
	g_return_val_if_fail (data_len > 0, FALSE);

	conn = RR_CONNECTION (tcpc);

	tcpc->ibuffer_offset = 0;

	for (;;) {
		RRFrame *frame;
		gint bytes;

		bytes = rr_framefactory_parse_frame (RR_CONNECTION (tcpc),
						     buffer + offset, len, 
						     &frame, error);
		/* Parse error? */
		if (bytes < 0) 
			return FALSE;

		if (frame) {
			rr_debug4 ("connection::frame_divider parsed frame: "
				   "chan=%d msgno=%d seq=%d",
				 frame->channel_id, frame->msgno,
				 frame->seqno);
			if (RR_IS_FRAME_SEQ (frame)) {
				handle_seq_frame (tcpc, RR_FRAME_SEQ (frame));
			}
			else {
				RRChannel *channel;
				
				channel = rr_connection_get_channel_locked (conn,
									    frame->channel_id);
				if (channel == NULL) {

					rr_debug1 ("connection::frame_divider bogus channel id '%d'\n", 
						   frame->channel_id);
					g_object_unref (G_OBJECT (frame));
				}
				else {
					/* FIXME: this is ugly */
					if (frame->seqno == 0 &&
					    RR_MANAGER (conn->manager)->expects_greeting) {
						channel->seq_in = 0;
					} 
					else if (frame->seqno != channel->seq_in) {
						
						g_set_error (error,
							     RR_BEEP_ERROR,
							     RR_BEEP_CODE_SYNTAX_ERROR,
							     "seqno missmatch %d != %d", 
							     frame->seqno, 
							     channel->seq_in);

						g_object_unref (G_OBJECT (frame));
						rr_channel_unlock (channel);
						return FALSE;
					}

					channel->window_in -= frame->size;
					channel->seq_in    += frame->size;
					
					send_seq_frame (tcpc, channel);

					rr_channel_frame_available (channel, frame);
					rr_channel_unlock (channel);
					g_object_unref (G_OBJECT (frame));
				}
			}
		}
		if (bytes == 0) {
			memcpy (buffer, buffer + offset, len);
			tcpc->ibuffer_offset = len;
			break;
		}
		else {
			offset += bytes;
			len -= bytes;
			if (len == 0)
				break;
		}
	}
	return TRUE;
}

static void
report_error_and_disconnect (RRTCPConnection *tcpc, 
			     const gchar *function_name, Events ignore,
			     GError *error)
{
	if (error) {
		
		rr_debug2 ("connection::disconnect (%p):%s failed: '%s'", 
			   tcpc,
			   function_name, error->message);
		g_error_free (error);
	}
	else
		rr_debug2 ("connection::disconnect (%p):%s failed.", 
			   tcpc,
			   function_name);
		
	if (ignore)
		real_disconnect (tcpc, ignore, NULL);
}

static gboolean
in_event (GIOChannel *source, GIOCondition condition, gpointer data)
{
	RRTCPConnection *tcpc;
	RRConnection *conn;
	GIOStatus status;
	GError *error = NULL;
	gsize read_bytes;

/* 	g_print ("in_event: %p condition = %0x\n", data, condition); */

	conn = RR_CONNECTION (data);
	tcpc = RR_TCP_CONNECTION (data);

	/* Is the buffer full */
	if (BUFFER_SIZE  - tcpc->ibuffer_offset <= 0) {

		report_error_and_disconnect (tcpc, 
					     "frame_divider failed: buffer full", 
					     EVENT_IN, error);
		return FALSE;
	}

	status = rr_filterstack_read (conn->filter_stack,
				      tcpc->ibuffer + tcpc->ibuffer_offset,
				      BUFFER_SIZE  - tcpc->ibuffer_offset,
				      &read_bytes, &error);

	if (status == G_IO_STATUS_EOF) {
		return real_disconnect (tcpc, EVENT_IN, NULL);
	}
	else if (status != G_IO_STATUS_NORMAL) {
		report_error_and_disconnect (tcpc, "g_io_channel_read_chars",
					     EVENT_IN, error);
		return FALSE;
	}
	if (!frame_divider (tcpc, tcpc->ibuffer, 
			    read_bytes + tcpc->ibuffer_offset, &error)) {
		
		report_error_and_disconnect (tcpc, "frame_divider", EVENT_IN, error);
		return FALSE;
	}
	return TRUE;
}

static gboolean
out_event (GIOChannel *source, GIOCondition condition, gpointer data)
{
	RRFrame *frame;
	GError  *error = NULL;
	GIOStatus status;
	RRConnection *conn;
	RRTCPConnection *tcpc;
	gboolean ret;

	conn = RR_CONNECTION (data);
	tcpc = RR_TCP_CONNECTION (data);

/* 	g_print ("out_event: %p condition = %0x\n", data, condition); */

	frame = rr_connection_get_next_frame (conn, BUFFER_SIZE);
	if (frame == NULL) {

		g_static_mutex_lock (&tcpc->event_lock);
		tcpc->out_event = 0;
		g_static_mutex_unlock (&tcpc->event_lock);
		return FALSE;
	}
	else {
		gsize len, written;
		
		len = rr_frame_build (frame, tcpc->obuffer);
		status = rr_filterstack_write (conn->filter_stack, 
					       tcpc->obuffer,
					       len, &written, &error);

		if (status != G_IO_STATUS_NORMAL) {
			report_error_and_disconnect (tcpc, 
						     "g_io_channel_write_chars",
						     EVENT_OUT, error);
			return FALSE;
		}
		status = g_io_channel_flush (tcpc->iochannel, &error);
		if (status != G_IO_STATUS_NORMAL) {
			report_error_and_disconnect (tcpc, 
						     "g_io_channel_flush",
						     EVENT_OUT, error);
			return FALSE;
		}
		g_object_unref (G_OBJECT (frame));
	}

	g_static_mutex_lock (&tcpc->event_lock);
	ret = rr_connection_pending_transmissions_p (conn);
	if (ret == FALSE) {
		tcpc->out_event = 0;
	}
	g_static_mutex_unlock (&tcpc->event_lock);
	if (ret == FALSE) {
		g_mutex_lock (conn->out_queue_lock);
		g_cond_broadcast (conn->out_queue_cond);
		g_mutex_unlock (conn->out_queue_lock);
		rr_callback_list_execute (conn->quiescence_cb);
		rr_callback_list_free (conn->quiescence_cb);
		conn->quiescence_cb = NULL;
	}
	return ret;
}

static gboolean
error_event (GIOChannel *source, GIOCondition condition, gpointer data)
{
	RRTCPConnection *tcpc;

	g_return_val_if_fail (RR_IS_TCP_CONNECTION (data), FALSE);

	tcpc = RR_TCP_CONNECTION (data);

	rr_debug3 ("connection::error %p error event on  cond= %0x\n", data, 
		   condition);

	real_disconnect (tcpc, EVENT_ERR, NULL);

	return FALSE;
}

/**
 * rr_tcp_connection_connect_fd:
 * @tcpc: A #RRTCPConnection
 * @fd: an open file descriptor
 * @role: a #RRRole
 * @error: @error: location to return an error of type G_IO_ERROR, RR_ERROR or 
 * RR_BEEP_ERROR.
 * 
 * Opens a BEEP connection to a beep peer, using an already open socket.
 * 
 * Return value: %TRUE on success, %FALSE on failure.
 **/
gboolean
rr_tcp_connection_connect_fd (RRTCPConnection *tcpc,
			   gint fd, RRRole role, GError **error)
{
	RRConnection *conn = RR_CONNECTION (tcpc);

	g_return_val_if_fail (RR_IS_TCP_CONNECTION (tcpc), FALSE);
	g_return_val_if_fail (fd > 0, FALSE);

	rr_debug3 ("connection::connect_fd %p fd=%d\n", tcpc, fd);

	tcpc->iochannel = g_io_channel_unix_new (fd);
	rr_tcp_filter_set_iochannel (tcpc->filter, tcpc->iochannel);
	g_io_channel_set_close_on_unref (tcpc->iochannel, TRUE);
	g_io_channel_set_encoding (tcpc->iochannel, NULL, NULL);

#ifdef G_PLATFORM_WIN32
	if (!rr_win32_enable_nonblock (fd)) {
		g_set_error (error, RR_ERROR, RR_ERROR_OTHER,
			     "rr_win32_enable_nonblock failed.");
		return FALSE;
	}
#else
	if (g_io_channel_set_flags (tcpc->iochannel, 
				    G_IO_FLAG_NONBLOCK, error) == G_IO_STATUS_ERROR)
		return FALSE;
#endif
	set_active (tcpc, EVENT_ERR);

	tcpc->err_event = add_watch_full (tcpc->iochannel, 0, 
					  G_IO_HUP | G_IO_ERR, 
					  error_event, tcpc,
					  err_removed);

	set_active (tcpc, EVENT_IN);

	tcpc->in_event = add_watch_full (tcpc->iochannel, 0, G_IO_IN, 
					 in_event, tcpc, in_removed);
	conn->role = role;
	conn->connected = TRUE;

	return rr_manager_send_greeting (RR_MANAGER(conn->manager), error);
}

/**
 * rr_tcp_connection_new:
 * @profreg: a RRProfileRegistry
 * @hostname: a hostname
 * @port: a port
 * @error: location to return an error of type G_IO_ERROR, RR_ERROR or 
 * RR_BEEP_ERROR.
 * 
 * Creates a new RRTCPConnection instance and tries to opens a connection
 * to a beep peer, using the TCP/IP protocol.
 * 
 * Return value: A new (connected) RRTCPConnection instance on success,
 * and NULL on failure.
 **/
RRConnection *
rr_tcp_connection_new (RRProfileRegistry *profreg,
		       const gchar *hostname, 
		       gint port, GError **error)
{
	RRTCPConnection *tcpc;
	
	tcpc = rr_tcp_connection_new_unconnected (profreg);
	if (tcpc == NULL)
		return FALSE;

	if (rr_tcp_connection_connect (tcpc, hostname, port, error))
		return RR_CONNECTION (tcpc);
	else {
		g_object_unref (G_OBJECT (tcpc));
		return NULL;
	}
}

/**
 * rr_tcp_connection_connect:
 * @connection: A #RRTCPConnection
 * @hostname: a hostname
 * @port: a port
 * @error: location to return an error of type G_IO_ERROR, RR_ERROR or 
 * RR_BEEP_ERROR.
 * 
 * Opens a connection to a beep peer, using the TCP/IP protocol.
 * 
 * Return value: %TRUE on success, %FALSE on failure.
 **/
gboolean
rr_tcp_connection_connect (RRTCPConnection *tcpc,
			   const gchar *hostname, 
			   gint port, GError **error)
{
	RRConnection *conn = RR_CONNECTION (tcpc);
	struct hostent *he;
        struct in_addr *haddr;
        struct sockaddr_in saddr;
	gint fd;

	rr_debug1 ("connection::connect %p '%s':%d", tcpc,
		   hostname, port);

        he = gethostbyname(hostname);
        if (he == NULL) {
		g_set_error (error,
			     RR_ERROR,                /* error domain */
			     RR_ERROR_GETHOSTBYNAME,  /* error code */
			     "gethostbyname failed"); /* error message */
		return FALSE;
        }

        haddr = ((struct in_addr *) (he->h_addr_list)[0]);

        fd = socket(AF_INET, SOCK_STREAM, 0);
        memset(&saddr, 0, sizeof(saddr));
        memcpy(&saddr.sin_addr, haddr, sizeof(struct in_addr));
        saddr.sin_family = AF_INET;
        saddr.sin_port = htons(port);
 
        if (connect(fd, (struct sockaddr *)&saddr, sizeof(saddr)) < 0) {
		g_set_error (error,
			     RR_ERROR,            /* error domain */
			     RR_ERROR_CONNECT,    /* error code */
			     "connect() failed"); /* error message */
		close (fd);
		return FALSE;
        }
	if (!rr_tcp_connection_connect_fd (tcpc, fd, RR_ROLE_INITIATOR, error))
		return FALSE;
	return rr_manager_wait_for_greeting (RR_MANAGER(conn->manager), error);

}

static gboolean
real_disconnect (RRTCPConnection *tcpc, Events ignore, GError **error)
{
	RRConnection *conn = RR_CONNECTION (tcpc);

	conn->connected = FALSE;

	if (tcpc->iochannel) {
		GIOChannel *iochannel = tcpc->iochannel;
		tcpc->iochannel = NULL;

		rr_debug1 ("connection::disconnect: %p", conn);

		rr_main_work_pool_join (RRWPGROUP (conn));

		/* Remove the iostream from the mainloop */
		g_static_mutex_lock (&tcpc->event_lock);
		if (tcpc->in_event) {
			source_remove (tcpc->in_event);
			tcpc->in_event = 0;
		}
		if (tcpc->err_event) {
			source_remove (tcpc->err_event);
			tcpc->err_event = 0;
		}
		if (tcpc->out_event) {
			source_remove (tcpc->out_event);
			tcpc->out_event = 0;
		}
		g_static_mutex_unlock (&tcpc->event_lock);

		if (!(ignore & EVENT_IN))
			event_join (tcpc, EVENT_IN);

		if (!(ignore & EVENT_ERR))
			event_join (tcpc, EVENT_ERR);

		if (!(ignore & EVENT_OUT))
			event_join (tcpc, EVENT_OUT);

		rr_connection_close_all (conn);

		/* Close the channel */
		g_io_channel_unref (iochannel);

		if (conn->listener)
			rr_listener_remove_connection (conn->listener, conn);
	}
	return TRUE;
}

static gboolean
disconnect (RRConnection *connection, GError **error)
{
	RRTCPConnection *tcpc = RR_TCP_CONNECTION (connection);
	RRManager *manager = RR_MANAGER (connection->manager);
	gboolean ret;

	if (!rr_connection_wait_quiescence (connection, error))
		return FALSE;

	if (!rr_manager_close_channel (manager, RR_CHANNEL (manager), 
				       200, NULL, "disconnect()", error)) {

		/* failed to close channel 0, but we don't care because,
		   we are going to disconnect anyway */
		if (error && *error) {
			g_error_free (*error);
			*error = NULL;
		}
	}
	ret = real_disconnect (tcpc, 0, error);
	g_object_unref (G_OBJECT (tcpc));

	return ret;
}

/**
 * rr_tcp_connection_get_fd:
 * @tcpc: A #RRTCPConnection
 * 
 * Returns the file-descriptor associated with the socket.
 * 
 * Return value: the file-descriptor associated with the socket.
 **/
gint
rr_tcp_connection_get_fd (RRTCPConnection *tcpc)
{
	return g_io_channel_unix_get_fd (tcpc->iochannel);
}

static void
finalize (GObject *object)
{
	RRTCPConnection *tcpc = RR_TCP_CONNECTION (object);

/* 	rr_tcp_connection_disconnect (tcpc, NULL); */
	real_disconnect (tcpc, 0, NULL);

	g_static_mutex_free (&tcpc->event_lock);

	/* Make sure the watches are removed */
	event_join (tcpc, EVENT_IN);
	event_join (tcpc, EVENT_OUT);
	event_join (tcpc, EVENT_ERR);
	g_mutex_free (tcpc->active_mutex);
	g_cond_free  (tcpc->active_cond);

	g_free (tcpc->ibuffer);
	g_free (tcpc->obuffer);

	parent_class->finalize (object);
}


syntax highlighted by Code2HTML, v. 0.9.1