/*
 * Copyright 2008-2014 Arsen Chaloyan
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * 
 * $Id: mrcp_client_connection.c 2249 2014-11-19 05:26:24Z achaloyan@gmail.com $
 */

#include "mrcp_connection.h"
#include "mrcp_client_connection.h"
#include "mrcp_control_descriptor.h"
#include "mrcp_resource_factory.h"
#include "mrcp_message.h"
#include "apt_text_stream.h"
#include "apt_poller_task.h"
#include "apt_log.h"


struct mrcp_connection_agent_t {
	/** List (ring) of MRCP connections */
	APR_RING_HEAD(mrcp_connection_head_t, mrcp_connection_t) connection_list;

	apr_pool_t                           *pool;
	apt_poller_task_t                    *task;
	const mrcp_resource_factory_t        *resource_factory;

	apr_uint32_t                          request_timeout;
	apt_bool_t                            offer_new_connection;
	apr_size_t                            tx_buffer_size;
	apr_size_t                            rx_buffer_size;

	void                                 *obj;
	const mrcp_connection_event_vtable_t *vtable;
};

typedef enum {
	CONNECTION_TASK_MSG_ADD_CHANNEL,
	CONNECTION_TASK_MSG_MODIFY_CHANNEL,
	CONNECTION_TASK_MSG_REMOVE_CHANNEL,
	CONNECTION_TASK_MSG_SEND_MESSAGE
} connection_task_msg_type_e;

typedef struct connection_task_msg_t connection_task_msg_t;
struct connection_task_msg_t {
	connection_task_msg_type_e type;
	mrcp_connection_agent_t   *agent;
	mrcp_control_channel_t    *channel;
	mrcp_control_descriptor_t *descriptor;
	mrcp_message_t            *message;
};


static apt_bool_t mrcp_client_agent_msg_process(apt_task_t *task, apt_task_msg_t *task_msg);
static apt_bool_t mrcp_client_poller_signal_process(void *obj, const apr_pollfd_t *descriptor);
static void mrcp_client_timer_proc(apt_timer_t *timer, void *obj);

/** Create connection agent. */
MRCP_DECLARE(mrcp_connection_agent_t*) mrcp_client_connection_agent_create(
											const char *id,
											apr_size_t max_connection_count, 
											apt_bool_t offer_new_connection,
											apr_pool_t *pool)
{
	apt_task_t *task;
	apt_task_vtable_t *vtable;
	apt_task_msg_pool_t *msg_pool;
	mrcp_connection_agent_t *agent;

	apt_log(APT_LOG_MARK,APT_PRIO_NOTICE,"Create MRCPv2 Agent [%s] [%"APR_SIZE_T_FMT"]",
				id,	max_connection_count);
	agent = apr_palloc(pool,sizeof(mrcp_connection_agent_t));
	agent->pool = pool;
	agent->request_timeout = 0;
	agent->offer_new_connection = offer_new_connection;
	agent->rx_buffer_size = MRCP_STREAM_BUFFER_SIZE;
	agent->tx_buffer_size = MRCP_STREAM_BUFFER_SIZE;

	msg_pool = apt_task_msg_pool_create_dynamic(sizeof(connection_task_msg_t),pool);

	agent->task = apt_poller_task_create(
					max_connection_count,
					mrcp_client_poller_signal_process,
					agent,
					msg_pool,
					pool);
	if(!agent->task) {
		return NULL;
	}

	task = apt_poller_task_base_get(agent->task);
	if(task) {
		apt_task_name_set(task,id);
	}

	vtable = apt_poller_task_vtable_get(agent->task);
	if(vtable) {
		vtable->process_msg = mrcp_client_agent_msg_process;
	}

	APR_RING_INIT(&agent->connection_list, mrcp_connection_t, link);
	return agent;
}

/** Destroy connection agent. */
MRCP_DECLARE(apt_bool_t) mrcp_client_connection_agent_destroy(mrcp_connection_agent_t *agent)
{
	apt_log(APT_LOG_MARK,APT_PRIO_NOTICE,"Destroy MRCPv2 Agent [%s]",
		mrcp_client_connection_agent_id_get(agent));
	return apt_poller_task_destroy(agent->task);
}

