CommunicationManagerImpl.cpp

Go to the documentation of this file.

/** @file    CommunicationManagerImpl.cpp
 *  @author  Alessandro Polo
 *  @version $Id: CommunicationManagerImpl.cpp 3775 2011-01-01 16:38:17Z alex $
 *  @brief
 * File containing methods for the CommunicationManagerImpl class.
 * The header for this class can be found in CommunicationManagerImpl.h, check that file
 * for class description.
 ****************************************************************************/
/* Copyright (c) 2007-2011, WOSH - Wide Open Smart Home 
 * by Alessandro Polo - OpenSmartHome.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the OpenSmartHome.com WOSH nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Alessandro Polo ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Alessandro Polo BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ****************************************************************************/

 #include "CommunicationManagerImpl.h"
 #include <core/UserManager.h>
 #include <core/SecurityManager.h>
 #include <core/Response.h>
 #include <core/Variant.h>
 #include <stdlib.h>
 #include <time.h>
 #include <algorithm>


using namespace std;
using namespace wosh;

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS

CommunicationManagerImpl::CommunicationManagerImpl( BundleGeneric& bundle ) : BundleGenericWorker(bundle)
 {
    this->listener = NULL;

    this->flowTimeout = 0;
    this->flowRetryTime = 20;
    this->flowRetryCount = 10;
    
    this->messageArchiveSize = -1;
 }


CommunicationManagerImpl::~CommunicationManagerImpl() {
    if ( isThreadRunning() ) {
        quitThread(10000);
     }

    if ( this->comms.count() != 0 ) {
        Log(LOG_VERBOSE, ":~CommunicationManagerImpl() : Freeing Communicators.." );
        this->comms.clear();
     }
    if ( this->transl.count() != 0 ) {
        Log(LOG_VERBOSE, ":~CommunicationManagerImpl() : Freeing Interpreters.." );
        this->transl.clear();
     }
    if ( this->messages.count() != 0 ) {
        Log(LOG_VERBOSE, ":~CommunicationManagerImpl() : Freeing Messages.." );
        this->messages.clear();
     }
    if ( this->messagesArchived.count() != 0 ) {
        Log(LOG_VERBOSE, ":~CommunicationManagerImpl() : Freeing Archived Messages.." );
        this->messagesArchived.clear();
     }

 }

//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////// RUN THREAD

void CommunicationManagerImpl::runThread()
 {
    Log(LOG_VERBOSE, ":runThread(): STARTING" );

    bool cProcessed = false;
    while( this->running )
     {
        ThreadImpl::sleepForMSec(100);
        setThreadAlive();

        // walk the notification queue
        this->messages.transactionBeginWrite();
        tLongNotificationFlowMap::Iterator it;
        tLongNotificationFlowMap::ConstIterator it_end = this->messages.end();
        for ( it=this->messages.begin(); it!=it_end; ++it ) {
            if ( it->second == NULL ) {
                this->messages.erase(it);
                continue;
             }
            cProcessed = processNotification_( it->second );
            if ( !cProcessed ) break; // reset the iterator
         }
        this->messages.transactionEnd();
     }

    Log(LOG_VERBOSE, ":runThread(): EXITING" );
 }

////////////////////////////////////////////////////////////////////////////////////////////////// RUN THREAD
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////// PROCESSING

bool CommunicationManagerImpl::processNotification_( NotificationFlow* notification )
 {
    if ( notification == NULL ) return false;
    if ( !notification->hasOriginal() ) {
        notification->raiseError();
        Log(LOG_CRITICAL, ":processNotification_(%ld) : Notification MISSING [%ld errors] [ARCHIVING]", notification->getID(), notification->getErrorsCount() );
        archiveNotification_( notification );
        return false;
     }

    if ( !notification->isProcessing() ) {
        // start the processing for first time!
        Log(LOG_VERBOSE, ":processNotification_(%ld) : Starting Processing! [of %ld]", notification->getID(), notification->getOriginal()->getID() );
        // flag it as under-processing and try to send (process) it
        notification->setProcessing(true);
        sendNotification_(notification);
        return true;
     }

    // we are/were already processing the flow!
    // cases are:
    //  - we picked a communicator/interpreter and waiting for response.
    //  - we are waiting for READ flag 
    //  - something went wrong!?

    // check if we processed the flow too many times
    if ( this->flowRetryCount != 0 ) {
        if ( notification->getProcessingCount() >= this->flowRetryCount ) {
            Log(LOG_WARNING, ":processNotification_(%ld) : ARCHIVING due to retry-max [%ld] (sent=%ld)", notification->getID(), notification->getProcessingCount(), notification->getSentCount() );
            notification->raiseError();
            archiveNotification_( notification );
            return false;
         }
     }
    // check if the flow is timed out
    if ( this->flowTimeout != 0 ) {
        long timedout = (DateTime::std_time_ms() - notification->getOriginal()->getTimeStamp()) / 1000;
        if ( timedout > this->flowTimeout ) {
            Log(LOG_WARNING, ":processNotification_(%ld) : ARCHIVING due to timeout [%lds] (sent=%ld)", notification->getID(), timedout, notification->getSentCount() );
            notification->raiseError();
            ///@todo reply to sender that it is (èartially) archived
            archiveNotification_( notification );
            return false;
         }
     }

    // check if we 'just' (a while ago) processed the notification
    if ( this->flowRetryTime != 0 ) {
        long timespan = (DateTime::std_time_ms() - notification->getProcessingTs()) / 1000;
        if ( timespan <= this->flowRetryTime ) {
            Log(LOG_VERBOSE, ":processNotification_(%ld) : WAITING due to reply-span [%lds] (sent=%ld)", notification->getID(), timespan, notification->getSentCount() );
            return true;
         }
     }
    // retry again !?

    notification->setProcessing(true);
//  notification->processing();
    sendNotification_(notification);
    return true;
 }

