/*
 * 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_cond.h>
#include "media_processor.h"
#include "media_timer.h"
#include "codec_manager.h"
#include "media_context.h"


typedef enum {
	MEDIA_REQUEST_ADD_CONTEXT,
	MEDIA_REQUEST_REMOVE_CONTEXT,

	MEDIA_REQUEST_ADD_SOURCE,
	MEDIA_REQUEST_REMOVE_SOURCE,

	MEDIA_REQUEST_ADD_SINK,
	MEDIA_REQUEST_REMOVE_SINK
} media_request_id;

typedef struct media_request_msg_t media_request_msg_t;

struct media_request_msg_t {
	media_request_id     id;
	rtp_session_t       *rtp_session;

	media_context_t     *context;
	audio_source_t      *source;
	audio_sink_t        *sink;

	media_request_type   type;

	media_request_msg_t *next;
};

struct media_processor_t {
	apr_pool_t          *pool;
	apr_thread_mutex_t  *guard;
	apr_thread_cond_t   *wait_object;

	media_timer_t       *timer;
	codec_manager_t     *codec_manager;

	media_request_msg_t *request_queue;

	media_context_t     *context_head;
	media_context_t     *context_tail;
};


static media_request_msg_t* media_processor_request_create(media_request_id id, media_request_type type, apr_pool_t *pool);
static apr_status_t media_processor_request_send(media_processor_t *processor, media_request_msg_t *request);

static void media_processor_main(media_timer_t *timer, void *data);

apr_status_t g711u_codec_init(codec_manager_t *codec_manager);
apr_status_t g711a_codec_init(codec_manager_t *codec_manager);


media_processor_t* media_processor_create(apr_pool_t *pool)
{
	media_processor_t *processor = apr_palloc(pool,sizeof(media_processor_t));
	processor->pool = pool;
	processor->request_queue = NULL;
	processor->context_head = processor->context_tail = NULL;
	processor->timer = NULL;

	processor->codec_manager = codec_manager_create(pool);
	if(processor->codec_manager) {
		g711u_codec_init(processor->codec_manager);
		g711a_codec_init(processor->codec_manager);
	}
	return processor;
}

apr_status_t media_processor_start(media_processor_t *processor)
{
	apr_thread_mutex_create(&processor->guard,APR_THREAD_MUTEX_UNNESTED,processor->pool);
	apr_thread_cond_create(&processor->wait_object, processor->pool);
	processor->timer = media_timer_start(CODEC_FRAME_TIME_BASE,media_processor_main,processor,processor->pool);
	return APR_SUCCESS;
}

apr_status_t media_processor_terminate(media_processor_t *processor)
{
	media_timer_stop(processor->timer);
	apr_thread_cond_destroy(processor->wait_object);
	apr_thread_mutex_destroy(processor->guard);
	processor->guard = NULL;
	return APR_SUCCESS;
}

apr_status_t media_processor_destroy(media_processor_t *processor)
{
	if(processor->codec_manager) {
		codec_manager_destroy(processor->codec_manager);
		processor->codec_manager = NULL;
	}
	return APR_SUCCESS;
}

apr_status_t media_processor_context_add(media_processor_t *processor, media_context_t *context, media_request_type type)
{
	media_request_msg_t *request = media_processor_request_create(MEDIA_REQUEST_ADD_CONTEXT,type,context->pool);
	request->context = context;
	return media_processor_request_send(processor,request);
}

apr_status_t media_processor_context_remove(media_processor_t *processor, media_context_t *context, media_request_type type)
{
	media_request_msg_t *request = media_processor_request_create(MEDIA_REQUEST_REMOVE_CONTEXT,type,context->pool);
	request->context = context;
	return media_processor_request_send(processor,request);
}

apr_status_t media_context_source_add(media_processor_t *processor, media_context_t *context, audio_source_t *source, media_request_type type)
{
	media_request_msg_t *request = media_processor_request_create(MEDIA_REQUEST_ADD_SOURCE,type,context->pool);
	request->context = context;
	request->source = source;
	return media_processor_request_send(processor,request);
}

apr_status_t media_context_source_remove(media_processor_t *processor, media_context_t *context, audio_source_t *source, media_request_type type)
{
	media_request_msg_t *request = media_processor_request_create(MEDIA_REQUEST_REMOVE_SOURCE,type,context->pool);
	request->context = context;
	request->source = source;
	return media_processor_request_send(processor,request);
}

apr_status_t media_context_sink_add(media_processor_t *processor, media_context_t *context, audio_sink_t *sink, media_request_type type)
{
	media_request_msg_t *request = media_processor_request_create(MEDIA_REQUEST_ADD_SINK,type,context->pool);
	request->context = context;
	request->sink = sink;
	return media_processor_request_send(processor,request);
}

apr_status_t media_context_sink_remove(media_processor_t *processor, media_context_t *context, audio_sink_t *sink, media_request_type type)
{
	media_request_msg_t *request = media_processor_request_create(MEDIA_REQUEST_REMOVE_SINK,type,context->pool);
	request->context = context;
	request->sink = sink;
	return media_processor_request_send(processor,request);
}

apr_status_t media_context_destroy(media_context_t *context)
{
	if(context && context->method_set->destroy) {
		context->method_set->destroy(context);
	}
	return APR_SUCCESS;
}

codec_manager_t* media_processor_codec_manager_get(media_processor_t *processor)
{
	return processor->codec_manager;
}



static apr_status_t media_processor_do_context_add(media_processor_t *processor, media_context_t *context)
{
	if(!context) {
		return APR_INCOMPLETE;
	}

	context->next = NULL;
	context->prev = processor->context_tail;
	if(processor->context_tail) {
		processor->context_tail->next = context;
		processor->context_tail = context;
	}
	else {
		processor->context_head = processor->context_tail = context;
	}

	return APR_SUCCESS;
}

static apr_status_t media_processor_do_context_remove(media_processor_t *processor, media_context_t *context)
{
	if(!context) {
		return APR_INCOMPLETE;
	}

	if(context->prev) {
		context->prev->next = context->next;
	}
	else {
		processor->context_head = context->next;
	}
	if(context->next) {
		context->next->prev = context->prev;
	}
	else {
		processor->context_tail = context->prev;
	}
	context->next = NULL;
	context->prev = NULL;

	return APR_SUCCESS;
}

static apr_status_t request_queue_process(media_processor_t *processor)
{
	apr_thread_mutex_lock(processor->guard);
		
	while(processor->request_queue) {
		media_request_msg_t *request;
		request = processor->request_queue;
		switch(request->id) {
			case MEDIA_REQUEST_ADD_CONTEXT:
				media_processor_do_context_add(processor,request->context);
				break;
			case MEDIA_REQUEST_REMOVE_CONTEXT:
				media_processor_do_context_remove(processor,request->context);
				break;
			case MEDIA_REQUEST_ADD_SOURCE:
				if(request->context->method_set->audio_source_add) {
					request->context->method_set->audio_source_add(request->context,request->source);
				}
				break;
			case MEDIA_REQUEST_REMOVE_SOURCE:
				if(request->context->method_set->audio_source_remove) {
					request->context->method_set->audio_source_remove(request->context,request->source);
				}
				break;
			case MEDIA_REQUEST_ADD_SINK:
				if(request->context->method_set->audio_sink_add) {
					request->context->method_set->audio_sink_add(request->context,request->sink);
				}
				break;
			case MEDIA_REQUEST_REMOVE_SINK:
				if(request->context->method_set->audio_sink_remove) {
					request->context->method_set->audio_sink_remove(request->context,request->sink);
				}
				break;
		}
		if(request->type == MEDIA_REQUEST_TYPE_SYNC) {
			apr_thread_cond_signal(processor->wait_object);
		}
		processor->request_queue = request->next;
	}

	apr_thread_mutex_unlock(processor->guard);
	return APR_SUCCESS;
}

static media_request_msg_t* media_processor_request_create(media_request_id id, media_request_type type, apr_pool_t *pool)
{
	media_request_msg_t *request = apr_palloc(pool,sizeof(media_request_msg_t));
	request->id = id;
	request->type = type;
	request->next = NULL;
	return request;
}

static apr_status_t media_processor_request_send(media_processor_t *processor, media_request_msg_t *request)
{
	apr_thread_mutex_lock(processor->guard);
	if(processor->request_queue) {
		media_request_msg_t *tail = processor->request_queue;
		while(tail->next) {
			tail = tail->next;
		}
		tail->next = request;
	}
	else {
		processor->request_queue = request;
	}

	if(request->type == MEDIA_REQUEST_TYPE_SYNC) {
		apr_thread_cond_wait(processor->wait_object,processor->guard);
	}
	apr_thread_mutex_unlock(processor->guard);
	return APR_SUCCESS;
}

static void media_processor_main(media_timer_t *timer, void *data)
{
	media_processor_t *processor = data;
	media_context_t *context;

	request_queue_process(processor);

	context = processor->context_head;
	while(context) {
		context->method_set->process(context);

		context = context->next;
	}
}