/** Start connection agent. */
MRCP_DECLARE(apt_bool_t) mrcp_client_connection_agent_start(mrcp_connection_agent_t *agent)
{
	return apt_poller_task_start(agent->task);
}

/** Terminate connection agent. */
MRCP_DECLARE(apt_bool_t) mrcp_client_connection_agent_terminate(mrcp_connection_agent_t *agent)
{
	return apt_poller_task_terminate(agent->task);
}

/** Set connection event handler. */
MRCP_DECLARE(void) mrcp_client_connection_agent_handler_set(
									mrcp_connection_agent_t *agent, 
									void *obj, 
									const mrcp_connection_event_vtable_t *vtable)
{
	agent->obj = obj;
	agent->vtable = vtable;
}

/** Set MRCP resource factory */
MRCP_DECLARE(void) mrcp_client_connection_resource_factory_set(
								mrcp_connection_agent_t *agent,
								const mrcp_resource_factory_t *resource_factroy)
{
	agent->resource_factory = resource_factroy;
}

/** Set rx buffer size */
MRCP_DECLARE(void) mrcp_client_connection_rx_size_set(
								mrcp_connection_agent_t *agent,
								apr_size_t size)
{
	if(size < MRCP_STREAM_BUFFER_SIZE) {
		size = MRCP_STREAM_BUFFER_SIZE;
	}
	agent->rx_buffer_size = size;
}

/** Set tx buffer size */
MRCP_DECLARE(void) mrcp_client_connection_tx_size_set(
								mrcp_connection_agent_t *agent,
								apr_size_t size)
{
	if(size < MRCP_STREAM_BUFFER_SIZE) {
		size = MRCP_STREAM_BUFFER_SIZE;
	}
	agent->tx_buffer_size = size;
}

/** Set request timeout */
MRCP_DECLARE(void) mrcp_client_connection_timeout_set(
								mrcp_connection_agent_t *agent,
								apr_size_t timeout)
{
	agent->request_timeout = (apr_uint32_t)timeout;
}

/** Get task */
MRCP_DECLARE(apt_task_t*) mrcp_client_connection_agent_task_get(const mrcp_connection_agent_t *agent)
{
	return apt_poller_task_base_get(agent->task);
}

/** Get external object */
MRCP_DECLARE(void*) mrcp_client_connection_agent_object_get(const mrcp_connection_agent_t *agent)
{
	return agent->obj;
}

/** Get string identifier */
MRCP_DECLARE(const char*) mrcp_client_connection_agent_id_get(const mrcp_connection_agent_t *agent)
{
	apt_task_t *task = apt_poller_task_base_get(agent->task);
	return apt_task_name_get(task);
}


/** Create control channel */
MRCP_DECLARE(mrcp_control_channel_t*) mrcp_client_control_channel_create(mrcp_connection_agent_t *agent, void *obj, apr_pool_t *pool)
{
	mrcp_control_channel_t *channel = apr_palloc(pool,sizeof(mrcp_control_channel_t));
	channel->agent = agent;
	channel->connection = NULL;
	channel->active_request = NULL;
	channel->request_timer = NULL;
	channel->removed = FALSE;
	channel->obj = obj;
	channel->log_obj = NULL;
	channel->pool = pool;

	channel->request_timer = apt_poller_task_timer_create(
								agent->task,
								mrcp_client_timer_proc,
								channel,
								pool);
	return channel;
}

/** Set the logger object */
MRCP_DECLARE(void) mrcp_client_control_channel_log_obj_set(mrcp_control_channel_t *channel, void *log_obj)
{
	channel->log_obj = log_obj;
}

/** Destroy MRCPv2 control channel */
MRCP_DECLARE(apt_bool_t) mrcp_client_control_channel_destroy(mrcp_control_channel_t *channel)
{
	if(channel && channel->connection && channel->removed == TRUE) {
		mrcp_connection_t *connection = channel->connection;
		channel->connection = NULL;
		apt_log(APT_LOG_MARK,APT_PRIO_NOTICE,"Destroy TCP/MRCPv2 Connection %s",connection->id);
		mrcp_connection_destroy(connection);
	}
	return TRUE;
}

