/*
 * OpenMRCP - Open Source Media Resource Control Protocol Stack
 * Copyright (C) 2007, Cepstral LLC
 *
 * Version: MPL 1.1
 *
 * The contents of this file are subject to the Mozilla Public License Version
 * 1.1 (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.mozilla.org/MPL/
 *
 * Software distributed under the License is distributed on an "AS IS" basis,
 * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
 * for the specific language governing rights and limitations under the
 * License.
 *
 * Author(s):
 * Arsen Chaloyan <achaloyan@yahoo.com>
 *
 * Contributor(s):
 * Helder Oliveira Ribeiro <Helder.Ribeiro@altitude.com>
 *
 */

#include <apr_network_io.h>
#include <apr_poll.h>
#include "mrcp_client_proto_agent.h"
#include "apt_producer_task.h"
#include "mrcp_parser.h"

#define MRCP_CONNECTION_MAX_COUNT 10

typedef struct mrcp_v2_connection_t mrcp_v2_connection_t;

struct mrcp_v2_connection_t {
	mrcp_connection_t    *base;
	apr_pool_t           *pool;

	apr_socket_t         *sock; /* connected socket */
	apr_pollfd_t          sock_pfd;

	char                 *remote_ip;
	unsigned short        remote_port;
	size_t                access_count;

	mrcp_v2_connection_t *next;
	mrcp_v2_connection_t *prev;
};


typedef struct mrcp_v2_agent_t mrcp_v2_agent_t;

struct mrcp_v2_agent_t {
	mrcp_proto_agent_t    proto_agent;
	apt_producer_task_t  *producer_task;

	apr_pool_t           *pool;

	apr_pollset_t        *pollset;

	apr_sockaddr_t       *sockaddr;
	apr_socket_t         *listen_sock; /* listening socket */
	apr_pollfd_t          listen_sock_pfd;
	apr_socket_t         *accept_sock; /* accepted socket */
	apr_pollfd_t          accept_sock_pfd;

	apr_socket_t         *control_sock; /* control socket */

	mrcp_v2_connection_t *connection_head;
	mrcp_v2_connection_t *connection_tail;
};


typedef enum {
	PROTO_AGENT_EVENT_CONNECTED,
	PROTO_AGENT_EVENT_DISCONNECTED,
	PROTO_AGENT_EVENT_RECEIVED,
	PROTO_AGENT_EVENT_STARTED,
	PROTO_AGENT_EVENT_TERMINATED
} mrcp_proto_agent_event_id;


typedef struct mrcp_proto_agent_event_t mrcp_proto_agent_event_t;

struct mrcp_proto_agent_event_t {
	mrcp_proto_agent_event_id id;
	mrcp_v2_connection_t     *connection;
	mrcp_message_t           *message;
};


typedef enum {
	PROTO_AGENT_SIGNAL_CONNECT,
	PROTO_AGENT_SIGNAL_DISCONNECT,
	PROTO_AGENT_SIGNAL_SEND,
} mrcp_proto_agent_signal_id;

typedef struct mrcp_proto_agent_signal_t mrcp_proto_agent_signal_t;

struct mrcp_proto_agent_signal_t {
	mrcp_proto_agent_signal_id id;
	mrcp_v2_connection_t      *connection;
	mrcp_message_t            *message;
};


static mrcp_status_t mrcp_client_agent_destroy(mrcp_module_t *module);
static mrcp_module_state_t mrcp_client_agent_open(mrcp_module_t *module);
static mrcp_module_state_t mrcp_client_agent_close(mrcp_module_t *module);
static mrcp_status_t mrcp_client_agent_signal_handler(mrcp_module_t *module, apt_task_msg_t *msg);

static const mrcp_module_method_set_t module_method_set = {
	mrcp_client_agent_destroy,
	mrcp_client_agent_open,
	mrcp_client_agent_close,
	mrcp_client_agent_signal_handler
};


static mrcp_connection_t* mrcp_client_agent_connect(mrcp_proto_agent_t *proto_agent, mrcp_signaling_channel_t *channel, const char *remote_ip, unsigned short remote_port);

static const mrcp_proto_agent_method_set_t agent_method_set = {
	mrcp_client_agent_connect,
};


