/*
 * 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 <apr_thread_proc.h>
#include <apr_queue.h>
#include "apt_consumer_task.h"

struct apt_consumer_task_t {
	apr_queue_t         *msg_queue;
	apt_task_t          *task;

	void                *object;
	apt_task_msg_handler msg_handler;
	apt_bool_t           running;
};



static void apt_consumer_task_run(void *data);
static void apt_consumer_task_on_terminate_requested(void *data);

APT_DECLARE(apt_consumer_task_t*) apt_consumer_task_create(void *object, apt_task_msg_handler msg_handler, apr_pool_t *pool)
{
	apt_consumer_task_t *consumer_task = apr_palloc(pool,sizeof(apt_consumer_task_t));
	consumer_task->object = object;
	consumer_task->msg_handler = msg_handler;
	consumer_task->running = FALSE;
	consumer_task->msg_queue = NULL;
	if(apr_queue_create(&consumer_task->msg_queue,1024,pool) != APR_SUCCESS) {
		return NULL;
	}

	consumer_task->task = apt_task_create(consumer_task,apt_consumer_task_run,pool);
	apt_task_event_handler_set(consumer_task->task,TASK_STATE_TERMINATE_REQUESTED,consumer_task,apt_consumer_task_on_terminate_requested);
	return consumer_task;
}

APT_DECLARE(apt_bool_t) apt_consumer_task_destroy(apt_consumer_task_t *consumer_task)
{
	return apt_task_destroy(consumer_task->task);
}

APT_DECLARE(apt_bool_t) apt_consumer_task_start(apt_consumer_task_t *consumer_task)
{
	return apt_task_start(consumer_task->task);
}

APT_DECLARE(apt_bool_t) apt_consumer_task_terminate(apt_consumer_task_t *consumer_task)
{
	return apt_task_terminate(consumer_task->task,TRUE);
}

APT_DECLARE(apt_task_t*) apt_consumer_task_get(apt_consumer_task_t *consumer_task)
{
	return consumer_task->task;
}

APT_DECLARE(apt_bool_t) apt_consumer_task_signal(apt_consumer_task_t *consumer_task, apt_task_msg_t *task_msg)
{
	apt_bool_t wait_for_msg = apt_task_msg_is_blocking(task_msg);
	if(apr_queue_push(consumer_task->msg_queue,task_msg) != APR_SUCCESS) {
		return FALSE;
	}
	
	if(wait_for_msg == TRUE) {
		apt_task_msg_wait(task_msg);
	}
	return TRUE;
}

APT_DECLARE(void) apt_consumer_task_running_set(apt_consumer_task_t *consumer_task, apt_bool_t running)
{
	consumer_task->running = running;
}

static APR_INLINE apt_bool_t apt_consumer_task_is_terminating(apt_consumer_task_t *consumer_task)
{
	if(apt_task_state_get(consumer_task->task) == TASK_STATE_TERMINATE_IN_PROGRESS && 
		consumer_task->running == FALSE) {
		return TRUE;
	}
	return FALSE;
}

static void apt_consumer_task_run(void *data)
{
	apr_status_t rv;
	void *msg;
	apt_consumer_task_t *consumer_task = data;
	if(!consumer_task) {
		return;
	}
	while(apt_consumer_task_is_terminating(consumer_task) != TRUE) {
		rv = apr_queue_pop(consumer_task->msg_queue,&msg);
		if(rv == APR_SUCCESS) {
			if(msg) {
				apt_task_msg_t *task_msg = msg;
				consumer_task->msg_handler(consumer_task->object,task_msg);
				apt_task_msg_release(task_msg);
			}
		}
		if(apt_task_state_get(consumer_task->task) == TASK_STATE_TERMINATE_REQUESTED) {
			apt_task_state_set(consumer_task->task,TASK_STATE_TERMINATE_IN_PROGRESS);
		}
	}
}

static void apt_consumer_task_on_terminate_requested(void *data)
{
	apt_consumer_task_t *consumer_task = data;
	if(consumer_task->msg_queue) {
		apr_queue_interrupt_all(consumer_task->msg_queue);
	}
}