/** Signal task message */
static apt_bool_t mrcp_client_control_message_signal(
								connection_task_msg_type_e type,
								mrcp_connection_agent_t *agent,
								mrcp_control_channel_t *channel,
								mrcp_control_descriptor_t *descriptor,
								mrcp_message_t *message)
{
	apt_task_t *task = apt_poller_task_base_get(agent->task);
	apt_task_msg_t *task_msg = apt_task_msg_get(task);
	if(task_msg) {
		connection_task_msg_t *msg = (connection_task_msg_t*)task_msg->data;
		msg->type = type;
		msg->agent = agent;
		msg->channel = channel;
		msg->descriptor = descriptor;
		msg->message = message;
		apt_task_msg_signal(task,task_msg);
	}
	return TRUE;
}

/** Add MRCPv2 control channel */
MRCP_DECLARE(apt_bool_t) mrcp_client_control_channel_add(mrcp_control_channel_t *channel, mrcp_control_descriptor_t *descriptor)
{
	return mrcp_client_control_message_signal(CONNECTION_TASK_MSG_ADD_CHANNEL,channel->agent,channel,descriptor,NULL);
}

/** Modify MRCPv2 control channel */
MRCP_DECLARE(apt_bool_t) mrcp_client_control_channel_modify(mrcp_control_channel_t *channel, mrcp_control_descriptor_t *descriptor)
{
	return mrcp_client_control_message_signal(CONNECTION_TASK_MSG_MODIFY_CHANNEL,channel->agent,channel,descriptor,NULL);
}

/** Remove MRCPv2 control channel */
MRCP_DECLARE(apt_bool_t) mrcp_client_control_channel_remove(mrcp_control_channel_t *channel)
{
	return mrcp_client_control_message_signal(CONNECTION_TASK_MSG_REMOVE_CHANNEL,channel->agent,channel,NULL,NULL);
}

/** Send MRCPv2 message */
MRCP_DECLARE(apt_bool_t) mrcp_client_control_message_send(mrcp_control_channel_t *channel, mrcp_message_t *message)
{
	return mrcp_client_control_message_signal(CONNECTION_TASK_MSG_SEND_MESSAGE,channel->agent,channel,NULL,message);
}

