// 	Copyright 11/07/01 Sun Microsystems, Inc. All Rights Reserved.

#pragma ident  "@(#)event_consumer.cc	1.21 01/11/07 Sun Microsystems"

#include <sys/types.h>
#include <em_c++utils/auto_ptr.hh>
#include <em_c++utils/dynamic_output_string_stream.hh>
#include <em_c++utils/em_exception.hh>
#include <cgw_utils/cgw_global.hh>
#include "egw_global.hh"
#include <cgw_utils/cgw_utils.hh>
#include "event_consumer.hh"
#include <auth_helper/auth_server_handle.hh>
#include <cppiac/idl_value.hh>
#include <acf/asn1_converter_fw.hh>
#ifdef ORBACUS
#include <jidm_ext/CMIExt.h>
#else
#ifdef ORBIX
#include <jidm_ext/CMIExt.hh>
#else
#include <jidm_ext/CMIExt_c.hh>
#endif
#endif

#include <pmi/hi.hh>


static unsigned rw_string_hash(const RWCString& s) 
{
    return RWCString::hash(s);
}

TSPtrHashTable<RWCString, EventConsumer>
    EventConsumer::consumer_list_(rw_string_hash);

/******************************************************************************/
EventConsumer* const
EventConsumer::get_consumer(
    const RWCString& ae_title
) throw (
    JIDM::NoEventPort,
    JIDM::InvalidCriteria,
    GeneralPMIException,
    CORBA::SystemException,
    SEMTHRInvalidPointer
)
{
    EventConsumer* ec_p = consumer_list_.find_value(&ae_title);
    if (ec_p != NULL) {
	return ec_p;
    }

    egw_debug.print(
	"Contacting EventPortRegistry to get consumer %s's EventPort.\n",
	ae_title.data());
    
    // Check with EventPortRegistry if an EventPort with this AETitle exists. If
    // so, add that EventPort to our consumer_list_

    JIDM::EventPort_var ep =
	EGWGlobal::epr_->find_event_port_by_ae_title(ae_title);

    if (CORBA::is_nil(ep)) {
	throw JIDM::NoEventPort();
    }

    ec_p = new EventConsumer(ae_title, ep);
    consumer_list_.insert_key_and_value(&(ec_p->ae_title_), ec_p);

    egw_debug.print("Updating local cache with info about consumer %s, whose"
		    " user_id is:",
		    ae_title.data());
    ec_p->user_id_.print(egw_debug);

    return ec_p;
}

void EventConsumer::remove_consumer(
    const RWCString& ae_title
)throw()
{
    EventConsumer* ec_p = consumer_list_.find_value(&ae_title);
    if (ec_p != NULL) {
	delete ec_p;
    }

}

/******************************************************************************/
void
EventConsumer::disconnect_all_consumers() throw()
{
    RWTPtrHashMap<RWCString, EventConsumer RWDefHArgs(RWCString)> hash_table =
	consumer_list_.get_hash_dictionary();
    RWTPtrHashMapIterator<RWCString, EventConsumer RWDefHArgs(RWCString)> iterator(hash_table);
    while (++iterator) {
	EventConsumer* consumer = iterator.value();
	consumer->disconnect_push_supplier();
    }
}

