/*
 * Copyright 2004,2005 The Apache Software Foundation.
 *
 * 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.
 */
#include <sandesha2_polling_mgr.h>
#include <sandesha2_constants.h>
#include <sandesha2_utils.h>
#include <sandesha2_storage_mgr.h>
#include <sandesha2_terminate_mgr.h>
#include <sandesha2_seq_property_bean.h>
#include <sandesha2_seq_property_mgr.h>
#include <sandesha2_sender_mgr.h>
#include <sandesha2_next_msg_mgr.h>
#include <sandesha2_permanent_seq_property_mgr.h>
#include <sandesha2_permanent_sender_mgr.h>
#include <sandesha2_permanent_next_msg_mgr.h>
#include <sandesha2_msg_ctx.h>
#include <sandesha2_seq.h>
#include <sandesha2_msg_init.h>
#include <sandesha2_msg_creator.h>
#include <sandesha2_sender_bean.h>
#include <axis2_addr.h>
#include <axis2_engine.h>
#include <axutil_uuid_gen.h>
#include <axutil_rand.h>
#include <stdio.h>
#include <platforms/axutil_platform_auto_sense.h>
#include <axutil_types.h>
#include <axiom_soap_const.h>
#include <axis2_http_transport_utils.h>


/** 
 * @brief Polling Manager struct impl
 *	Sandesha2 Polling Manager
 */
typedef struct sandesha2_polling_mgr_args sandesha2_polling_mgr_args_t;

struct sandesha2_polling_mgr_args
{
    axutil_env_t *env;
	axis2_conf_ctx_t *conf_ctx;
    sandesha2_msg_ctx_t *rm_msg_ctx;
    axis2_char_t *internal_sequence_id;
    axis2_char_t *sequence_id;
};
            
static axis2_status_t AXIS2_CALL
sandesha2_polling_mgr_process_make_connection_msg_response(
    const axutil_env_t *env, 
    axis2_msg_ctx_t *msg_ctx,
    sandesha2_storage_mgr_t *storage_mgr);

/**
 * Thread worker function.
 */
static void * AXIS2_THREAD_FUNC
sandesha2_polling_mgr_worker_func(
    axutil_thread_t *thd, 
    void *data);