static mrcp_connection_t* mrcp_client_agent_connection_create(mrcp_connection_agent_t *agent, mrcp_control_descriptor_t *descriptor)
{
	char *local_ip = NULL;
	char *remote_ip = NULL;
	mrcp_connection_t *connection = mrcp_connection_create();

	apr_sockaddr_info_get(&connection->r_sockaddr,descriptor->ip.buf,APR_INET,descriptor->port,0,connection->pool);
	if(!connection->r_sockaddr) {
		mrcp_connection_destroy(connection);
		return NULL;
	}

	if(apr_socket_create(&connection->sock,connection->r_sockaddr->family,SOCK_STREAM,APR_PROTO_TCP,connection->pool) != APR_SUCCESS) {
		mrcp_connection_destroy(connection);
		return NULL;
	}

	apr_socket_opt_set(connection->sock, APR_SO_NONBLOCK, 0);
	apr_socket_timeout_set(connection->sock, -1);
	apr_socket_opt_set(connection->sock, APR_SO_REUSEADDR, 1);

	if(apr_socket_connect(connection->sock, connection->r_sockaddr) != APR_SUCCESS) {
		apr_socket_close(connection->sock);
		mrcp_connection_destroy(connection);
		return NULL;
	}

	if(apr_socket_addr_get(&connection->l_sockaddr,APR_LOCAL,connection->sock) != APR_SUCCESS) {
		apr_socket_close(connection->sock);
		mrcp_connection_destroy(connection);
		return NULL;
	}

	apr_sockaddr_ip_get(&local_ip,connection->l_sockaddr);
	apr_sockaddr_ip_get(&remote_ip,connection->r_sockaddr);
	connection->id = apr_psprintf(connection->pool,"%s:%hu <-> %s:%hu",
		local_ip,connection->l_sockaddr->port,
		remote_ip,connection->r_sockaddr->port);

	memset(&connection->sock_pfd,0,sizeof(apr_pollfd_t));
	connection->sock_pfd.desc_type = APR_POLL_SOCKET;
	connection->sock_pfd.reqevents = APR_POLLIN;
	connection->sock_pfd.desc.s = connection->sock;
	connection->sock_pfd.client_data = connection;
	if(apt_poller_task_descriptor_add(agent->task, &connection->sock_pfd) != TRUE) {
		apt_log(APT_LOG_MARK,APT_PRIO_WARNING,"Failed to Add to Pollset %s",connection->id);
		apr_socket_close(connection->sock);
		mrcp_connection_destroy(connection);
		return NULL;
	}
	
	apt_log(APT_LOG_MARK,APT_PRIO_NOTICE,"Established TCP/MRCPv2 Connection %s",connection->id);
	connection->agent = agent;
	APR_RING_INSERT_TAIL(&agent->connection_list,connection,mrcp_connection_t,link);
	
	connection->parser = mrcp_parser_create(agent->resource_factory,connection->pool);
	connection->generator = mrcp_generator_create(agent->resource_factory,connection->pool);

	connection->tx_buffer_size = agent->tx_buffer_size;
	connection->tx_buffer = apr_palloc(connection->pool,connection->tx_buffer_size+1);

	connection->rx_buffer_size = agent->rx_buffer_size;
	connection->rx_buffer = apr_palloc(connection->pool,connection->rx_buffer_size+1);
	apt_text_stream_init(&connection->rx_stream,connection->rx_buffer,connection->rx_buffer_size);

	if(apt_log_masking_get() != APT_LOG_MASKING_NONE) {
		connection->verbose = FALSE;
		mrcp_parser_verbose_set(connection->parser,TRUE);
		mrcp_generator_verbose_set(connection->generator,TRUE);
	}

	return connection;
}

static mrcp_connection_t* mrcp_client_agent_connection_find(mrcp_connection_agent_t *agent, mrcp_control_descriptor_t *descriptor)
{
	apr_sockaddr_t *sockaddr;
	mrcp_connection_t *connection;

	for(connection = APR_RING_FIRST(&agent->connection_list);
			connection != APR_RING_SENTINEL(&agent->connection_list, mrcp_connection_t, link);
				connection = APR_RING_NEXT(connection, link)) {
		if(apr_sockaddr_info_get(&sockaddr,descriptor->ip.buf,APR_INET,descriptor->port,0,connection->pool) == APR_SUCCESS) {
			if(apr_sockaddr_equal(sockaddr,connection->r_sockaddr) != 0 && 
				descriptor->port == connection->r_sockaddr->port) {
				return connection;
			}
		}
	}

	return NULL;
}

static apt_bool_t mrcp_client_agent_connection_remove(mrcp_connection_agent_t *agent, mrcp_connection_t *connection)
{
	/* remove from the list */
	APR_RING_REMOVE(connection,link);

	if(connection->sock) {
		apt_log(APT_LOG_MARK,APT_PRIO_INFO,"Close TCP/MRCPv2 Connection %s",connection->id);
		apt_poller_task_descriptor_remove(agent->task,&connection->sock_pfd);
		apr_socket_close(connection->sock);
		connection->sock = NULL;
	}
	return TRUE;
}

static apt_bool_t mrcp_client_agent_channel_add(mrcp_connection_agent_t *agent, mrcp_control_channel_t *channel, mrcp_control_descriptor_t *descriptor)
{
	if(agent->offer_new_connection == TRUE) {
		descriptor->connection_type = MRCP_CONNECTION_TYPE_NEW;
	}
	else {
		descriptor->connection_type = MRCP_CONNECTION_TYPE_EXISTING;
		if(APR_RING_EMPTY(&agent->connection_list, mrcp_connection_t, link)) {
			/* offer new connection if there is no established connection yet */
			descriptor->connection_type = MRCP_CONNECTION_TYPE_NEW;
		}
	}
	/* send response */
	return mrcp_control_channel_add_respond(agent->vtable,channel,descriptor,TRUE);
}

