fipa::acl::ConversationMonitor Class Reference

This library provides a statemachine to monitor a message flow based on performatives. The conversation monitor is capable of monitoring the conversation based on the protocol specified in the messages. There are scxml documents specifying how these protocols are defined. Protocols may contain subprotocols, which is partially supported. More...

#include <conversation_monitor.h>

Public Member Functions
	ConversationMonitor (const AgentID &self, const std::string &protocolDirectory="")
virtual	~ConversationMonitor ()
ConversationPtr	updateConversation (const fipa::acl::ACLMessage &msg)
ConversationPtr	startConversation (const std::string &topic)
bool	removeConversation (const fipa::acl::ConversationID &conversationId)
ConversationPtr	getConversation (const fipa::acl::ConversationID &conversationId)
ConversationPtr	getOrCreateConversation (const fipa::acl::ConversationID &conversationId)
void	cleanup ()
std::vector< fipa::acl::ConversationID >	getActiveConversations ()

Detailed Description

This library provides a statemachine to monitor a message flow based on performatives. The conversation monitor is capable of monitoring the conversation based on the protocol specified in the messages. There are scxml documents specifying how these protocols are defined. Protocols may contain subprotocols, which is partially supported.

The following sections provide an overview how to use the monitor and which constraints it has.

Examples

To verify a conversation, one could do the following:

ConversationMonitor monitor (selfAgentId);
ConversationPtr convPtr = monitor.startConversation("myTopic");
ACLMessage myMsg(ACLMessage::INFORM);
// Continue creating message, e.g. using inform protocol
myMsg.setProtocol("inform");
// ...
convPtr->update(myMsg);
if(convPtr->hasEnded())
{
    // ...
}

If the conversation's protocol has subprotocols, at the point before starting a subProtocol, this becomes necessary:

// ...
convPtr->setNumberOfSubConversations(n);
// where n is the number of subconversations the agent plans to start.

If one wishes to be notified of changes to a conversation, one can use the following after obtaining a ConversationPtr:

ConversationObserverPtr observer(new ConversationObserver());
convPtr->addObserver(observer);
// Method A:
conversation::Event event = observer->waitForNextEvent();
// ...

// Method B:
conversation::Event event;
if(observer->hasEvents() && observer->getNextEvent(event))
{
    // ...
}

Constraints

At the moment, to handle subprotocols correctly a few constraints apply. If the agents to not obey these rules, the monitoring will fail.

• There can be only one substatemachine per state. More than one could work, but could also yield unexpected behaviour.
• For proxied protocols (like "brokering") that send the responses back there must be exactly one forwarded reply; unless the proxied protocol is "inform", in which case there must be no forwarded reply.
• Protocols where responses are sent directly to the initiator (not via a broker), like "recruiting", are not supported.
• There is no way to check if the started subprotocol was indeed the one specified in the proxy/propagate message.
• A propagating protocol cannot be terminated with a failure/cancel/not-understood.
• "setNumberOfSubConversations" has to be used for all sub-conversations.

Definition at line 75 of file conversation_monitor.h.

Constructor & Destructor Documentation

ConversationMonitor()

fipa::acl::ConversationMonitor::ConversationMonitor	(	const AgentID &	self,
		const std::string &	protocolDirectory = ""
	)

Construct the conversation monitor using the system configuration object which for example holds the name of the agent

Parameters

self	AgendID of the current agent
protocolDirectory	Directory of interaction protocols

Definition at line 7 of file conversation_monitor.cpp.

    : mSelf(self)
{
    LOG_DEBUG("Creating conversation monitor for agent: '%s'", self.getName().c_str());
    if(!protocolDirectory.empty())
    {
        LOG_DEBUG("Setting protocol resource directory: '%s'",protocolDirectory.c_str());
        fipa::acl::StateMachineFactory::setProtocolResourceDir( protocolDirectory );
    }
}

~ConversationMonitor()

fipa::acl::ConversationMonitor::~ConversationMonitor ( )

virtual

Default deconstructor

Definition at line 18 of file conversation_monitor.cpp.

{}

Member Function Documentation

cleanup()

void fipa::acl::ConversationMonitor::cleanup

(

)

Cleanup all conversations, that ended

Definition at line 93 of file conversation_monitor.cpp.

{
    LOG_DEBUG_S << "Cleaning up conversation monitor.";
    boost::unique_lock<boost::mutex> lock(mMutex);

    std::map<fipa::acl::ConversationID, ConversationPtr>::iterator it = mActiveConversations.begin();
    std::vector<fipa::acl::ConversationID> endedConversations;
    std::vector<fipa::acl::ConversationID>::iterator eit;

    // If conversation has ended move from active conversation to ended conversation after detaching 
    // any existing observers
    for(; it != mActiveConversations.end(); ++it)
    {
        LOG_DEBUG_S << "Cleaning up conversation monitor: Processing conversation " << it->first;
        ConversationPtr conversation = it->second;
        assert(it->second);
        if(conversation->hasEnded())
        {
            LOG_DEBUG_S << "Detaching observers from ended conversation " << it->first;
            conversation->detachObservers();
            endedConversations.push_back(it->first);
        }
    }
    
    LOG_DEBUG_S << "Erasing ended conversations from active conversations.";
    for( eit = endedConversations.begin(); eit != endedConversations.end(); ++eit)
    {
        LOG_DEBUG_S << "Erasing ended conversation " << *eit << " from active conversations.";
        mActiveConversations.erase(*eit);
    }
}