axis2_status_t AXIS2_CALL 
sandesha2_polling_mgr_start (
    const axutil_env_t *env,
    axis2_conf_ctx_t *conf_ctx,
    sandesha2_storage_mgr_t *storage_mgr,
    sandesha2_sender_mgr_t *sender_mgr,
    sandesha2_msg_ctx_t *rm_msg_ctx,
    const axis2_char_t *internal_sequence_id,
    axis2_char_t *sequence_id,
    const axis2_char_t *reply_to)
{
    axutil_thread_t *worker_thread = NULL;
    sandesha2_polling_mgr_args_t *args = NULL;
    axis2_char_t *wsmc_anon_reply_to_uri = NULL;
    sandesha2_msg_ctx_t *make_conn_rm_msg_ctx = NULL;
    axis2_char_t *make_conn_msg_store_key = NULL;
    axis2_msg_ctx_t *make_conn_msg_ctx = NULL;
    sandesha2_sender_bean_t *make_conn_sender_bean = NULL;
    axis2_status_t status = AXIS2_SUCCESS;
    axis2_engine_t *engine = NULL;
    axiom_soap_envelope_t *res_envelope = NULL;
    axutil_property_t *property = NULL;
    
    args = AXIS2_MALLOC(env->allocator, sizeof(sandesha2_polling_mgr_args_t)); 
    args->env = axutil_init_thread_env(env);
    args->conf_ctx = conf_ctx;
    args->internal_sequence_id = (axis2_char_t *) internal_sequence_id;
    args->sequence_id = (axis2_char_t *) sequence_id;

    if(sandesha2_utils_is_wsrm_anon_reply_to(env, reply_to))
    {
        wsmc_anon_reply_to_uri = axutil_strcat(env, AXIS2_WS_RM_ANONYMOUS_URL, sequence_id, NULL);
    }

    make_conn_rm_msg_ctx = sandesha2_msg_creator_create_make_connection_msg(env, rm_msg_ctx, 
        sequence_id, internal_sequence_id, wsmc_anon_reply_to_uri, NULL);
   
    if(wsmc_anon_reply_to_uri)
    {
        AXIS2_FREE(env->allocator, wsmc_anon_reply_to_uri);
    }

    args->rm_msg_ctx = make_conn_rm_msg_ctx;

    make_conn_msg_ctx = sandesha2_msg_ctx_get_msg_ctx(make_conn_rm_msg_ctx, env);

    property = axutil_property_create_with_args(env, 0, 0, 0, AXIS2_VALUE_TRUE);
    axis2_msg_ctx_set_property(make_conn_msg_ctx, env, SANDESHA2_SEQ_PROP_MAKE_CONNECTION_OUT_PATH, 
            property);

    make_conn_sender_bean = sandesha2_sender_bean_create(env);
    if(make_conn_sender_bean)
    {
        axis2_char_t *msg_id = NULL;
        long millisecs = 0;
        axis2_endpoint_ref_t *to = NULL;

        millisecs = sandesha2_utils_get_current_time_in_millis(env);
        sandesha2_sender_bean_set_time_to_send(make_conn_sender_bean, env, millisecs);
        make_conn_msg_store_key = axutil_uuid_gen(env);
        sandesha2_sender_bean_set_msg_ctx_ref_key(make_conn_sender_bean, env, 
                make_conn_msg_store_key);
        msg_id = sandesha2_msg_ctx_get_msg_id(make_conn_rm_msg_ctx, env);
        sandesha2_sender_bean_set_msg_id(make_conn_sender_bean, env, msg_id);
        sandesha2_sender_bean_set_msg_type(make_conn_sender_bean, env, 
            SANDESHA2_MSG_TYPE_MAKE_CONNECTION_MSG);
        sandesha2_sender_bean_set_resend(make_conn_sender_bean, env, AXIS2_FALSE);
        sandesha2_sender_bean_set_send(make_conn_sender_bean, env, AXIS2_TRUE);
        sandesha2_sender_bean_set_internal_seq_id(make_conn_sender_bean, env, 
                (axis2_char_t *) internal_sequence_id);

        to = sandesha2_msg_ctx_get_to(make_conn_rm_msg_ctx, env);
        if(to)
        {
            axis2_char_t *address = NULL;
            
            address = (axis2_char_t *) axis2_endpoint_ref_get_address(
                    (const axis2_endpoint_ref_t *) to, env);
            sandesha2_sender_bean_set_to_address(make_conn_sender_bean, env, address);
        }
    }
    else
    {
        return AXIS2_FAILURE;
    }

    if(sender_mgr)
    {
        sandesha2_sender_mgr_insert(sender_mgr, env, make_conn_sender_bean);
        sandesha2_sender_bean_free(make_conn_sender_bean, env);
    }
    
    engine = axis2_engine_create(env, conf_ctx);
    status = axis2_engine_send(engine, env, make_conn_msg_ctx);
    if(engine)
    {
        axis2_engine_free(engine, env);
    }

    sandesha2_storage_mgr_store_msg_ctx(storage_mgr, env, make_conn_msg_store_key, make_conn_msg_ctx, 
            AXIS2_TRUE);

    res_envelope = axis2_msg_ctx_get_response_soap_envelope(make_conn_msg_ctx, env);

    if(!res_envelope)
    {
        axis2_char_t *soap_ns_uri = NULL;

        soap_ns_uri = axis2_msg_ctx_get_is_soap_11(make_conn_msg_ctx, env) ?
             AXIOM_SOAP11_SOAP_ENVELOPE_NAMESPACE_URI:
             AXIOM_SOAP12_SOAP_ENVELOPE_NAMESPACE_URI;

        AXIS2_LOG_DEBUG(env->log, AXIS2_LOG_SI, "[sandesha2] Response envelope not found");

        res_envelope = (axiom_soap_envelope_t *) axis2_http_transport_utils_create_soap_msg(env, 
                make_conn_msg_ctx, soap_ns_uri);
    }

    if(res_envelope)
    {
        axis2_msg_ctx_set_response_soap_envelope(make_conn_msg_ctx, env, res_envelope);
        status = sandesha2_polling_mgr_process_make_connection_msg_response(env, make_conn_msg_ctx, 
                storage_mgr);

        if(AXIS2_SUCCESS != status)
        {
            AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, 
                "[sandesha2] Make connection message response process failed for sequence %s", 
                internal_sequence_id);

            return AXIS2_FAILURE;
        }
    }

    worker_thread = axutil_thread_pool_get_thread(env->thread_pool, 
            sandesha2_polling_mgr_worker_func, (void*)args);

    if(!worker_thread)
    {
        AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, 
                "Thread creation failed sandesha2_polling_mgr_run");

        return AXIS2_FAILURE;
    }

    axutil_thread_pool_thread_detach(env->thread_pool, worker_thread); 
        
    return AXIS2_SUCCESS;
}