static apt_bool_t mrcp_client_agent_channel_modify(mrcp_connection_agent_t *agent, mrcp_control_channel_t *channel, mrcp_control_descriptor_t *descriptor)
{
	apt_bool_t status = TRUE;
	if(descriptor->port) {
		if(!channel->connection) {
			mrcp_connection_t *connection = NULL;
			apt_id_resource_generate(&descriptor->session_id,&descriptor->resource_name,'@',&channel->identifier,channel->pool);
			/* no connection yet */
			if(descriptor->connection_type == MRCP_CONNECTION_TYPE_EXISTING) {
				/* try to find existing connection */
				connection = mrcp_client_agent_connection_find(agent,descriptor);
				if(!connection) {
					apt_obj_log(APT_LOG_MARK,APT_PRIO_WARNING,channel->log_obj,"Found No Existing TCP/MRCPv2 Connection");
				}
			}
			if(!connection) {
				/* create new connection */
				connection = mrcp_client_agent_connection_create(agent,descriptor);
				if(!connection) {
					apt_obj_log(APT_LOG_MARK,APT_PRIO_WARNING,channel->log_obj,"Failed to Establish TCP/MRCPv2 Connection");
				}
			}

			if(connection) {
				mrcp_connection_channel_add(connection,channel);
				apt_obj_log(APT_LOG_MARK,APT_PRIO_INFO,channel->log_obj,"Add Control Channel <%s> %s [%d]",
						channel->identifier.buf,
						connection->id,
						apr_hash_count(connection->channel_table));
				if(descriptor->connection_type == MRCP_CONNECTION_TYPE_NEW) {
					/* set connection type to existing for the next offers / if any */
					descriptor->connection_type = MRCP_CONNECTION_TYPE_EXISTING;
				}
			}
			else {
				descriptor->port = 0;
				status = FALSE;
			}
		}
	}
	/* send response */
	return mrcp_control_channel_modify_respond(agent->vtable,channel,descriptor,status);
}

static apt_bool_t mrcp_client_agent_channel_remove(mrcp_connection_agent_t *agent, mrcp_control_channel_t *channel)
{
	if(channel->connection) {
		mrcp_connection_t *connection = channel->connection;
		mrcp_connection_channel_remove(connection,channel);
		apt_obj_log(APT_LOG_MARK,APT_PRIO_INFO,channel->log_obj,"Remove Control Channel <%s> [%d]",
				channel->identifier.buf,
				apr_hash_count(connection->channel_table));
		if(!connection->access_count) {
			mrcp_client_agent_connection_remove(agent,connection);
			/* set connection to be destroyed on channel destroy */
			channel->connection = connection;
			channel->removed = TRUE;
		}
	}
	
	/* send response */
	return mrcp_control_channel_remove_respond(agent->vtable,channel,TRUE);
}

static apt_bool_t mrcp_client_agent_request_cancel(mrcp_connection_agent_t *agent, mrcp_control_channel_t *channel, mrcp_message_t *message)
{
	mrcp_message_t *response;
	apt_obj_log(APT_LOG_MARK,APT_PRIO_WARNING,channel->log_obj,"Cancel MRCP Request <%s@%s> [%d]",
		MRCP_MESSAGE_SIDRES(message),
		message->start_line.request_id);
	response = mrcp_response_create(message,message->pool);
	response->start_line.status_code = MRCP_STATUS_CODE_METHOD_FAILED;
	return mrcp_connection_message_receive(agent->vtable,channel,response);
}