static mrcp_status_t mrcp_client_agent_disconnect(mrcp_connection_t *base);
static mrcp_status_t mrcp_client_agent_message_send(mrcp_connection_t *base, mrcp_message_t *mrcp_message);

static const mrcp_connection_method_set_t connection_method_set = {
	mrcp_client_agent_disconnect,
	mrcp_client_agent_message_send
};



static mrcp_status_t mrcp_client_agent_socket_create(mrcp_v2_agent_t *agent)
{
	apr_status_t status;

	apr_sockaddr_info_get(&agent->sockaddr,"127.0.0.1",APR_INET,0,0,agent->pool);
	if(!agent->sockaddr) {
		return MRCP_STATUS_FAILURE;
	}

	status = apr_socket_create(&agent->listen_sock, agent->sockaddr->family, SOCK_STREAM, APR_PROTO_TCP, agent->pool);
	if(status != APR_SUCCESS) {
		return MRCP_STATUS_FAILURE;
	}

	apr_socket_opt_set(agent->listen_sock, APR_SO_NONBLOCK, 0);
	apr_socket_timeout_set(agent->listen_sock, -1);
	apr_socket_opt_set(agent->listen_sock, APR_SO_REUSEADDR, 1);

	status = apr_socket_bind(agent->listen_sock, agent->sockaddr);
	if(status != APR_SUCCESS) {
		apr_socket_close(agent->listen_sock);
		agent->listen_sock = NULL;
		return MRCP_STATUS_FAILURE;
	}
	apr_socket_addr_get(&agent->sockaddr,APR_LOCAL,agent->listen_sock);

	status = apr_socket_listen(agent->listen_sock, SOMAXCONN);
	if(status != APR_SUCCESS) {
		apr_socket_close(agent->listen_sock);
		agent->listen_sock = NULL;
		return MRCP_STATUS_FAILURE;
	}
	status = apr_pollset_create(&agent->pollset, MRCP_CONNECTION_MAX_COUNT + 3, agent->pool, 0);
	if(status != APR_SUCCESS) {
		apr_socket_close(agent->listen_sock);
		agent->listen_sock = NULL;
		return MRCP_STATUS_FAILURE;
	}
	
	agent->listen_sock_pfd.desc_type = APR_POLL_SOCKET;
	agent->listen_sock_pfd.reqevents = APR_POLLIN;
	agent->listen_sock_pfd.desc.s = agent->listen_sock;
	agent->listen_sock_pfd.client_data = agent->listen_sock;
	status = apr_pollset_add(agent->pollset, &agent->listen_sock_pfd);
	if(status != APR_SUCCESS) {
		apr_socket_close(agent->listen_sock);
		agent->listen_sock = NULL;
		apr_pollset_destroy(agent->pollset);
		agent->pollset = NULL;
		return MRCP_STATUS_FAILURE;
	}
	agent->connection_head = NULL;
	agent->connection_tail = NULL;

	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_socket_destroy(mrcp_v2_agent_t *agent)
{
	mrcp_v2_connection_t *connection = agent->connection_head;
	for(; connection; connection = agent->connection_head) {
		if(connection->sock) {
			apr_pollset_remove(agent->pollset,&connection->sock_pfd);
			apr_socket_close(connection->sock);
		}
		agent->connection_head = connection->next;
		apr_pool_destroy(connection->pool);
	}
	agent->connection_head = NULL;
	agent->connection_tail = NULL;

	if(agent->accept_sock) {
		apr_pollset_remove(agent->pollset,&agent->accept_sock_pfd);
		apr_socket_close(agent->accept_sock);
		agent->accept_sock = NULL;
	}
	if(agent->listen_sock) {
		apr_pollset_remove(agent->pollset,&agent->listen_sock_pfd);
		apr_socket_close(agent->listen_sock);
		agent->listen_sock = NULL;
	}
	if(agent->pollset) {
		apr_pollset_destroy(agent->pollset);
		agent->pollset = NULL;
	}
	return MRCP_STATUS_SUCCESS;
}

static void mrcp_client_agent_signal_event(mrcp_v2_agent_t *agent, mrcp_proto_agent_event_id id, mrcp_v2_connection_t *connection, mrcp_message_t *message)
{
	apt_task_msg_t *task_msg;
	task_msg = apt_producer_task_msg_get(agent->producer_task);
	if(task_msg) {
		mrcp_proto_agent_event_t *agent_event = (mrcp_proto_agent_event_t*)task_msg->data;
		agent_event->id = id;
		agent_event->connection = connection;
		agent_event->message = message;
		agent->proto_agent.module.signal(&agent->proto_agent.module,task_msg);
	}
}

static void mrcp_client_agent_signal_request(mrcp_v2_agent_t *agent, mrcp_proto_agent_signal_id id, mrcp_v2_connection_t *connection, mrcp_message_t *message)
{
	mrcp_proto_agent_signal_t signal;
	apr_size_t size;

	signal.id = id;
	signal.connection = connection;
	signal.message = message;
	size = sizeof(signal);
	if(apr_socket_send(agent->control_sock,(char*)&signal,&size) != APR_SUCCESS) {
		apt_log(APT_PRIO_WARNING,"Failed to Signal Reqest\n");
	}
}

static void mrcp_client_agent_on_start(void *data)
{
	mrcp_v2_agent_t *agent = data;
	mrcp_client_agent_signal_event(agent,PROTO_AGENT_EVENT_STARTED,NULL,NULL);
}

static void mrcp_client_agent_on_terminate_complete(void *data)
{
	mrcp_v2_agent_t *agent = data;
	mrcp_client_agent_signal_event(agent,PROTO_AGENT_EVENT_TERMINATED,NULL,NULL);
}



static mrcp_status_t mrcp_client_agent_do_connect(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection)
{
	apr_sockaddr_t *sockaddr = NULL;

	if(!connection || connection->sock) {
		return MRCP_STATUS_FAILURE;
	}

	apr_sockaddr_info_get(&sockaddr,connection->remote_ip,APR_INET,connection->remote_port,0,connection->pool);
	if(!sockaddr) {
		return MRCP_STATUS_FAILURE;
	}

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

	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, sockaddr) != APR_SUCCESS) {
		apr_socket_close(connection->sock);
		connection->sock = NULL;
		return MRCP_STATUS_FAILURE;
	}

	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(apr_pollset_add(agent->pollset, &connection->sock_pfd) != APR_SUCCESS) {
		apr_socket_close(connection->sock);
		connection->sock = NULL;
		return MRCP_STATUS_FAILURE;
	}
	apt_log(APT_PRIO_NOTICE,"Connected to MRCPv2 Server\n");
	mrcp_client_agent_signal_event(agent,PROTO_AGENT_EVENT_CONNECTED,connection,NULL);
	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_do_disconnect(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection)
{
	if(!connection) {
		return MRCP_STATUS_FAILURE;
	}
	
	if(connection->sock) {
		apr_pollset_remove(agent->pollset, &connection->sock_pfd);
		apr_socket_close(connection->sock);
		connection->sock = NULL;
	}

	apt_log(APT_PRIO_NOTICE,"Disconnected from MRCPv2 Server\n");
	mrcp_client_agent_signal_event(agent,PROTO_AGENT_EVENT_DISCONNECTED,connection,NULL);
	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_do_accept(mrcp_v2_agent_t *agent)
{
	apr_socket_t *sock;
	if(apr_socket_accept(&sock, agent->listen_sock, agent->pool) != APR_SUCCESS) {
		return MRCP_STATUS_FAILURE;
	}

	if(agent->accept_sock) {
		apt_log(APT_PRIO_DEBUG,"Control Peer Rejected\n");
		apr_socket_close(sock);
	}
	else {
		apt_log(APT_PRIO_DEBUG,"Control Peer Connected\n");
		agent->accept_sock = sock;
		agent->accept_sock_pfd.desc_type = APR_POLL_SOCKET;
		agent->accept_sock_pfd.reqevents = APR_POLLIN;
		agent->accept_sock_pfd.desc.s = agent->accept_sock;
		agent->accept_sock_pfd.client_data = agent->accept_sock;

		apr_socket_opt_set(sock, APR_SO_NONBLOCK, 1);
		apr_socket_timeout_set(sock, 0);

		apr_pollset_add(agent->pollset, &agent->accept_sock_pfd);
	}
	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_peer_connect(mrcp_v2_agent_t *agent)
{
	if(agent->control_sock) {
		return MRCP_STATUS_FAILURE;
	}

	if(!agent->sockaddr) {
		return MRCP_STATUS_FAILURE;
	}

	if(apr_socket_create(&agent->control_sock, agent->sockaddr->family, SOCK_STREAM, APR_PROTO_TCP, agent->pool) != APR_SUCCESS) {
		return MRCP_STATUS_FAILURE;
	}

	apr_socket_opt_set(agent->control_sock, APR_SO_NONBLOCK, 0);
	apr_socket_timeout_set(agent->control_sock, -1);
	apr_socket_opt_set(agent->control_sock, APR_SO_REUSEADDR, 1);

	if(apr_socket_connect(agent->control_sock, agent->sockaddr) != APR_SUCCESS) {
		apr_socket_close(agent->control_sock);
		agent->control_sock = NULL;
		return MRCP_STATUS_FAILURE;
	}

	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_peer_disconnect(mrcp_v2_agent_t *agent)
{
	if(!agent->control_sock) {
		return MRCP_STATUS_FAILURE;
	}
	apr_socket_close(agent->control_sock);
	agent->control_sock = NULL;
	return MRCP_STATUS_SUCCESS;
}

static void mrcp_client_agent_on_terminate_request(void *data)
{
	mrcp_v2_agent_t *agent = data;
	if(!agent->control_sock) {
		mrcp_client_agent_peer_connect(agent);
	}

	mrcp_client_agent_peer_disconnect(agent);
}

static mrcp_status_t mrcp_client_agent_do_message_send(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection, mrcp_message_t *mrcp_message)
{
	char buffer[MRCP_MESSAGE_MAX_SIZE];
	apt_text_stream_t text_stream;
	size_t size = sizeof(buffer)-1;
	
	if(!connection || !connection->sock) {
		return MRCP_STATUS_FAILURE;
	}

	text_stream.buffer = buffer;
	text_stream.size = size;
	text_stream.pos = text_stream.buffer;

	mrcp_message->start_line.message_type = MRCP_MESSAGE_TYPE_REQUEST;
	mrcp_message->start_line.version = MRCP_VERSION_2;
	
	if(mrcp_message_generate(agent->proto_agent.resource_container,mrcp_message,&text_stream) != MRCP_STATUS_SUCCESS) {
		apt_log(APT_PRIO_WARNING,"Failed to Generate MRCPv2 Message\n");
		return MRCP_STATUS_FAILURE;
	}
	*text_stream.pos = '\0';
	apt_log(APT_PRIO_DEBUG,"Send MRCPv2 Message size=%lu\n%s\n",text_stream.size,text_stream.buffer);
	apr_socket_send(connection->sock,text_stream.buffer,&text_stream.size);
	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_do_messsage_receive(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection)
{
	char buffer[MRCP_MESSAGE_MAX_SIZE];
	int more_messages_on_buffer = FALSE;
	apr_size_t size = sizeof(buffer)-1;
	apr_status_t status;
	apt_text_stream_t text_stream;
	mrcp_message_t *mrcp_message;

	if(!connection || !connection->sock) {
		return MRCP_STATUS_FAILURE;
	}

	status = apr_socket_recv(connection->sock, buffer, &size);
	if(status == APR_EOF || size == 0) {
		apt_log(APT_PRIO_NOTICE,"MRCPv2 Server Disconnected\n");
		apr_pollset_remove(agent->pollset,&connection->sock_pfd);
		apr_socket_close(connection->sock);
		connection->sock = NULL;

		mrcp_client_agent_signal_event(agent,PROTO_AGENT_EVENT_DISCONNECTED,connection,NULL);
		return MRCP_STATUS_SUCCESS;
	}
	buffer[size] = '\0';

	text_stream.buffer = buffer;
	text_stream.size = size;
	text_stream.pos = buffer;

	apt_log(APT_PRIO_DEBUG,"Receive MRCPv2 Message size=%lu\n%s\n",text_stream.size,text_stream.buffer);
	if(!connection->access_count) {
		return MRCP_STATUS_FAILURE;
	}

	do {
		mrcp_message = mrcp_message_create(connection->pool);
		if(mrcp_message_parse(agent->proto_agent.resource_container,mrcp_message,&text_stream) == MRCP_STATUS_SUCCESS) {
			apt_log(APT_PRIO_DEBUG,"Signal MRCPv2 Message\n");
			mrcp_client_agent_signal_event(agent,PROTO_AGENT_EVENT_RECEIVED,connection,mrcp_message);
		}
		else {
			apt_log(APT_PRIO_WARNING,"Failed to Parse MRCPv2 Message\n");
			if(mrcp_message->start_line.version == MRCP_VERSION_2) {
				/* assume that at least message length field is valid */
				if(mrcp_message->start_line.length <= text_stream.size) {
					/* skip to the end of the message */
					text_stream.pos = text_stream.buffer + mrcp_message->start_line.length;
				}
				else {
					/* skip to the end of the buffer (support incomplete) */
					text_stream.pos = text_stream.buffer + text_stream.size;
				}
			}
		}

		more_messages_on_buffer = FALSE;
		if(text_stream.size > (size_t)(text_stream.pos - text_stream.buffer)) {
			/* there are more MRCPv2 messages to signal */
			more_messages_on_buffer = TRUE;
			text_stream.size -= text_stream.pos - text_stream.buffer;
			text_stream.buffer = text_stream.pos;
			apt_log(APT_PRIO_DEBUG,"Saving remaining buffer for next message...\n");
		}
	}
	while(more_messages_on_buffer);

	return MRCP_STATUS_SUCCESS;
}

static void mrcp_client_agent_main_loop(void *data)
{
	mrcp_v2_agent_t *agent = data;
	apr_status_t status;
	apr_int32_t num;
	const apr_pollfd_t *ret_pfd;
	int i;

	if(!agent || !agent->pollset) {
		apt_log(APT_PRIO_WARNING,"Failed to Start MRCPv2 Agent\n");
		return;
	}

	while(apt_producer_task_is_terminating(agent->producer_task) != TRUE) {
		status = apr_pollset_poll(agent->pollset, -1, &num, &ret_pfd);
		if(status != APR_SUCCESS) {
			continue;
		}
		for(i = 0; i < num; i++) {
			if(ret_pfd[i].desc.s == agent->listen_sock) {
				if(apt_producer_task_is_terminating(agent->producer_task) == TRUE) {
					break;
				}
				mrcp_client_agent_do_accept(agent);
			}
			else if(ret_pfd[i].desc.s == agent->accept_sock) {
				mrcp_proto_agent_signal_t signal;
				size_t size = sizeof(signal);
				status = apr_socket_recv(agent->accept_sock, (char*)&signal, &size);
				if(status == APR_EOF || size == 0) {
					apt_log(APT_PRIO_DEBUG,"Control Peer Disconnected\n");
					apr_pollset_remove(agent->pollset,&agent->accept_sock_pfd);
					apr_socket_close(agent->accept_sock);
					agent->accept_sock = NULL;
					continue;
				}
				switch(signal.id) {
					case PROTO_AGENT_SIGNAL_CONNECT:
						mrcp_client_agent_do_connect(agent,signal.connection);
						break;
					case PROTO_AGENT_SIGNAL_DISCONNECT:
						mrcp_client_agent_do_disconnect(agent,signal.connection);
						break;
					case PROTO_AGENT_SIGNAL_SEND:
						mrcp_client_agent_do_message_send(agent,signal.connection,signal.message);
						break;
				}
			}
			else {
				mrcp_client_agent_do_messsage_receive(agent,ret_pfd[i].client_data);
			}
		}
	}
}


static APR_INLINE mrcp_v2_agent_t* mrcp_v2_agent_get(mrcp_module_t *module)
{
	return ((mrcp_proto_agent_t*)module)->object;
}

static mrcp_status_t mrcp_client_agent_on_disconnect(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection);

static mrcp_status_t mrcp_client_agent_signal_handler(mrcp_module_t *module, apt_task_msg_t *msg)
{
	mrcp_v2_agent_t *agent = mrcp_v2_agent_get(module);
	mrcp_proto_agent_event_t *agent_event = (mrcp_proto_agent_event_t*)msg->data;

	if(!agent || !agent_event) {
		return MRCP_STATUS_FAILURE;
	}

	switch(agent_event->id) {
		case PROTO_AGENT_EVENT_CONNECTED:
			break;
		case PROTO_AGENT_EVENT_DISCONNECTED:
			mrcp_client_agent_on_disconnect(agent,agent_event->connection);
			break;
		case PROTO_AGENT_EVENT_RECEIVED:
		{
			mrcp_connection_t *base = agent_event->connection->base;
			base->event_set->on_receive(base,agent_event->message);
			break;
		}
		case PROTO_AGENT_EVENT_STARTED:
			mrcp_client_agent_peer_connect(agent);
			
			if(module->event_set->on_open) {
				module->event_set->on_open(module);
			}
			break;
		case PROTO_AGENT_EVENT_TERMINATED:
			mrcp_client_agent_socket_destroy(agent);
			
			if(module->event_set->on_close) {
				module->event_set->on_close(module);
			}
			break;
	}

	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_connection_add(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection)
{
	connection->next = NULL;
	connection->prev = agent->connection_tail;
	if(agent->connection_tail) {
		agent->connection_tail->next = connection;
		agent->connection_tail = connection;
	}
	else {
		agent->connection_head = agent->connection_tail = connection;
	}
	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_connection_remove(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection)
{
	if(connection->prev) {
		connection->prev->next = connection->next;
	}
	else {
		agent->connection_head = connection->next;
	}
	if(connection->next) {
		connection->next->prev = connection->prev;
	}
	else {
		agent->connection_tail = connection->prev;
	}
	connection->next = NULL;
	connection->prev = NULL;
	return MRCP_STATUS_SUCCESS;
}

static mrcp_v2_connection_t* mrcp_client_agent_connection_find(mrcp_v2_agent_t *agent, const char *remote_ip, unsigned short remote_port)
{
	mrcp_v2_connection_t *connection = agent->connection_head;
	for(; connection; connection = connection->next) {
		if(connection->remote_port == remote_port && strcmp(connection->remote_ip,remote_ip) == 0) {
			return connection;
		}
	}
	return NULL;
}

static mrcp_connection_t* mrcp_client_agent_connect(mrcp_proto_agent_t *proto_agent, mrcp_signaling_channel_t *channel, const char *remote_ip, unsigned short remote_port)
{
	mrcp_v2_agent_t *agent = proto_agent->object;
	mrcp_v2_connection_t *connection;

	if(!remote_ip) {
		return NULL;
	}

	connection = mrcp_client_agent_connection_find(agent,remote_ip,remote_port);
	if(!connection) {
		mrcp_connection_t *base;
		
		apr_pool_t *pool;
		if(apr_pool_create(&pool,NULL) != APR_SUCCESS) {
			return NULL;
		}

		base = apr_palloc(pool,sizeof(mrcp_connection_t));
		base->agent = proto_agent;
		base->method_set = &connection_method_set;
		base->event_set = proto_agent->connection_event_set;

		connection = apr_palloc(pool,sizeof(mrcp_v2_connection_t));
		connection->pool = pool;
		connection->base = base;
		base->object = connection;
		connection->remote_ip = apr_pstrdup(pool,remote_ip);
		connection->remote_port = remote_port;
		connection->access_count = 0;
		connection->sock = NULL;
		mrcp_client_agent_connection_add(agent,connection);

		apt_log(APT_PRIO_INFO,"MRCPv2 Agent Connect TCP %s:%hu\n",remote_ip,remote_port);
		if(!agent->control_sock) {
			mrcp_client_agent_peer_connect(agent);
		}

		mrcp_client_agent_signal_request(agent,PROTO_AGENT_SIGNAL_CONNECT,connection,NULL);
	}

	connection->access_count++;
	return connection->base;
}

static mrcp_status_t mrcp_client_agent_disconnect(mrcp_connection_t *base)
{
	mrcp_v2_agent_t *agent = base->agent->object;
	mrcp_v2_connection_t *connection = base->object;
	if(!connection || !connection->access_count) {
		return MRCP_STATUS_FAILURE;
	}
	
	connection->access_count--;
	if(!connection->access_count) {
		mrcp_client_agent_connection_remove(agent,connection);

		apt_log(APT_PRIO_INFO,"MRCPv2 Agent Disconnect\n");
		mrcp_client_agent_signal_request(agent,PROTO_AGENT_SIGNAL_DISCONNECT,connection,NULL);
	}

	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_on_disconnect(mrcp_v2_agent_t *agent, mrcp_v2_connection_t *connection)
{
	if(connection->access_count) {
		/* remove from the list and notify client stack */
		mrcp_client_agent_connection_remove(agent,connection);
		if(agent->proto_agent.connection_event_set) {
			agent->proto_agent.connection_event_set->on_disconnect(connection->base);
		}
	}
	else {
		apr_pool_destroy(connection->pool);
	}
	return MRCP_STATUS_SUCCESS;
}

static mrcp_status_t mrcp_client_agent_message_send(mrcp_connection_t *base, mrcp_message_t *mrcp_message)
{
	mrcp_v2_agent_t *agent = base->agent->object;
	mrcp_v2_connection_t *connection = base->object;
	if(!connection) {
		return MRCP_STATUS_FAILURE;
	}

	apt_log(APT_PRIO_DEBUG,"MRCPv2 Send Request\n");
	mrcp_client_agent_signal_request(agent,PROTO_AGENT_SIGNAL_SEND,connection,mrcp_message);
	return MRCP_STATUS_SUCCESS;
}

static mrcp_module_state_t mrcp_client_agent_open(mrcp_module_t *module)
{
	mrcp_v2_agent_t *agent = mrcp_v2_agent_get(module);
	apt_log(APT_PRIO_INFO,"Open MRCPv2 Agent\n");
	if(mrcp_client_agent_socket_create(agent) != MRCP_STATUS_SUCCESS) {
		apt_log(APT_PRIO_WARNING,"Failed to Open MRCPv2 Agent\n");
		return MODULE_STATE_NONE;
	}
	apt_producer_task_start(agent->producer_task);
	return MODULE_STATE_OPEN_INPROGRESS;
}

static mrcp_module_state_t mrcp_client_agent_close(mrcp_module_t *module)
{
	mrcp_v2_agent_t *agent = mrcp_v2_agent_get(module);
	apt_log(APT_PRIO_INFO,"Close MRCPv2 Agent\n");
	apt_producer_task_terminate(agent->producer_task,MRCP_STATUS_FAILURE);
	return MODULE_STATE_CLOSE_INPROGRESS;
}


static mrcp_status_t mrcp_client_agent_destroy(mrcp_module_t *module)
{
	mrcp_v2_agent_t *agent = mrcp_v2_agent_get(module);
	apt_log(APT_PRIO_NOTICE,"Destroy MRCPv2 Agent\n");
	if(agent->producer_task) {
		apt_producer_task_destroy(agent->producer_task);
	}
	agent->pool = NULL;
	return MRCP_STATUS_SUCCESS;
}


mrcp_proto_agent_t* mrcp_client_v2_agent_create(apr_pool_t *pool)
{
	mrcp_v2_agent_t *agent;
	apt_task_msg_pool_t *msg_pool;

	apt_log(APT_PRIO_NOTICE,"Create MRCPv2 Agent\n");
	agent = apr_palloc(pool,sizeof(mrcp_v2_agent_t));
	agent->pool = pool;
	agent->control_sock = NULL;
	agent->listen_sock = NULL;
	agent->accept_sock = NULL;
	agent->pollset = NULL;
	agent->sockaddr = NULL;
	agent->proto_agent.object = agent;
	agent->proto_agent.agent_method_set = &agent_method_set;
	agent->proto_agent.connection_event_set = NULL;

	mrcp_module_init(&agent->proto_agent.module,&module_method_set);

	msg_pool = apt_task_msg_pool_create_waitable_static(sizeof(mrcp_proto_agent_event_t),pool);
	agent->producer_task = apt_producer_task_create(agent,mrcp_client_agent_main_loop,msg_pool,pool);
	if(agent->producer_task) {
		apt_task_t *task = apt_producer_task_get(agent->producer_task);
		apt_task_event_handler_set(task,TASK_STATE_START_IN_PROGRESS,agent,mrcp_client_agent_on_start);
		apt_task_event_handler_set(task,TASK_STATE_TERMINATE_REQUESTED,agent,mrcp_client_agent_on_terminate_request);
		apt_task_event_handler_set(task,TASK_STATE_TERMINATE_COMPLETED,agent,mrcp_client_agent_on_terminate_complete);
	}
	return &agent->proto_agent;
}