/*
 * 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):
 *
 */

#include <assert.h>
#include <stdlib.h>
#include <apr_thread_proc.h>
#include <apr_thread_cond.h>
#include "apt_task_msg.h"

/** Abstract pool of task messages to allocate task messages from */
struct apt_task_msg_pool_t {
	void (*destroy)(apt_task_msg_pool_t *pool);

	apt_task_msg_t* (*acquire_msg)(apt_task_msg_pool_t *pool);
	void (*wait_for_msg)(apt_task_msg_t *task_msg);
	void (*release_msg)(apt_task_msg_t *task_msg);

	void       *object;
	apr_pool_t *pool;
};


/** Dynamic allocation of messages (no actual pool exist)*/
typedef struct apt_msg_pool_dynamic_t apt_msg_pool_dynamic_t;

struct apt_msg_pool_dynamic_t {
	apr_size_t size;
};

static apt_task_msg_t* dynamic_pool_acquire_msg(apt_task_msg_pool_t *task_msg_pool)
{
	apt_msg_pool_dynamic_t *dynamic_pool = task_msg_pool->object;
	apt_task_msg_t *task_msg = malloc(dynamic_pool->size);
	task_msg->msg_pool = task_msg_pool;
	return task_msg;
}

static void dynamic_pool_release_msg(apt_task_msg_t *task_msg)
{
	if(task_msg) {
		free(task_msg);
	}
}

static void dynamic_pool_destroy(apt_task_msg_pool_t *task_msg_pool)
{
	/* nothing to do */
}

APT_DECLARE(apt_task_msg_pool_t*) apt_task_msg_pool_create_dynamic(apr_size_t msg_size, apr_pool_t *pool)
{
	apt_task_msg_pool_t *task_msg_pool = apr_palloc(pool,sizeof(apt_task_msg_pool_t));
	apt_msg_pool_dynamic_t *dynamic_pool = apr_palloc(pool,sizeof(apt_msg_pool_dynamic_t));
	dynamic_pool->size = msg_size + sizeof(apt_task_msg_t) - 1;

	task_msg_pool->pool = pool;
	task_msg_pool->object = dynamic_pool;
	task_msg_pool->acquire_msg = dynamic_pool_acquire_msg;
	task_msg_pool->wait_for_msg = NULL;
	task_msg_pool->release_msg = dynamic_pool_release_msg;
	task_msg_pool->destroy = dynamic_pool_destroy;
	return task_msg_pool;
}


/** Static allocation of a single message, which forces producer to wait for message release by consumer (no actual pool exist)*/
typedef struct apt_msg_pool_waitable_static_t apt_msg_pool_waitable_static_t;

struct apt_msg_pool_waitable_static_t {
	apt_task_msg_t     *task_msg;

	apr_thread_cond_t  *wait_object;
	apr_thread_mutex_t *wait_object_mutex;
};

static apt_task_msg_t* waitable_static_pool_acquire_msg(apt_task_msg_pool_t *task_msg_pool)
{
	apt_msg_pool_waitable_static_t *waitable_static_pool = task_msg_pool->object;
	apr_thread_mutex_lock(waitable_static_pool->wait_object_mutex);
	return waitable_static_pool->task_msg;
}

static void waitable_static_pool_wait_for_msg(apt_task_msg_t *task_msg)
{
	apt_msg_pool_waitable_static_t *waitable_static_pool = task_msg->msg_pool->object;
	apr_thread_cond_wait(waitable_static_pool->wait_object,waitable_static_pool->wait_object_mutex);
	apr_thread_mutex_unlock(waitable_static_pool->wait_object_mutex);
}

static void waitable_static_pool_release_msg(apt_task_msg_t *task_msg)
{
	apt_msg_pool_waitable_static_t *waitable_static_pool = task_msg->msg_pool->object;
	apr_thread_mutex_lock(waitable_static_pool->wait_object_mutex);
	apr_thread_cond_signal(waitable_static_pool->wait_object);
	apr_thread_mutex_unlock(waitable_static_pool->wait_object_mutex);
}

static void waitable_static_pool_destroy(apt_task_msg_pool_t *task_msg_pool)
{
	apt_msg_pool_waitable_static_t *waitable_static_pool = task_msg_pool->object;
	if(waitable_static_pool) {
		apr_thread_cond_destroy(waitable_static_pool->wait_object);
		apr_thread_mutex_destroy(waitable_static_pool->wait_object_mutex);
	}
}

APT_DECLARE(apt_task_msg_pool_t*) apt_task_msg_pool_create_waitable_static(apr_size_t msg_size, apr_pool_t *pool)
{
	apt_task_msg_pool_t *task_msg_pool = apr_palloc(pool,sizeof(apt_task_msg_pool_t));
	apt_msg_pool_waitable_static_t *waitable_static_pool = apr_palloc(pool,sizeof(apt_msg_pool_waitable_static_t));
	waitable_static_pool->task_msg = apr_palloc(pool,msg_size + sizeof(apt_task_msg_t) - 1);
	waitable_static_pool->task_msg->msg_pool = task_msg_pool;
	apr_thread_mutex_create(&waitable_static_pool->wait_object_mutex,APR_THREAD_MUTEX_UNNESTED,pool);
	apr_thread_cond_create(&waitable_static_pool->wait_object,pool);

	task_msg_pool->pool = pool;
	task_msg_pool->object = waitable_static_pool;
	task_msg_pool->acquire_msg = waitable_static_pool_acquire_msg;
	task_msg_pool->wait_for_msg = waitable_static_pool_wait_for_msg;
	task_msg_pool->release_msg = waitable_static_pool_release_msg;
	task_msg_pool->destroy = waitable_static_pool_destroy;

	return task_msg_pool;
}



/** Static allocation of messages from message pool (not implemented yet) */
APT_DECLARE(apt_task_msg_pool_t*) apt_task_msg_pool_create_static(apr_size_t msg_size, apr_size_t pool_size, apr_pool_t *pool)
{
	return NULL;
}



APT_DECLARE(void) apt_task_msg_pool_destroy(apt_task_msg_pool_t *msg_pool)
{
	if(msg_pool->destroy) {
		msg_pool->destroy(msg_pool);
	}
}

APT_DECLARE(apt_task_msg_t*) apt_task_msg_acquire(apt_task_msg_pool_t *task_msg_pool)
{
	if(!task_msg_pool->acquire_msg)
		return NULL;
	return task_msg_pool->acquire_msg(task_msg_pool);
}

APT_DECLARE(apt_bool_t) apt_task_msg_is_blocking(apt_task_msg_t *task_msg)
{
	apt_task_msg_pool_t *task_msg_pool = task_msg->msg_pool;
	return (task_msg_pool->wait_for_msg ? TRUE : FALSE);
}

APT_DECLARE(void) apt_task_msg_wait(apt_task_msg_t *task_msg)
{
	apt_task_msg_pool_t *task_msg_pool = task_msg->msg_pool;
	if(task_msg_pool->wait_for_msg)
		task_msg_pool->wait_for_msg(task_msg);
}

APT_DECLARE(void) apt_task_msg_release(apt_task_msg_t *task_msg)
{
	apt_task_msg_pool_t *task_msg_pool = task_msg->msg_pool;
	if(task_msg_pool->release_msg)
		task_msg_pool->release_msg(task_msg);
}