static apt_bool_t mrcp_client_agent_disconnect_raise(mrcp_connection_agent_t *agent, mrcp_connection_t *connection)
{
	mrcp_control_channel_t *channel;
	void *val;
	apr_hash_index_t *it = apr_hash_first(connection->pool,connection->channel_table);
	/* walk through the list of channels and raise disconnect event for them */
	for(; it; it = apr_hash_next(it)) {
		apr_hash_this(it,NULL,NULL,&val);
		channel = val;
		if(!channel) continue;

		if(channel->active_request) {
			mrcp_client_agent_request_cancel(channel->agent,channel,channel->active_request);
			channel->active_request = NULL;
			if(channel->request_timer) {
				apt_timer_kill(channel->request_timer);
			}
		}
		else if(agent->vtable->on_disconnect){
			agent->vtable->on_disconnect(channel);
		}
	}
	return TRUE;
}

static apt_bool_t mrcp_client_agent_messsage_send(mrcp_connection_agent_t *agent, mrcp_control_channel_t *channel, mrcp_message_t *message)
{
	apt_bool_t status = FALSE;
	mrcp_connection_t *connection = channel->connection;
	apt_text_stream_t stream;
	apt_message_status_e result;

	if(!connection || !connection->sock) {
		apt_obj_log(APT_LOG_MARK,APT_PRIO_WARNING,channel->log_obj,"Null MRCPv2 Connection "APT_SIDRES_FMT,MRCP_MESSAGE_SIDRES(message));
		mrcp_client_agent_request_cancel(agent,channel,message);
		return FALSE;
	}

	do {
		apt_text_stream_init(&stream,connection->tx_buffer,connection->tx_buffer_size);
		result = mrcp_generator_run(connection->generator,message,&stream);
		if(result != APT_MESSAGE_STATUS_INVALID) {
			stream.text.length = stream.pos - stream.text.buf;
			*stream.pos = '\0';

			apt_obj_log(APT_LOG_MARK,APT_PRIO_INFO,channel->log_obj,"Send MRCPv2 Data %s [%"APR_SIZE_T_FMT" bytes]\n%.*s",
				connection->id,
				stream.text.length,
				connection->verbose == TRUE ? stream.text.length : 0,
				stream.text.buf);

			if(apr_socket_send(connection->sock,stream.text.buf,&stream.text.length) == APR_SUCCESS) {
				status = TRUE;
			}
			else {
				apt_obj_log(APT_LOG_MARK,APT_PRIO_WARNING,channel->log_obj,"Failed to Send MRCPv2 Data %s",
					connection->id);
			}
		}
		else {
			apt_obj_log(APT_LOG_MARK,APT_PRIO_WARNING,channel->log_obj,"Failed to Generate MRCPv2 Data %s",
				connection->id);
		}
	}
	while(result == APT_MESSAGE_STATUS_INCOMPLETE);

	if(status == TRUE) {
		channel->active_request = message;
		if(channel->request_timer && agent->request_timeout) {
			apt_timer_set(channel->request_timer,agent->request_timeout);
		}
	}
	else {
		mrcp_client_agent_request_cancel(agent,channel,message);
	}
	return status;
}

static apt_bool_t mrcp_client_message_handler(mrcp_connection_t *connection, mrcp_message_t *message, apt_message_status_e status)
{
	if(status == APT_MESSAGE_STATUS_COMPLETE) {
		/* message is completely parsed */
		mrcp_control_channel_t *channel;
		apt_str_t identifier;
		apt_id_resource_generate(&message->channel_id.session_id,&message->channel_id.resource_name,'@',&identifier,message->pool);
		channel = mrcp_connection_channel_find(connection,&identifier);
		if(channel) {
			mrcp_connection_agent_t *agent = connection->agent;
			if(message->start_line.message_type == MRCP_MESSAGE_TYPE_RESPONSE) {
				if(!channel->active_request || 
					channel->active_request->start_line.request_id != message->start_line.request_id) {
					apt_obj_log(APT_LOG_MARK,APT_PRIO_WARNING,channel->log_obj,"Unexpected MRCP Response "APT_SIDRES_FMT" [%d]",
						MRCP_MESSAGE_SIDRES(message),
						message->start_line.request_id);
					return FALSE;
				}
				if(channel->request_timer) {
					apt_timer_kill(channel->request_timer);
				}
				channel->active_request = NULL;
			}

			mrcp_connection_message_receive(agent->vtable,channel,message);
		}
		else {
			apt_log(APT_LOG_MARK,APT_PRIO_WARNING,"Failed to Find Channel "APT_SIDRES_FMT" in Connection %s [%d]",
				MRCP_MESSAGE_SIDRES(message),
				connection->id,
				apr_hash_count(connection->channel_table));
		}
	}
	return TRUE;
}