////////////////////////////////////////////////////////////////////////////////////////////////// PROCESSING
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////// SENDING

bool CommunicationManagerImpl::sendNotification_( NotificationFlow* notification )
 {
    if ( notification == NULL ) return false;

    Notification* selectedNotification = notification->getFinal();
    if ( selectedNotification == NULL ) return false;

    Log(LOG_VERBOSE, ":sendNotification_(%ld) : Sending.. [%ld]", notification->getID(), notification->getSentCount() );

    // FLOW:
    // - get user's communicator and merge with options (forced comm, preferred comm)
    // - filter results with local database and: communicator vs. notification check
    // - select communicator with best ranking
    // - use an interpreter when required

    UserInfoCommmunicators* usrComms = getCommunicatorsOf( selectedNotification->getRecipent_User() );

    // does the notification suggest a communicator? 
    string preferredCommURI = selectedNotification->getPreferredCommunicator().toString();
    if ( !preferredCommURI.empty() ) {
        // force URI intepreter
        URI fcu(preferredCommURI);
        preferredCommURI = fcu.toString();
        // update (increase) ranking of the communicator  (won't update timestamp)
        usrComms->updateRankingOf(preferredCommURI, 0.2, true);
     }

    // does the user force a communicator
    string forceCommURI = UserManager::getUserProperty(selectedNotification->getRecipent_User(), "CommunicatorForce").toString();
    if ( !forceCommURI.empty() ) {
        // force URI intepreter
        URI fcu(forceCommURI);
        forceCommURI = fcu.toString();
        // update (increase) ranking of the communicator  (won't update timestamp)
        usrComms->updateRankingOf(forceCommURI, 1000, true);
     }

    usrComms->filterTimedOut(0);
    if ( usrComms->getCommmunicators().empty() ) {
        Log(LOG_WARNING, ":sendNotification_(%ld) : FAILED NO ONLINE communicators for user %s", notification->getID(), selectedNotification->getRecipent_User().c_str() );
notification->raiseError();
        return false;
     }

    // does the user WANT a custom protocol? (example: 'WOSH' may talk only using PROTOCOL_SHELL!)
    std::string forceProto = UserManager::getUserProperty(selectedNotification->getRecipent_User(), "ProtocolForce").toString();
    std::string location = UserManager::getUserLocation( selectedNotification->getRecipent_User() );

    intersectCommunicators_(usrComms, selectedNotification, location );
    if ( usrComms->getCommmunicators().empty() ) {
        Log(LOG_WARNING, ":sendNotification_(%ld) : FAILED No FILTERED communicators for user %s", notification->getID(), selectedNotification->getRecipent_User().c_str() );
        delete usrComms; usrComms = NULL;
notification->raiseError();
        return false;
     }

    Communicator* communicator = NULL;
    string communicatorURI = "";
    UserInfoCommmunicatorData communicatorData;
    usrComms->getCommmunicatorRank(communicatorURI, communicatorData);

    this->comms.transactionBeginRead();
    communicator = this->comms.find( communicatorURI );
    if ( communicator == NULL ) {
        this->comms.transactionEnd();
        UserManager::setUserCommunicator( selectedNotification->getRecipent_User(), communicatorURI, 0, -0.15 );
        Log(LOG_WARNING, ":sendNotification_(%ld) : FAILED accessing communicator %s for user %s ", notification->getID(), communicatorURI.c_str(), selectedNotification->getRecipent_User().c_str() );
        delete usrComms; usrComms = NULL;
notification->raiseError();
        return false;
     }

    Log(LOG_VERBOSE, ":sendNotification_(%ld) : Selected %s (out of %d) for user %s ", notification->getID(), communicatorURI.c_str(), usrComms->getCommmunicators().size(), selectedNotification->getRecipent_User().c_str() );
    delete usrComms; usrComms = NULL;

    string interpeterURI = "";

    // interpeter needed? (rule: minimize use)
    bool iRequired = isInterpreterRequired_( selectedNotification, communicator );
    if ( iRequired ) {
        this->transl.transactionBeginRead();
        Interpreter* inter = selectInterpreter_( selectedNotification, communicator );
        if ( inter == NULL ) {
            Log(LOG_WARNING, ":sendNotification_(%ld) : FAILED selecting Interpreter! [for %s]", notification->getID(), communicatorURI.c_str() );
         }
        else {
            interpeterURI = inter->getURI().toString();
            Log(LOG_VERBOSE, ":sendNotification_(%ld) : Selected Interpreter %s [for %s]", notification->getID(), interpeterURI.c_str(), communicatorURI.c_str() );
         }
        this->transl.transactionEnd();
     }

    this->comms.transactionEnd();

    Message* message = new Message();
    message->setSource(this->getOwner()->getURI());
    message->setDestinationBus( _Bus_Core );
    SecurityManager::getCredentialImplicit().signMessage(message, notification->getOriginal()->getSender_User());

    if ( !interpeterURI.empty() && communicator != NULL ) {
        Log(LOG_INFO, ":sendNotification_(%ld) : Flow [%s=>%s] selected", notification->getID(), interpeterURI.c_str(), communicatorURI.c_str() );

        NotificationConversion* conv = new NotificationConversion();
        conv->setOriginal( selectedNotification->clone() );
        conv->setProtocol(communicator->getProtocol());
        //conv->setNotificationType(communicator->getChannelTypes());
        Request* request = new Request(_InterpreterService_METHOD_convert, conv );
        message->setContent(request);
        message->setDestination( interpeterURI );
     }
    else if ( !communicatorURI.empty() ) {
        Log(LOG_INFO, ":sendNotification_(%ld) : Communicator %s selected, sending[%ld]..", notification->getID(), communicatorURI.c_str(), notification->getSentCount() );

        notification->sending();
        notification->setSenderCommunicator(communicatorURI);
        message->setContent( selectedNotification->clone() );
        message->setDestination( communicatorURI );
        BusCore.postMessage(message);
        return true;
     }
    else {
        Log(LOG_CRITICAL, ":sendNotification_(%ld) : Internal ERROR!", notification->getID() );
        delete message;
notification->raiseError();
        return false;
     }
    return false;
 }