/**
 * Thread worker function.
 */
static void * AXIS2_THREAD_FUNC
sandesha2_polling_mgr_worker_func(
    axutil_thread_t *thd, 
    void *data)
{
    axis2_char_t *dbname = NULL;
    axis2_char_t *internal_sequence_id = NULL;
    axis2_char_t *sequence_id = NULL;
    axis2_conf_ctx_t *conf_ctx = NULL;
    sandesha2_storage_mgr_t *storage_mgr = NULL;
    sandesha2_seq_property_mgr_t *seq_prop_mgr = NULL;
    sandesha2_sender_mgr_t *sender_mgr = NULL;
    sandesha2_next_msg_mgr_t *next_msg_mgr = NULL;
    axis2_msg_ctx_t *make_conn_msg_ctx = NULL;
    sandesha2_property_bean_t *property_bean = NULL;
    axis2_conf_t *conf = NULL;
    int wait_time = 0;
    axis2_status_t status = AXIS2_FAILURE;
    sandesha2_sender_bean_t *find_sender_bean = NULL;
    sandesha2_sender_bean_t *sender_bean = NULL;
    axis2_char_t *key = NULL;
    
    sandesha2_polling_mgr_args_t *args = (sandesha2_polling_mgr_args_t*)data;
    axutil_env_t *env = args->env;
    conf_ctx = args->conf_ctx;
    internal_sequence_id = axutil_strdup(env, args->internal_sequence_id);
    sequence_id = axutil_strdup(env, args->sequence_id);

    AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, "[sandesha2] Entry:sandesha2_polling_mgr_worker_func");

    dbname = sandesha2_util_get_dbname(env, conf_ctx);
    
    storage_mgr = sandesha2_utils_get_storage_mgr(env, dbname);
    if(!storage_mgr)
    {
        AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, "[sandesha2] Could not create storage manager.");
        AXIS2_ERROR_SET(env->error, SANDESHA2_ERROR_COULD_NOT_CREATE_STORAGE_MANAGER, 
                AXIS2_FAILURE);
        return NULL;
    }
    seq_prop_mgr = sandesha2_permanent_seq_property_mgr_create(env, dbname);
    sender_mgr = sandesha2_permanent_sender_mgr_create(env, dbname);
    next_msg_mgr = sandesha2_permanent_next_msg_mgr_create(env, dbname);
 
    conf = axis2_conf_ctx_get_conf(conf_ctx, env);
    /*property_bean = sandesha2_utils_get_property_bean(env, conf);*/
    wait_time = sandesha2_property_bean_get_polling_delay(property_bean, env);

    find_sender_bean = sandesha2_sender_bean_create(env);
    sandesha2_sender_bean_set_msg_type(find_sender_bean, env, SANDESHA2_MSG_TYPE_MAKE_CONNECTION_MSG);
    sandesha2_sender_bean_set_internal_seq_id(find_sender_bean, env, internal_sequence_id);
    sandesha2_sender_bean_set_send(find_sender_bean, env, AXIS2_TRUE);

    sender_bean = sandesha2_sender_mgr_find_unique(sender_mgr, env, find_sender_bean);
    if(find_sender_bean)
    {
        sandesha2_sender_bean_free(find_sender_bean, env);
    }
    if(sender_bean)
    {
        key = sandesha2_sender_bean_get_msg_ctx_ref_key(sender_bean, env);
    }

    while(AXIS2_TRUE)
    {
        axiom_soap_envelope_t *res_envelope = NULL;
        axis2_char_t *soap_ns_uri = NULL;
        axutil_property_t *property = NULL;

        axis2_transport_out_desc_t *transport_out = NULL;
        axis2_transport_sender_t *transport_sender = NULL;
        axis2_bool_t successfully_sent = AXIS2_FALSE;

        AXIS2_SLEEP(wait_time);
        
        make_conn_msg_ctx = sandesha2_storage_mgr_retrieve_msg_ctx(storage_mgr, env, key, conf_ctx, 
                AXIS2_TRUE);

        property = axutil_property_create_with_args(env, 0, 0, 0, AXIS2_VALUE_TRUE);
        axis2_msg_ctx_set_property(make_conn_msg_ctx, env, SANDESHA2_SEQ_PROP_MAKE_CONNECTION_OUT_PATH, 
                property);

        soap_ns_uri = axis2_msg_ctx_get_is_soap_11(make_conn_msg_ctx, env) ?
             AXIOM_SOAP11_SOAP_ENVELOPE_NAMESPACE_URI:
             AXIOM_SOAP12_SOAP_ENVELOPE_NAMESPACE_URI;

        AXIS2_LOG_DEBUG(env->log, AXIS2_LOG_SI, 
            "[sandesha2] Sending make connection message for sequence with internal sequence id %s", 
            internal_sequence_id);

        transport_out = axis2_msg_ctx_get_transport_out_desc(make_conn_msg_ctx, env);
        if(transport_out)
        {
            transport_sender = axis2_transport_out_desc_get_sender(transport_out, env);
        }
        if(transport_sender)
        {
            /* This is neccessary to avoid a double free at http_sender.c */
            axis2_msg_ctx_set_property(make_conn_msg_ctx, env, AXIS2_TRANSPORT_IN, NULL);
            if(AXIS2_TRANSPORT_SENDER_INVOKE(transport_sender, env, make_conn_msg_ctx))
            {
                successfully_sent = AXIS2_TRUE;
            }else
            {
                successfully_sent = AXIS2_FALSE;
            }
        }

        if(successfully_sent)
        {
            res_envelope = axis2_msg_ctx_get_response_soap_envelope(make_conn_msg_ctx, env);
        }

        if(!res_envelope)
        {
            AXIS2_LOG_DEBUG(env->log, AXIS2_LOG_SI, "[sandesha2] Response envelope not found");

            res_envelope = (axiom_soap_envelope_t *) axis2_http_transport_utils_create_soap_msg(env, 
                    make_conn_msg_ctx, soap_ns_uri);
        }
        
        if(res_envelope)
        {
            axis2_msg_ctx_set_response_soap_envelope(make_conn_msg_ctx, env, res_envelope);
            status = sandesha2_polling_mgr_process_make_connection_msg_response(env, make_conn_msg_ctx, 
                    storage_mgr);

            if(AXIS2_SUCCESS != status)
            {
                AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, 
                    "[sandesha2] Make connection message response process failed for sequence %s", 
                    internal_sequence_id);

                break;
            }
        }
    }

    if(seq_prop_mgr)
    {
        sandesha2_seq_property_mgr_free(seq_prop_mgr, env);
    }

    if(sender_mgr)
    {
        sandesha2_sender_mgr_free(sender_mgr, env);
    }

    if(next_msg_mgr)
    {
        sandesha2_next_msg_mgr_free(next_msg_mgr, env);
    }

    AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, "[sandesha2] Exit:sandesha2_polling_mgr_worker_func");

    return NULL;
}