/* Receive MRCP message through TCP/MRCPv2 connection */
static apt_bool_t mrcp_client_poller_signal_process(void *obj, const apr_pollfd_t *descriptor)
{
	mrcp_connection_agent_t *agent = obj;
	mrcp_connection_t *connection = descriptor->client_data;
	apr_status_t status;
	apr_size_t offset;
	apr_size_t length;
	apt_text_stream_t *stream;
	mrcp_message_t *message;
	apt_message_status_e msg_status;

	if(!connection || !connection->sock) {
		return FALSE;
	}
	stream = &connection->rx_stream;

	/* calculate offset remaining from the previous receive / if any */
	offset = stream->pos - stream->text.buf;
	/* calculate available length */
	length = connection->rx_buffer_size - offset;

	status = apr_socket_recv(connection->sock,stream->pos,&length);
	if(status == APR_EOF || length == 0) {
		apt_log(APT_LOG_MARK,APT_PRIO_INFO,"TCP/MRCPv2 Peer Disconnected %s",connection->id);
		apt_poller_task_descriptor_remove(agent->task,&connection->sock_pfd);
		apr_socket_close(connection->sock);
		connection->sock = NULL;

		mrcp_client_agent_disconnect_raise(agent,connection);
		return TRUE;
	}
	
	/* calculate actual length of the stream */
	stream->text.length = offset + length;
	stream->pos[length] = '\0';
	apt_log(APT_LOG_MARK,APT_PRIO_INFO,"Receive MRCPv2 Data %s [%"APR_SIZE_T_FMT" bytes]\n%.*s",
			connection->id,
			length,
			connection->verbose == TRUE ? length : 0,
			stream->pos);

	/* reset pos */
	apt_text_stream_reset(stream);

	do {
		msg_status = mrcp_parser_run(connection->parser,stream,&message);
		if(mrcp_client_message_handler(connection,message,msg_status) == FALSE) {
			return FALSE;
		}
	}
	while(apt_text_is_eos(stream) == FALSE);

	/* scroll remaining stream */
	apt_text_stream_scroll(stream);
	return TRUE;
}

/* Process task message */
static apt_bool_t mrcp_client_agent_msg_process(apt_task_t *task, apt_task_msg_t *task_msg)
{
	apt_poller_task_t *poller_task = apt_task_object_get(task);
	mrcp_connection_agent_t *agent = apt_poller_task_object_get(poller_task);
	connection_task_msg_t *msg = (connection_task_msg_t*) task_msg->data;

	switch(msg->type) {
		case CONNECTION_TASK_MSG_ADD_CHANNEL:
			mrcp_client_agent_channel_add(agent,msg->channel,msg->descriptor);
			break;
		case CONNECTION_TASK_MSG_MODIFY_CHANNEL:
			mrcp_client_agent_channel_modify(agent,msg->channel,msg->descriptor);
			break;
		case CONNECTION_TASK_MSG_REMOVE_CHANNEL:
			mrcp_client_agent_channel_remove(agent,msg->channel);
			break;
		case CONNECTION_TASK_MSG_SEND_MESSAGE:
			mrcp_client_agent_messsage_send(agent,msg->channel,msg->message);
			break;
	}

	return TRUE;
}

/* Timer callback */
static void mrcp_client_timer_proc(apt_timer_t *timer, void *obj)
{
	mrcp_control_channel_t *channel = obj;
	if(!channel) {
		return;
	}

	if(channel->request_timer == timer) {
		if(channel->active_request) {
			mrcp_client_agent_request_cancel(channel->agent,channel,channel->active_request);
			channel->active_request = NULL;
		}
	}
}