///////////////////////////////////////////////////////////////////////////////////////////////////// SENDING
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////// UPDATING

void CommunicationManagerImpl::evalMessage( const Message& message )
 {
    if ( message.isEmpty() ) return;

    if ( message.getContent()->isNotification() ) {
        const Notification* notification = message.getContent()->asNotification();
        this->messages.transactionBeginWrite();
        evalNotification_(notification, &message);
        this->messages.transactionEnd();
        return;
     }
    else if ( message.getContent()->isResponse() ) {
        const Response* response = message.getContent()->asResponse();
        if ( !response->hasData() ) return;

        if ( response->getMethod() == _Communicator_RESPONSE_notif_sent ) {
            // seems a communicator replied to our sending, means it was SENT
            long not_id = 0;
            if ( response->getData()->isKindOf<Variant>() ) {
                not_id = dynamic_cast<const Variant*>(response->getData())->toLong(0);
             }
            if ( not_id == 0 ) {
                Log(LOG_WARNING, ":evalResponse() : Unsupported/Missing value in send_message response [by %s]", message.getSource().toString().c_str() );
             }
            else
                communicator_reply_sent(response->getReturnValue(), not_id, message.getSource() );
            return;
         }
        else if ( response->getMethod() == _Communicator_RESPONSE_notif_read ) {
            // seems a communicator replied to our sending, means it was READ
            long not_id = 0;
            if ( response->getData()->isKindOf<Variant>() ) {
                not_id = dynamic_cast<const Variant*>(response->getData())->toLong(0);
             }
            if ( not_id == 0 ) {
                Log(LOG_WARNING, ":evalResponse() : Unsupported/Missing value in send_message response [by %s]", message.getSource().toString().c_str() );
             }
            else
                communicator_reply_read(response->getReturnValue(), not_id, message.getSource() );
            return;
         }
     }
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

NotificationFlow* CommunicationManagerImpl::evalNotification_( const Notification* notification, const Message* message )
 {
    if ( notification == NULL || message == NULL ) return NULL;

    NotificationFlow* notificationEx = NULL;
    if ( notification->getFlowID() == 0 ) {
        // notification is NEW (at least for myself)! => create a new flow and set as original (associating uid)
        notificationEx = new NotificationFlow( notification->clone() );
        notificationEx->setSourceCommunicator(message->getSource());

        Log(LOG_VERBOSE, ":evalNotification_(%ld) : New FLOW#%ld from %s to %s [%s]",
                        notification->getID(), notificationEx->getID(), notification->getSender_User().c_str(), notification->getRecipent_User().c_str(),
                        notification->getNotificationTypeAsString() );

        this->messages.set( notificationEx->getID(), notificationEx );
        UserManager::setUserCommunicator( notification->getSender_User(), message->getSource().toString(), DateTime::std_time_ms(), 0.01 );
     }
    else {
        // so, maybe we are updating a known Notification
        notificationEx = this->messages.find(notification->getFlowID());
        if ( notificationEx == NULL ) {
            Log(LOG_WARNING, ":evalNotification_(%ld) : FAILED Flow#%ld not found!", notification->getID(), notification->getFlowID() );
            notificationEx = this->messagesArchived.find(notification->getFlowID());
            if ( notificationEx == NULL ) {
                Log(LOG_WARNING, ":evalNotification_(%ld) : FAILED Flow#%ld not found [even in Archived Cache]!", notification->getID(), notification->getFlowID() );
                return NULL;
             }
         }

        Log(LOG_VERBOSE, ":evalNotification_(%ld) : Updating Flow#%ld!", notification->getID(), notification->getFlowID() );
        notificationEx->addInterpreted(notification->clone());
        notificationEx->setProcessing(false);
     }

    return notificationEx;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void CommunicationManagerImpl::communicator_reply_read( WRESULT ret, long notification_id, const URI& source )
 {
    this->messages.transactionBeginWrite();
    tLongNotificationFlowMap::Iterator it;
    tLongNotificationFlowMap::ConstIterator it_end = this->messages.end();
    for ( it=this->messages.begin(); it!=it_end; ++it ) {
        if ( it->second == NULL || !it->second->hasOriginal() ) continue;
        if ( it->second->getOriginal()->getID() != notification_id ) continue;
        if ( WSUCCEEDED(ret) ) {
            it->second->getOriginal()->setRead(true);
            UserManager::setUserCommunicator( it->second->getOriginal()->getRecipent_User(), source.toString(), DateTime::std_time_ms(), 0.10 );
            Log(LOG_INFO, ":communicator_reply_read(%ld) : Flow#%ld Archived", notification_id, it->second->getID() );
            archiveNotification_( it->second );
         }
        else {
            it->second->getOriginal()->setRead(false);
            Log(LOG_WARNING, ":communicator_reply_read(%ld) : ERROR#%d by %s", notification_id, ret, source.toString().c_str() );
            it->second->raiseError();
         }
        break;
     }
    this->messages.transactionEnd();
 }

void CommunicationManagerImpl::communicator_reply_sent( WRESULT ret, long notification_id, const URI& source )
 {
    this->messages.transactionBeginWrite();
    tLongNotificationFlowMap::Iterator it;
    tLongNotificationFlowMap::ConstIterator it_end = this->messages.end();
    for ( it=this->messages.begin(); it!=it_end; ++it ) {
        if ( it->second == NULL || !it->second->hasOriginal() ) continue;
        if ( it->second->getOriginal()->getID() !=  notification_id ) continue;
        if ( WSUCCEEDED(ret) ) {
            it->second->getOriginal()->setSent(true);
            it->second->setSenderCommunicator(source);
            Log(LOG_VERBOSE, ":communicator_reply_sent() : Notification#%ld of Flow#%ld has been Sent!", notification_id, it->second->getID() );
//          Log(LOG_INFO, ":communicator_reply_sent(%ld) : Flow#%ld Archived", notification_id, it->second->getID() );
//          it->second->setArchived(true);
//archiveNotification_( it->second );
         }
        else {
            Log(LOG_WARNING, ":communicator_reply_sent(%ld) : ERROR#%d by %s", notification_id, ret, source.toString().c_str() );
            it->second->raiseError();
            UserManager::setUserCommunicator( it->second->getOriginal()->getRecipent_User(), source.toString(), DateTime::std_time_ms(), -0.2 );
         }
        break;
     }
    this->messages.transactionEnd();
 }

//////////////////////////////////////////////////////////////////////////////////////////////////// UPDATING
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void CommunicationManagerImpl::archiveNotification_( NotificationFlow* notification )
 {
    if ( notification == NULL ) return;
    Log(LOG_INFO, ":archiveNotification_(%ld) : Archiving Notification [%ld errors]", notification->getID(), notification->getErrorsCount() );

    notification->setProcessing(false);
    notification->setArchived(true);
    this->messages.erase( notification->getID(), false );
    this->messagesArchived.set( notification->getID(), notification );
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////