static axis2_status_t AXIS2_CALL
sandesha2_polling_mgr_process_make_connection_msg_response(
    const axutil_env_t *env, 
    axis2_msg_ctx_t *msg_ctx,
    sandesha2_storage_mgr_t *storage_mgr)
{
    axis2_char_t *soap_ns_uri = NULL;
    axis2_msg_ctx_t *response_msg_ctx = NULL;
    axiom_soap_envelope_t *response_envelope = NULL;
    axis2_conf_ctx_t *conf_ctx = NULL;
    axis2_engine_t *engine = NULL;
    axis2_status_t status = AXIS2_FAILURE;
    axis2_endpoint_ref_t *to = NULL;
   
    AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI,
        "[sandesha2] Entry:sandesha2_polling_mgr_process_make_connection_msg_response");

    AXIS2_PARAM_CHECK(env->error, msg_ctx, AXIS2_FAILURE);

    conf_ctx = axis2_msg_ctx_get_conf_ctx(msg_ctx, env);

    soap_ns_uri = axis2_msg_ctx_get_is_soap_11(msg_ctx, env) ?
         AXIOM_SOAP11_SOAP_ENVELOPE_NAMESPACE_URI:
         AXIOM_SOAP12_SOAP_ENVELOPE_NAMESPACE_URI;

    response_envelope = axis2_msg_ctx_get_response_soap_envelope(msg_ctx, env);
    if(!response_envelope)
    {
        response_envelope = (axiom_soap_envelope_t *) axis2_http_transport_utils_create_soap_msg(env, 
                msg_ctx, soap_ns_uri);
        if(!response_envelope)
        {
            /* There is no response message context. */

            AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, "[sandesha2] Response envelope not found");

            return AXIS2_SUCCESS;
        }
    }

    AXIS2_LOG_DEBUG(env->log, AXIS2_LOG_SI, 
        "[sandesha2] Response envelope for make connection message found");

    response_msg_ctx = axis2_msg_ctx_create(env, conf_ctx, axis2_msg_ctx_get_transport_in_desc(msg_ctx, 
                env), axis2_msg_ctx_get_transport_out_desc(msg_ctx, env));
   
    to = axis2_endpoint_ref_create(env, 
        "http://localhost/axis2/services/__ANONYMOUS_SERVICE__/__OPERATION_OUT_IN__");
    axis2_msg_ctx_set_to(response_msg_ctx, env, to);

    axis2_msg_ctx_set_wsa_action(response_msg_ctx, env, 
            "http://localhost/axis2/services/__ANONYMOUS_SERVICE__/__OPERATION_OUT_IN__");

    axis2_msg_ctx_set_soap_envelope(response_msg_ctx, env, response_envelope);

    /*axis2_msg_ctx_set_server_side(response_msg_ctx, env, AXIS2_TRUE);*/

    axis2_msg_ctx_set_op_ctx(response_msg_ctx, env, axis2_msg_ctx_get_op_ctx(msg_ctx, env));
    axis2_msg_ctx_set_svc_ctx(response_msg_ctx, env, axis2_msg_ctx_get_svc_ctx(msg_ctx, env));
    axis2_msg_ctx_set_svc_grp_ctx(response_msg_ctx, env, axis2_msg_ctx_get_svc_grp_ctx(msg_ctx, env));
    axis2_msg_ctx_set_conf_ctx(response_msg_ctx, env, conf_ctx);


    engine = axis2_engine_create(env, conf_ctx);
    if(engine)
    {
        if(sandesha2_util_is_fault_envelope(env, response_envelope))
        {
            status = axis2_engine_receive_fault(engine, env, response_msg_ctx);
        }
        else
        {
            status = axis2_engine_receive(engine, env, response_msg_ctx);
        }
            axis2_engine_free(engine, env);
    }

    axis2_msg_ctx_set_paused(response_msg_ctx, env, AXIS2_FALSE);
    axis2_msg_ctx_free(response_msg_ctx, env);

    AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI,
        "[sandesha2] Exit:sandesha2_polling_mgr_process_make_connection_msg_response");

    return AXIS2_SUCCESS;
}