/******************************************************************************/
EventConsumer::EventConsumer(
    const RWCString& ae_title,
    const JIDM::EventPort_var& event_port
) throw(
    JIDM::InvalidCriteria,
    CORBA::SystemException,
    GeneralPMIException,
    SEMTHRInvalidPointer
)
    : ae_title_(ae_title),
      want_text_events_(true),
      event_port_(event_port),
      event_scheduler_(CGWGlobal::dispatch_thread_pool_),
      cleanup_started(false)
{
    /*
    ** First, check if the consumer wants text events or IDL 'any' events.
    ** The default is that consumer wants text events. Hence, the criteria is 
    ** wrong for the EventInfoFormat or if it's missing, by default, text format
    ** is used for events.
    */
    JIDM::Criteria_var criteria = event_port_->associated_criteria();
    CORBA::ULong criteria_len = criteria->length();
#if defined(ORBACUS)  || defined(ORBIX)
    const JIDM::Criteria  * ptr_access_criteria;
    const char* event_info_format = NULL;
#else
    char* event_info_format = NULL;
    JIDM::Criteria access_criteria;
#endif
    
    for (CORBA::ULong i = 0; i < criteria_len; ++i) {
	if (CGWUtils::equal(
	    criteria[i].name, "ProxyAgent Access Criteria"))
	{
#if defined(ORBACUS) || defined (ORBIX)
	    if (criteria[i].value >>= ptr_access_criteria) 
	    	user_id_ = extract_user_id(*ptr_access_criteria);
#else
	    if (criteria[i].value >>= access_criteria) 
	    	user_id_ = extract_user_id(access_criteria);
#endif
	    else
	   {
		CGWUtils::print_any(
		    egw_error,
		    "InvalidCriteria: Value for ProxyAgent Access Criteria"
		    " is not of JIDM::Criteria type. "
		    "Instead, it is as below:",
		    criteria[i].value);
		
		throw JIDM::InvalidCriteria();
	    }
	}	    
	else if (CGWUtils::equal(
	    (*criteria)[i].name, EGWGlobal::EVENT_INFO_FORMAT))
	{
	    if (!((*criteria)[i].value >>= event_info_format)) {
		CGWUtils::print_any(
		    egw_error,
		    "InvalidCriteria: Value for EventInfoFormat:"
		    " criteria is not of string type:\n\tInstead, it is as "
		    "below:",
		    (*criteria)[i].value);
	    }
	    else if (CGWUtils::equal(event_info_format, "CORBA::Any")) {
		want_text_events_ = false;
	    }
#ifndef ORBACUS  
#ifndef ORBIX
	if (event_info_format!=NULL)
		delete []event_info_format;
#endif
#endif
	}
    }
#if defined(ORBACUS) || defined(ORBIX)
    if (ptr_access_criteria->length() == 0) {
#else
    if (access_criteria.length() == 0) {
#endif
	egw_error.print(
	    "InvalidCriteria: Criteria does not contain "
	    "ProxyAgent Access Criteria\n");
	throw JIDM::InvalidCriteria();
    }

    // Next, connect to the consumer
    CosEventChannelAdmin::SupplierAdmin_var supplier_admin =
	event_port_->supplier_admin();

    if (CORBA::is_nil(supplier_admin)) {
	egw_error.print(
	    "Invalid supplier_admin passed to create EventPort %s\n",
	    ae_title.data());
	throw SEMTHRInvalidPointer("EventConsumer (Invalid supplier_admin):");
    }
    consumer_ = supplier_admin->obtain_push_consumer();
    if (CORBA::is_nil(consumer_)) {
	egw_error.print("Unable to obtain ProxyPushConsumer for EventPort %s\n",
			ae_title.data());
	throw SEMTHRInvalidPointer(
		"EventConsumer (Unable to obtain ProxyPushConsumer) :");
    }
    CosEventComm::PushSupplier_var tptr= _this();
    consumer_->connect_push_supplier(tptr);
}

/******************************************************************************/
EventConsumer::~EventConsumer() throw()
{
    cleanup();
}

/******************************************************************************/
void
EventConsumer::disconnect_push_supplier() throw (CORBA::SystemException)
{
    cleanup();
}

/******************************************************************************/
void
EventConsumer::push(
    const CORBA::Any& any_event
) throw()
{
	 CorbaEvent *tmpEvent=NULL;
    try {
// event_q_.write_w(any_event);
		tmpEvent= CorbaEvent::mkCorbaEvent(consumer_,  ae_title_, any_event);
	event_scheduler_->run_task_in_thread( tmpEvent);
	return;
    }
    catch (const SEMTHRPoolException &e)
    {
		egw_error.print(
	    "Consumer '%s' push failed: Local event cache closed.\n",
	    ae_title_.data());
		if (tmpEvent!=NULL)
			delete tmpEvent;
		return;
    }
    catch (const SEMTHRQueueClosedException& e) {
	egw_error.print(
	    "Consumer '%s' push failed: Local event cache closed.\n",
	    ae_title_.data());
		if (tmpEvent!=NULL)
			delete tmpEvent;
	return;
    }
    catch (const SEMTHRInvalidPointer& e) {
	egw_error.print("Consumer '%s' push failed: %s", ae_title_.data(),
			e.what());
		if (tmpEvent!=NULL)
			delete tmpEvent;
	return;
    }
    catch (... ) {
		egw_error.print(
	    "Consumer '%s' push failed: Local event cache closed.\n",
	    ae_title_.data());
		if (tmpEvent!=NULL)
			delete tmpEvent;
	}
}

/******************************************************************************/
/* This pushes a pmi event to corba client -- assumes that the calls are sequential */
void
EventConsumer::push(
	EventReq& event_msg,
	bool convert
) throw()
{
	static CORBA::Any any_event; // event_info format is CORBA::Any
        static CORBA::Any text_event; // event_info format is CORBA::string
	static bool converted_any=false;
	static bool converted_text=false;
	try{
	if (convert)
        {	
    		if (wants_text_events()) {
                convert_event_to_idl_text(event_msg, text_event);
		converted_text=true;
		converted_any=false;
                push(text_event);
            	}
            	else {
                convert_event_to_idl_any(event_msg, any_event);
		converted_text=false;
		converted_any=true;
                push(any_event);
            	}
	}
	else{
	    if (wants_text_events()){
		if (!converted_text)
		{
                	convert_event_to_idl_text(event_msg, text_event);
			converted_text=true;
		}
	        push(text_event);
		}
            else{
		if (!converted_any)
		{
                	convert_event_to_idl_text(event_msg, text_event);
			converted_any=true;
		}
                push(any_event);
		}
        }
	} // End of Try Block
	catch (const CORBA::Exception& e) {
	egw_error.print(
	"Failed to deliver event to consumer %s:\n\t %s Exception thrown\n",
	ae_title_.data(), e._name());
	}
	catch (const SEMTHRInvalidPointer& e) {
	egw_error.print("Failed to deliver event to consumer %s: %s.\n",
	ae_title_.data(), e.what());
	}
	catch (const GeneralPMIException e) {
        egw_error.print( "Failed to deliver event to consumer %s: %s.\n", ae_title_.data(), e.what());
	}
	catch (const MTEMException& e) {
        egw_error.print( "Failed to deliver event to consumer %s: %s.\n", ae_title_.data(), e.what());
	}
        return;
}


/******************************************************************************/
void
CorbaEvent::execute() throw() 
{
    try {
	consumer_->push(event);
    }
    catch (const CosEventComm::Disconnected& e) {
	egw_error.print(
	    "Could not deliver event to consumer %s because the consumer"
	    "disconnected.\n", ae_title_.data());

	/*
	** Since the consumer is disconnected, this object should start
	** cleanup. There is no point in any delivering any pending events to
	** this consumer.
	*/
    }
    catch (const CORBA::Exception& e) {
	CGWUtils::print_corba_exception(
	    egw_error,
	    "Could not deliver event to consumer %s because the following"
	    "exception was thrown:\n",
	    e);
	
	/*
	** Since the consumer is disconnected, this object should start
	** cleanup. There is no point in any delivering any pending events to
	** this consumer.
	*/
    }
    CorbaEvent *tPtr = (CorbaEvent *) this;
    delete tPtr;
//    delete this;
}
/******************************************************************************/
void
EventConsumer::cleanup() throw()
{
    egw_debug.print("EventConsumer::cleanup() starts for %s\n",
		    ae_title_.data());
    
    if (cleanup_started)
	return;
    cleanup_started = true;
    

    consumer_list_.remove(&ae_title_);

    egw_debug.print("EventConsumer::cleanup() ends for %s\n",
		    ae_title_.data());
}

/******************************************************************************/
Asn1Value
EventConsumer::extract_user_id(
    const JIDM::Criteria& access_criteria
) throw (
    JIDM::InvalidCriteria,
    GeneralPMIException,
    SEMTHRInvalidPointer,
    CORBA::SystemException
) {
    const CORBA::ULong len = access_criteria.length();
    CORBA::Any user_profile;
    for (CORBA::ULong i = 0; i < len; ++i) {
        if(CGWUtils::equal(access_criteria[i].name,
                           CGWGlobal::JIDM_USER_PROFILE))
	{
	    user_profile = access_criteria[i].value;
	    break;
        }
    }
    if (len == i) {
	egw_error.print("ProxyAgent Access Criteria does not have %s",
			CGWGlobal::JIDM_USER_PROFILE);
	throw JIDM::InvalidCriteria();
    }
   
    /*AutoPtr<AuthenticationServer> as =
	new AuthenticationServerHandle();*/
    AuthenticationServerHandle as;
 
    RWCString password;
    RWCString user_id;
    bool dummy;
    if(!as.decrypt_user_profile(user_profile, user_id, password,dummy)) {
	egw_error.print("Unable to decrypt JIDM_USER_PROFILE\n");
	throw JIDM::InvalidCriteria();
    }

    TRY {
	Asn1Value user_id_av;
	user_id_av.encode_octets(
	    TAG_GRAPHSTR, DataUnit((char*)user_id.data()));

	return user_id_av;
    }
    BEGHANDLERS
    CATCHALL {
	DynamicOutputStringStream msg;
	msg << "Unable to encode user_id: " << Xreason << endl;
	throw GeneralPMIException(msg.get_string());
    }
    ENDHANDLERS
}

/******************************************************************************/
void
EventConsumer::convert_event_to_idl_any(
    EventReq& event_req,
    CORBA::Any& any_event
) throw (GeneralPMIException, MTEMException)
{
    TRY {
	struct CMIExt::EventReport event_report;

	char buf[1024];
	Oid oc_oid;
	event_req.oc.decode_oid(oc_oid);
	oc_oid.format(buf, sizeof buf);
	event_report.object_class = CORBA::string_dup(buf);
    
	DataUnit fdn = oi2fdn(event_req.oi);
	if (fdn)
	    event_report.object_name = CORBA::string_dup(fdn.chp());
	else 
	    event_report.object_name = CORBA::string_dup("");

	DU event_time_du;
	event_req.event_time.decode_octets(event_time_du);
	if (event_time_du)
	    event_report.event_time = CORBA::string_dup(event_time_du.chp());
	else
	    event_report.event_time = CORBA::string_dup("");

	Oid event_oid;
	event_req.event_type.decode_oid(event_oid);
	event_oid.format(buf, sizeof buf);
	event_report.event_type = CORBA::string_dup(buf);

	IDLValue idl_value(event_report.event_info);
	ACF::GenericAsn1Converter gac =
	    ACF::GenericAsn1Converter(ACF::LOT_EVENT_INFO, event_oid);
	gac.convert_value_from_asn1(event_req.event_info, idl_value);

	any_event <<= event_report;
    }
    BEGHANDLERS
    CATCHALL {
	throw GeneralPMIException(Xreason);
    }
    ENDHANDLERS
}

/******************************************************************************/
void
EventConsumer::convert_event_to_idl_text(
    EventReq& event_req,
    CORBA::Any& text_event
) throw (GeneralPMIException, MTEMException)
{
    TRY {
	struct CMIExt::TextEventReport event_report;

	char buf[1024];
	Oid oc_oid;
	event_req.oc.decode_oid(oc_oid);
	oc_oid.format(buf, sizeof buf);
	event_report.object_class = CORBA::string_dup(buf);
    
	DataUnit fdn = oi2fdn(event_req.oi);
	if (fdn)
	    event_report.object_name = CORBA::string_dup(fdn.chp());
	else
	    event_report.object_name = CORBA::string_dup("");

	DU event_time_du;
	event_req.event_time.decode_octets(event_time_du);
	if (event_time_du)
	    event_report.event_time = CORBA::string_dup(event_time_du.chp());
	else
	    event_report.event_time = CORBA::string_dup("");

	Oid event_oid;
	event_req.event_type.decode_oid(event_oid);
	event_oid.format(buf, sizeof buf);
	event_report.event_type = CORBA::string_dup(buf);

	event_req.event_type.retag(TAG_OID);
	if (((void *)event_req.event_info) != NULL)
	{
		Asn1Type type = Asn1Type::lookup_type( "eventInfo", "eventType", event_req.event_type);
		if (type) {
		 char value[65536];
	    	 char* ptr = value;
	    	  U32 	len = sizeof(value) - 4;
	    	if (!type.format_value(
			event_req.event_info, ptr, len, 0, TAG_EXPLICIT,
			DataUnit(), FV_TRANSLATE_NAMES |FV_USE_C_ESCAPES| FV_USE_HEX))
	    	{
			egw_error.print( "convert_event_to_idl_text(): Asn1Type::format_value() failed for event info of %s \n" ,buf);
			event_report.event_info = CORBA::string_dup("");
	    	}
	    	else {
			ptr[0] = '\0';
			event_report.event_info = CORBA::string_dup(value);
	    	}
		}
		else {
			event_report.event_info = CORBA::string_dup("");
	    		egw_error.print(
	        	"convert_event_to_idl_text(): Asn1Type::lookup_type() failed for %s", buf);
		}
	}
	else
	{
		event_report.event_info = CORBA::string_dup("");
	}

	text_event <<= event_report;
    }
    BEGHANDLERS
    CATCHALL {
	throw GeneralPMIException(Xreason);
    }
    ENDHANDLERS
}

/******************************************************************************/