getActiveConversations()

std::vector< fipa::acl::ConversationID > fipa::acl::ConversationMonitor::getActiveConversations

(

)

Get a list of conversation ids of all active conversations

Returns

List of conversation ids currently active conversations

Definition at line 125 of file conversation_monitor.cpp.

{
    boost::unique_lock<boost::mutex> lock(mMutex);
    std::map<fipa::acl::ConversationID, ConversationPtr>::iterator it = mActiveConversations.begin();
    std::vector<fipa::acl::ConversationID> activeConversations;

    for(; it != mActiveConversations.end(); it++)
    {
        activeConversations.push_back(it->first);
    }

    return activeConversations;
}

getConversation()

ConversationPtr fipa::acl::ConversationMonitor::getConversation

(

const fipa::acl::ConversationID &

conversationId

)

Get conversation associated with a certain conversation id User has to verify that id of conversation is unqual to an empty string

Definition at line 66 of file conversation_monitor.cpp.

{
    // get conversation associated with command of uuid
    boost::unique_lock<boost::mutex> lock(mMutex);
    std::map<fipa::acl::ConversationID, ConversationPtr>::iterator it = mActiveConversations.find(conversationId);
    if(it != mActiveConversations.end())
        return it->second;

    return ConversationPtr();
}

getOrCreateConversation()

ConversationPtr fipa::acl::ConversationMonitor::getOrCreateConversation

(

const fipa::acl::ConversationID &

conversationId

)

Get conversation associated with a certain conversation id or create a new one if it does not exist yet

Definition at line 78 of file conversation_monitor.cpp.

{
    // get conversation associated with comman of uuid
    boost::unique_lock<boost::mutex> lock(mMutex);
    std::map<fipa::acl::ConversationID, ConversationPtr>::iterator it = mActiveConversations.find(conversationId);
    if(it != mActiveConversations.end())
        return it->second;

    // create if it does not exist
    ConversationPtr conversation(new Conversation(mSelf.getName(), conversationId));
    mActiveConversations.insert(std::pair<fipa::acl::ConversationID, ConversationPtr>(conversationId, conversation));
    LOG_INFO("Create new conversation '%p' with conversation id '%s'", this, conversationId.c_str());
    return conversation;
}

removeConversation()

bool fipa::acl::ConversationMonitor::removeConversation

(

const fipa::acl::ConversationID &

conversationId

)

Close a conversation with the provided conversationId

Definition at line 50 of file conversation_monitor.cpp.

{
    boost::unique_lock<boost::mutex> lock(mMutex);
    bool success = mActiveConversations.erase(conversationId);
    return success;
}

startConversation()

ConversationPtr fipa::acl::ConversationMonitor::startConversation

(

const std::string &

topic

)

Start a conversation without a message - this is required for internal requests refer to OutgoingMessageHandler::operator()() The id will be automatically generated at construction time of Conversation and extended by the topic if required

Parameters

topic

Topic to add to the automatically generated conversation id

Returns

Pointer to the created conversation

Definition at line 57 of file conversation_monitor.cpp.

{
    boost::unique_lock<boost::mutex> lock(mMutex);
    // Start conversation with this agent as owner
    ConversationPtr conversation(new Conversation(mSelf.getName(), Conversation::generateConversationID(topic) ));
    mActiveConversations.insert(std::pair<std::string, ConversationPtr>(conversation->getConversationId(), conversation));
    return conversation;
}

updateConversation()

ConversationPtr fipa::acl::ConversationMonitor::updateConversation

(

const fipa::acl::ACLMessage &

msg

)

When an internal message has been generated as a response or as an initiation of a conversation update the conversation monitor using this function

Parameters

the	outgoing message

Returns

pointer to the conversation

Exceptions

InvalidOperation	if conversation does not exist this message
ProtocolError	if conversation cannot consume the message due to an invalid protocol flow

Definition at line 21 of file conversation_monitor.cpp.

{
    boost::unique_lock<boost::mutex> lock(mMutex);
    std::map<std::string, ConversationPtr>::iterator it;
    std::string conversationId = msg.getConversationID();
    it = mActiveConversations.find(conversationId);
    // update if conversation already exists
    if(it != mActiveConversations.end())
    {
        bool conversationEnded = false;
        try {
            conversationEnded = it->second->hasEnded();
        } catch(const std::runtime_error& e)
        {}

        if(conversationEnded)
        {
            std::string errorMsg = "Trying to update already completed conversation: " + conversationId + " performative: '" + msg.getPerformative() + "' content: '" + msg.getContent() + "'";
            throw conversation::InvalidOperation(errorMsg);
        } else {
            LOG_INFO("Update existing conversation '%p' with conversation id '%s'", this, it->second->getConversationId().c_str());
            it->second->update(msg);
            return it->second;
        }
    }

    throw conversation::InvalidOperation("Can only update an existing conversation. Possibly the corresponding conversation did already end and this message does not conform to the respective protocol");
}

---

The documentation for this class was generated from the following files:

• /build/rock-master-18.09-multiagent-fipa-acl-0.20180914/src/conversation_monitor/conversation_monitor.h
• /build/rock-master-18.09-multiagent-fipa-acl-0.20180914/src/conversation_monitor/conversation_monitor.cpp