// [License]
// The Ariba-Underlay Copyright
//
// Copyright (c) 2008-2009, Institute of Telematics, Universität Karlsruhe (TH)
//
// Institute of Telematics
// Universität Karlsruhe (TH)
// Zirkel 2, 76128 Karlsruhe
// Germany
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// 2. 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.
//
// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``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 THE ARIBA PROJECT OR
// CONTRIBUTORS 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.
//
// The views and conclusions contained in the software and documentation
// are those of the authors and should not be interpreted as representing
// official policies, either expressed or implied, of the Institute of
// Telematics.
// [License]

#include "BaseOverlay.h"

#include "ariba/utility/misc/OvlVis.h"
#include "ariba/NodeListener.h"
#include "ariba/CommunicationListener.h"
#include "ariba/SideportListener.h"

#include "ariba/overlay/messages/OverlayMsg.h"
#include "ariba/overlay/messages/JoinRequest.h"
#include "ariba/overlay/messages/JoinReply.h"
#include "ariba/overlay/messages/LinkRequest.h"

namespace ariba {
namespace overlay {

use_logging_cpp(BaseOverlay);

BaseOverlay::BaseOverlay()
	: bc(NULL), overlayInterface(NULL), nodeId(NodeID::UNSPECIFIED),
		spovnetId(SpoVNetID::UNSPECIFIED), initiatorLink(LinkID::UNSPECIFIED),
		state(BaseOverlayStateInvalid), sideport(&SideportListener::DEFAULT){
}

BaseOverlay::~BaseOverlay(){
}

void BaseOverlay::start( BaseCommunication& _basecomm, const NodeID& _nodeid ){

	bc = &_basecomm;
	nodeId = _nodeid;

	logging_info("creating base overlay");

	bc->registerMessageReceiver( this );
	bc->registerEventListener( this );

	ovl.visCreate( ovlId, nodeId, string(""), string("") );
	ovl.visChangeNodeColor(ovlId, nodeId, OvlVis::NODE_COLORS_GREY);

// 	if (Identifier(Configuration::instance().read<unsigned long>("BASE_nodeid")) ==
// 	    Identifier(Configuration::instance().read<unsigned long>("SOURCE"))) {
// 		ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CAMERA);
// 	} else if (Identifier(Configuration::instance().read<unsigned long>("BASE_nodeid")) ==
// 	    Identifier(Configuration::instance().read<unsigned long>("MR_A"))) {
// 		ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CHARACTER_A);
// 	} else if (Identifier(Configuration::instance().read<unsigned long>("BASE_nodeid")) ==
// 	    Identifier(Configuration::instance().read<unsigned long>("MR_W"))) {
// 		ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CHARACTER_W);
// 	}

	// timer for auto link management
	Timer::setInterval( 5000 );
	Timer::start();
}

void BaseOverlay::stop() {

	logging_info("deleting base overlay");

	Timer::stop();
	bc->unregisterMessageReceiver( this );
	bc->unregisterEventListener( this );

	if(overlayInterface != NULL){
		delete overlayInterface;
		overlayInterface = NULL;
	}
}

void BaseOverlay::joinSpoVNet(const SpoVNetID& id, const EndpointDescriptor& bootstrapEp){

	ovl.visShowNodeBubble ( ovlId, nodeId, "joining..." );
	logging_info( "starting to join spovnet " << id.toString() <<
			" with nodeid " << nodeId.toString());

	//
	// contact the spovnet initiator and request
	// to join. if the join is granted we will
	// receive further information on the structure
	// of the overlay that is used in the spovnet
	//
	// but first, we have to establish a link to the initiator...
	//

	spovnetId = id;
	state = BaseOverlayStateJoinInitiated;

	initiatorLink = bc->establishLink( bootstrapEp );
	logging_info("join process initiated for " << id.toString() << "...");
}

void BaseOverlay::leaveSpoVNet(){

	logging_info( "leaving spovnet " << spovnetId );
	bool ret = ( state != this->BaseOverlayStateInvalid );

	logging_debug( "dropping all auto-links ..." );

	// now we start leaving the spovnet: fist delete all links
	// that we still have in the baseoverlay initiated by
	// some services, the leave the actual overlay structure,
	// then leave the spovnet

	// --> drop all service links

	vector<LinkID> servicelinks;
	BOOST_FOREACH( LinkPair item, linkMapping ){
		if( item.second.service != OverlayInterface::OVERLAY_SERVICE_ID )
			servicelinks.push_back( item.first );
	}
	BOOST_FOREACH( LinkID lnk, servicelinks ){
		// the dropLink function will remove
		// the item from the linkMapping
		dropLink( lnk );
	}

	// --> leave overlay structure

	logging_debug( "leaving overlay" );
	// first, leave the overlay interface
	if( overlayInterface != NULL )
		overlayInterface->leaveOverlay();

	// --> leave spovnet

	if( state != BaseOverlayStateInitiator ){

		// then, leave the spovnet baseoverlay
		OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeBye, nodeId );
		bc->sendMessage( initiatorLink, &overMsg );

		// drop the link and set to correct state
		bc->dropLink( initiatorLink );
		initiatorLink = LinkID::UNSPECIFIED;
	}

	state = BaseOverlayStateInvalid;
	ovl.visShutdown( ovlId, nodeId, string("") );

	// inform all registered services of the event
	BOOST_FOREACH( NodeListener* i, nodeListeners ){
		if( ret ) i->onLeaveCompleted( spovnetId );
		else      i->onLeaveFailed( spovnetId );
	}
}

void BaseOverlay::createSpoVNet(const SpoVNetID& id, const OverlayParameterSet& param, const SecurityParameterSet& sec, const QoSParameterSet& qos){

	// set the state that we are an initiator, this way incoming messages are
	// handled correctly
	logging_info( "creating spovnet " + id.toString() <<
			" with nodeid " << nodeId.toString() );

	spovnetId = id;
	state = BaseOverlayStateInitiator;

	overlayInterface = OverlayFactory::create( *this, param, nodeId, this );
	if( overlayInterface == NULL ){
		logging_fatal( "overlay structure not supported" );
		state = BaseOverlayStateInvalid;
		return;
	}

	// bootstrap against ourselfs
	overlayInterface->joinOverlay();
	BOOST_FOREACH( NodeListener* i, nodeListeners )
		i->onJoinCompleted( spovnetId );

	ovl.visChangeNodeIcon ( ovlId, nodeId, OvlVis::ICON_ID_CAMERA );
	ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN);
}


/// establishes a link between two arbitrary nodes
const LinkID BaseOverlay::establishLink( const NodeID& node,
		const ServiceID& service, const LinkID& link_id ) {

	if( !communicationListeners.contains( service ) ){
		logging_error( "no registered listener on serviceid " << service.toString() );
		return LinkID::UNSPECIFIED;
	}

	// copy link id
	LinkID linkid = link_id;

	// create link id if necessary
	if (linkid.isUnspecified()) linkid = LinkID::create();

	// debug message
	logging_debug( "BaseOverlay called to establish link between node " <<
			node.toString() << " for service " << service.toString() );

	// create link request message with own link id
	OverlayMsg overlay_msg( OverlayMsg::OverlayMessageTypeLinkRequest, service, nodeId );
	uint32_t nonce = (uint32_t)(rand() ^ (rand() << 16) ^ time(NULL));
	LinkRequest link_request_msg( nonce, &bc->getEndpointDescriptor() );
	overlay_msg.encapsulate( &link_request_msg );
	pendingLinks.insert( make_pair(nonce, linkid) );

	// debug message
	logging_debug( "BaseOverlay routes LinkRequest message to node " << node.toString() );

	// route message to overlay node
	overlayInterface->routeMessage( node, &overlay_msg );

	CommunicationListener* receiver = communicationListeners.get( service );
	assert( receiver != NULL );

	LinkItem item (linkid, NodeID::UNSPECIFIED, service, receiver);
	linkMapping.insert( make_pair(linkid, item) );

	return linkid;
}

const LinkID BaseOverlay::establishLink( const EndpointDescriptor& ep,
		const ServiceID& service, const LinkID& linkid  ){

	if( !communicationListeners.contains( service ) ){
		logging_error( "no registered listener on serviceid " << service.toString() );
		return LinkID::UNSPECIFIED;
	}

	const LinkID link = bc->establishLink( ep, linkid );

	CommunicationListener* receiver = communicationListeners.get( service );
	assert( receiver != NULL );

	LinkItem item (link, NodeID::UNSPECIFIED, service, receiver);
	linkMapping.insert( make_pair(link, item) );

	return link;
}

void BaseOverlay::dropLink(const LinkID& link){

	logging_debug( "baseoverlay dropping link " << link.toString() );
	LinkMapping::iterator i = linkMapping.find( link );

	// find the link item to drop
	if( i == linkMapping.end() ){
		logging_warn( "can't drop link, mapping unknown " << link.toString() );
		return;
	}

	LinkItem item = i->second;

	// delete all queued messages
	if( item.waitingmsg.size() > 0 ){

		logging_warn( "dropping link " << link.toString() <<
			" that has " << item.waitingmsg.size() << " waiting messages" );

		item.deleteWaiting();
	}

	// erase the mapping and drop the link
	linkMapping.erase( i );
	bc->dropLink( link );

	// tell sideports and listeners of the drop
	item.interface->onLinkDown( link, item.node );
	sideport->onLinkDown(link, this->nodeId, item.node, this->spovnetId );
}

seqnum_t BaseOverlay::sendMessage(const Message* message, const LinkID& link ){

	logging_debug( "baseoverlay is sending data message on link " << link.toString() );

	//
	// get the mapping for this link
	//

	LinkMapping::iterator i = linkMapping.find( link );
	if( i == linkMapping.end() ){
		logging_error( "could not send message. link not found " << link.toString() );
		return -1;
	}

	i->second.markused();

	//
	// check if the link is up yet, if its an autlink queue message
	//

	if( !i->second.linkup ){

		if( i->second.autolink ){
			logging_info( "auto link " << link.toString() << " is not up yet, queueing message" );
			Data data = data_serialize( message );
			const_cast<Message*>(message)->dropPayload();
			i->second.waitingmsg.push_back( new Message(data) );
		} else {
			logging_error("link " << link.toString() << " is not up yet, dropping message" );
		}

		return -1;
	}

	//
	// send the message through the basecomm
	//

	OverlayMsg overmsg( OverlayMsg::OverlayMessageTypeData,	i->second.service, nodeId );
	overmsg.encapsulate( const_cast<Message*>(message) );

	return bc->sendMessage( link, &overmsg );
}

seqnum_t BaseOverlay::sendMessage(const Message* message, const NodeID& node, const ServiceID& service){

	LinkID link = LinkID::UNSPECIFIED;

	LinkMapping::iterator i = linkMapping.begin();
	LinkMapping::iterator iend = linkMapping.end();

	//
	// see if we find a link for this node and service destination
	//

	for( ; i != iend; i++ ){
		if( i->second.node == node && i->second.service == service ){
			link = i->second.link;
			break;
		}
	}

	//
	// if we found no link, create an auto link
	//

	if( link == LinkID::UNSPECIFIED ){

		logging_info( "no link could be found to send message to node " <<
				node.toString() << " for service " << service.toString() <<
				". creating auto link ...");

		// call basecomm to create a link
		link = establishLink( node, service );

		// this will call onlinkup on us, if everything worked we now have a mapping
		LinkMapping::iterator i = linkMapping.find( link );
		i->second.autolink = true;

		if( i == linkMapping.end() || link == LinkID::UNSPECIFIED ){
			logging_error( "failed to establish auto link to node " << node.toString() <<
					" for service " << service.toString() );
			return -1;
		}

		logging_debug( "establishing autolink in progress to node "
				<< node.toString() << " with new link-id " << link.toString() );

	} // if( link != LinkID::UNSPECIFIED )

	assert( link != LinkID::UNSPECIFIED );

	// mark the link as used, as we
	// now send a message through it
	i->second.markused();

	// send the message through the new link. the link may not be functional,
	// but for us there is a link-id so we can send messages through it. if
	// the link is not yet up and the message needs to be cached, this is the
	// task of the BaseCommunication, it will cache and send it later.
	return sendMessage( message, link );
}

const EndpointDescriptor& BaseOverlay::getEndpointDescriptor(const LinkID& link) const {

	return bc->getEndpointDescriptor( link );
}

const EndpointDescriptor& BaseOverlay::getEndpointDescriptor(const NodeID& node) const {

	if( node == nodeId || node == NodeID::UNSPECIFIED )
		return bc->getEndpointDescriptor();

	if( overlayInterface == NULL ){
		logging_error( "overlay interface not set, cannot resolve endpoint" );
		return EndpointDescriptor::UNSPECIFIED;
	}

	// TODO: if this is not a onehop overlay the operation will go asynchronously
	return overlayInterface->resolveNode( node );
}


bool BaseOverlay::bind(CommunicationListener* listener, const ServiceID& sid){
	logging_debug( "binding communication listener " << listener
		<< " on serviceid " << sid.toString() );

	if( communicationListeners.contains( sid ) ){
		logging_error( "some listener already registered for service id "
			<< sid.toString() );
		return false;
	}

	communicationListeners.registerItem( listener, sid );
	return true;
}

bool BaseOverlay::registerSidePort(SideportListener* _sideport){
	sideport = _sideport;
	_sideport->configure( this );
}

bool BaseOverlay::unregisterSidePort(SideportListener* _sideport){
	sideport = &SideportListener::DEFAULT;
}

bool BaseOverlay::unbind(CommunicationListener* listener, const ServiceID& sid){
	logging_debug( "unbinding listener " << listener
		<< " from serviceid " << sid.toString() );

	if( !communicationListeners.contains( sid ) ){
		logging_warn( "cannot unbind listener. no listener registered on service id " << sid.toString() );
		return false;
	}

	if( communicationListeners.get(sid) != listener ){
		logging_warn( "listener bound to service id " << sid.toString()
			<< " is different than listener trying to unbind" );
		return false;
	}

	communicationListeners.unregisterItem( sid );
	return true;
}

bool BaseOverlay::bind(NodeListener* listener){
	logging_debug( "binding node listener " << listener );

	NodeListenerVector::iterator i = find( nodeListeners.begin(), nodeListeners.end(), listener );
	if( i != nodeListeners.end() ){
		logging_warn( "node listener " << listener << " is already bound, cannot bind" );
		return false;
	}

	nodeListeners.push_back( listener );
	return true;
}

bool BaseOverlay::unbind(NodeListener* listener){
	logging_debug( "unbinding node listener " << listener );

	NodeListenerVector::iterator i = find( nodeListeners.begin(), nodeListeners.end(), listener );
	if( i == nodeListeners.end() ){
		logging_warn( "node listener " << listener << " is not bound, cannot unbind" );
		return false;
	}

	nodeListeners.erase( i );
	return true;
}

void BaseOverlay::onLinkUp(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){

	logging_debug( "base overlay received linkup event " + id.toString() );
	// TODO: updateOvlVis( getNodeID(id) );

	//
	// if we get up a link while we are in the
	// join phase and this is the link that
	// we have initiated towards the spovnet owner
	// continue the join process by sending
	// a join request message through the link
	//

	if( state == BaseOverlayStateJoinInitiated && id == initiatorLink){

		logging_info(
			"Join has been initiated by me and the link is now up. " <<
			"sending out join request for SpoVNet " << spovnetId.toString()
		);

		OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeJoinRequest, nodeId );
		JoinRequest joinmsg( spovnetId, nodeId );
		overMsg.encapsulate( &joinmsg );

		state = BaseOverlayStateJoinInitiated; // state remains in JoinInitiated
		bc->sendMessage( id, &overMsg );

		return;

	} // if( state == BaseOverlayStateJoinInitiated && id == initiatorLink)

	//
	// otherwise this is a link initiated by a service
	// then we exchange update messages to exchange the
	// service id and node id for the link. in this case
	// we should have a link mapping for this link. if
	// we have no link mapping this link was initiated by
	// the remote side.
	//

	LinkMapping::iterator i = linkMapping.find( id );

	if( i == linkMapping.end() ){

		LinkItem item (id, NodeID::UNSPECIFIED, ServiceID::UNSPECIFIED, &CommunicationListener::DEFAULT );
		linkMapping.insert( make_pair(id, item) );

	} else {

		logging_debug( "sending out OverlayMessageTypeUpdate" <<
				" for service " << i->second.service.toString() <<
				" with local node id " << nodeId.toString() <<
				" on link " << id.toString() );

		OverlayMsg overMsg(
			OverlayMsg::OverlayMessageTypeUpdate,
			i->second.service,
			nodeId
			);

		bc->sendMessage( id, &overMsg );
		i->second.markused();

	} // if( i == linkMapping.end() )

	// the link is only valid for the service when we receive
	// the OverlayMessageTypeUpdate from the remote node and
	// have the nodeid and serviceid for the link!
}

void BaseOverlay::onLinkDown(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){

	logging_debug( "link went down " << id.toString() );

	//
	// tell the service that the link went
	// down and remove the mapping
	//

	LinkMapping::iterator i = linkMapping.find( id );
	if( i == linkMapping.end() ) {
		// this can also be one of the baseoverlay links that
		// no mapping is stored for. therefore we issue no warning.
		// it can also be a link that has been dropped and the
		// mapping is already deleted in the dropLink function.
		// also, the service notification is issued then in dropLink
		return;
	}

	i->second.interface->onLinkDown( id, i->second.node );
	sideport->onLinkDown( id, this->nodeId, i->second.node, this->spovnetId );

	// delete all queued messages
	if( i->second.waitingmsg.size() > 0 ){

		logging_warn( "dropping link " << id.toString() <<
			" that has " << i->second.waitingmsg.size() << " waiting messages" );

		i->second.deleteWaiting();
	}

	linkMapping.erase( i );
}

void BaseOverlay::onLinkChanged(const LinkID& id, const NetworkLocator* oldlocal, const NetworkLocator* newlocal, const NetworkLocator* oldremote, const NetworkLocator* newremote){

	logging_debug( "link changed " << id.toString() );

	//
	// tell the service that the link changed
	//

	LinkMapping::iterator i = linkMapping.find( id );
	if( i == linkMapping.end() ) return;

	i->second.interface->onLinkChanged( id, i->second.node );
	sideport->onLinkChanged( id, this->nodeId, i->second.node, this->spovnetId );

	// TODO call onLinkQoSChanged?

	i->second.markused();
}

void BaseOverlay::onLinkFail(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){

	logging_debug( "link failed " << id.toString() );

	//
	// tell the service that the link failed
	//

	LinkMapping::iterator i = linkMapping.find( id );
	if( i == linkMapping.end() ) return;

	i->second.interface->onLinkFail( id, i->second.node );
	sideport->onLinkFail( id, this->nodeId, i->second.node, this->spovnetId );

	i->second.markused();
}

void BaseOverlay::onLinkQoSChanged(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote, const QoSParameterSet& qos) {

	logging_debug( "link qos changed " << id.toString() );

	//
	// tell the service that the link qos has changed
	//

	LinkMapping::iterator i = linkMapping.find( id );
	if( i == linkMapping.end() ) return;

	// TODO: convert QoSParameterSet to the LinkProperties properties
	// TODO: currently not in the interface: i->second.interface->onLinkQoSChanged( id, i->second.node, LinkProperties::DEFAULT );

	i->second.markused();
}

bool BaseOverlay::onLinkRequest( const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote ){

	// also see in the receiveMessage function. there the higher layer service
	// is asked whether to accept link requests, but there a basic link association is
	// already built up, so we know the node id
	logging_debug("received link request from " << remote->toString() << ", accepting");
	return true;
}


bool BaseOverlay::receiveMessage(const Message* message,
	const LinkID& link, const NodeID&
	/*the nodeid is invalid in this case! removed var to prevent errors*/ ){

	// decapsulate overlay message
	logging_debug( "receiveMessage: " << message->toString());
	OverlayMsg* overlayMsg = const_cast<Message*>(message)->decapsulate<OverlayMsg>();
	if( overlayMsg == NULL ) return false;

	// mark the link as in action
	LinkMapping::iterator item = linkMapping.find( link );
	if( item != linkMapping.end() ) item->second.markused();

	/* ************************************************************************
	/* handle user date that we forward to the appropriate service using the
	 * service id in the message. as we don't know the class of message that
	 * the service handles, we forward it as a pure Message
	 */
	if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeData) ) {

		logging_debug( "baseoverlay received message of type OverlayMessageTypeData" );

		const ServiceID& service = overlayMsg->getService();
		CommunicationListener* serviceListener = communicationListeners.get( service );

		logging_debug( "received data for service " << service.toString() );

		if( serviceListener != NULL )
			serviceListener->onMessage( overlayMsg, overlayMsg->getSourceNode(), link );

		return true;

	} // if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeData) )

	/* ************************************************************************
	/* Handle spovnet instance join requests
	 */
	else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinRequest) &&
						state == BaseOverlayStateInitiator){

		logging_debug(
			"baseoverlay received message of type OverlayMessageTypeJoinRequest"
		);

		JoinRequest* joinReq = overlayMsg->decapsulate<JoinRequest>();
		logging_info( "received join request for spovnet " <<
					 joinReq->getSpoVNetID().toString() );

		/* make sure that the node actually wants to join
		 * the correct spovnet id that we administrate */
		if( joinReq->getSpoVNetID() != spovnetId ){
			logging_error( "received join request for spovnet we don't handle " <<
					joinReq->getSpoVNetID().toString() );
			return false;
		}

		//
		// only if all services allow the node to join it is allowed
		// using the isJoinAllowed interface security policies can be
		// implemented by higher layer services
		//

		// TODO: here you can implement mechanisms to deny joining of a node
		bool allow = true;

		logging_info( "sending back join reply for spovnet " <<
				spovnetId.toString() << " to node " <<
				overlayMsg->getSourceNode().toString() <<
				". result: " << (allow ? "allowed" : "denied") );

		joiningNodes.push_back( overlayMsg->getSourceNode() );

		//
		// send back our spovnetid, default overlay parameters, join allow
		// result, and ourself as the end-point to bootstrap the overlay against
		//

		assert( overlayInterface != NULL );
		OverlayParameterSet parameters = overlayInterface->getParameters();

		OverlayMsg retmsg( OverlayMsg::OverlayMessageTypeJoinReply, nodeId );
		JoinReply replyMsg( spovnetId, parameters,
					allow, getEndpointDescriptor() );

		retmsg.encapsulate(&replyMsg);
		bc->sendMessage( link, &retmsg );

		return true;

	} // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinRequest)
	  //          && state == BaseOverlayStateInitiator)

	/* ************************************************************************
	 * handle replies to spovnet instance join requests
	 */
	else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinReply) &&
						state == BaseOverlayStateJoinInitiated){

		logging_debug(
			"baseoverlay received message of type OverlayMessageTypeJoinReply");

		JoinReply* replyMsg = overlayMsg->decapsulate<JoinReply>();
		logging_info( "received spovnet join reply" );

		// ensure that we actually wanted to get into the spovnet whose id is
		// in the message
		if( replyMsg->getSpoVNetID() != spovnetId ){
			logging_error( "received spovnet join reply for spovnet " <<
					replyMsg->getSpoVNetID().toString() <<
					" but we wanted to join spovnet " <<
					spovnetId.toString() );

			// state does not change here, maybe the reply does come in later
			return false;
		}

		// if we did not get access to the spovnet notify of the failure and
		// close the link to the initiator
		if( ! replyMsg->getJoinAllowed() ){

			logging_error( "our join request has been denied" );

			bc->dropLink( initiatorLink );
			initiatorLink = LinkID::UNSPECIFIED;
			state = BaseOverlayStateInvalid;

			// inform all registered services of the event
			BOOST_FOREACH( NodeListener* i, nodeListeners ){
				i->onJoinFailed( spovnetId );
			}

			return true;
		}

		logging_info( "join request has been accepted for spovnet " <<
				spovnetId.toString() );

		// if we did get access to the spovnet we try to create the overlay
		// structure as given in the reply message
		overlayInterface = OverlayFactory::create( *this,
				replyMsg->getParam(), nodeId, this );

		if( overlayInterface == NULL ){
			logging_error( "overlay structure not supported" );

			bc->dropLink( initiatorLink );
			initiatorLink = LinkID::UNSPECIFIED;
			state = BaseOverlayStateInvalid;

			// inform all registered services of the event
			BOOST_FOREACH( NodeListener* i, nodeListeners )
				i->onJoinFailed( spovnetId );

			return true;
		}

		/* now start the join process for the overlay. the join process for the
		 * spovnet baseoverlay is now complete. we use the endpoint for overlay
		 * structure bootstrapping that the initiator provided in his reply
		 * message */
		state = BaseOverlayStateCompleted;
		ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN);

		overlayInterface->createOverlay();
		overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() );

		// inform all registered services of the event
		BOOST_FOREACH( NodeListener* i, nodeListeners ){
			i->onJoinCompleted( spovnetId );
		}

		return true;

	} // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinReply) && state == BaseOverlayStateJoinInitiated)


	/* ************************************************************************
     * handle update messages for link establishment
     */
	else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeUpdate) ){

		logging_debug(
			"baseoverlay received message of type OverlayMessageTypeUpdate"
		);

		const NodeID& sourcenode = overlayMsg->getSourceNode();
		const ServiceID& service = overlayMsg->getService();

		// linkmapping for the link available? no-> ignore
		LinkMapping::iterator i = linkMapping.find( link );
		if( i == linkMapping.end() ) {
			logging_warn( "received overlay update message for link " <<
					link.toString() << " for which we have no mapping" );
			return false;
		}

		// update our link mapping information for this link
		bool changed = ( i->second.node != sourcenode ) || ( i->second.service != service );
		i->second.node = sourcenode;
		i->second.service = service;

		// if our link information changed, we send out an update, too
		if( changed ){
			OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeUpdate, i->second.service, nodeId );
			bc->sendMessage( link, &overMsg );
		}

		// set the correct listener service for the linkitem
		// now we can tell the registered service of the linkup event
		if( !communicationListeners.contains( service ) ){
			logging_warn( "linkup event for service that has not been registered" );
			return false;
		}

		CommunicationListener* iface = communicationListeners.get( service );
		if( iface == NULL || iface == &CommunicationListener::DEFAULT ){
			logging_warn( "linkup event for service that has been registered "
				"with a NULL interface" );
			return true;
		}

		i->second.interface = iface;
		i->second.markused();

		// ask the service whether it wants to accept this link
		if( !iface->onLinkRequest(sourcenode) ){

			logging_debug("link " << link.toString() <<
								" has been denied by service " << service.toString() << ", dropping link");

			// prevent onLinkDown calls to the service
			i->second.interface = &CommunicationListener::DEFAULT;
			// drop the link
			dropLink( link );

			return true;
		}

		//
		// link has been accepted, link is now up, send messages out first
		//

		i->second.linkup = true;
		logging_debug("link " << link.toString() <<
						" has been accepted by service " << service.toString() << " and is now up");

		if( i->second.waitingmsg.size() > 0 ){
			logging_info( "sending out queued messages on link " << link.toString() );

			BOOST_FOREACH( Message* msg, i->second.waitingmsg ){
				sendMessage( msg, link );
				delete msg;
			}

			i->second.waitingmsg.clear();
		}

		// call the notification functions
		iface->onLinkUp( link, sourcenode );
		sideport->onLinkUp( link, nodeId, sourcenode, this->spovnetId );

		return true;

	} // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeUpdate) )

	/* ************************************************************************
	 * handle bye messages
	 */
	else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeBye) ) {

		logging_debug( "BaseOverlay received message of type OverlayMessageTypeBye" );
		logging_debug( "Received bye message from " <<
				overlayMsg->getSourceNode().toString() );

		/* if we are the initiator and receive a bye from a node
		 * the node just left. if we are a node and receive a bye
		 * from the initiator, we have to close, too.
		 */
		if( overlayMsg->getSourceNode() == spovnetInitiator ){

			bc->dropLink( initiatorLink );
			initiatorLink = LinkID::UNSPECIFIED;
			state = BaseOverlayStateInvalid;

			logging_fatal( "initiator ended spovnet" );

			// inform all registered services of the event
			BOOST_FOREACH( NodeListener* i, nodeListeners ){
				i->onLeaveFailed( spovnetId );
			}

		} else {
			// a node that said goodbye and we are the initiator don't have to
			// do much here, as the node also will go out of the overlay
			// structure
			logging_info( "node left " << overlayMsg->getSourceNode() );
		}

		return true;

	} // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeBye))

	/* ************************************************************************
	 * handle link request forwarded through the overlay
	 */
	else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeLinkRequest)) {
		LinkRequest* linkReq = overlayMsg->decapsulate<LinkRequest>();
		const ServiceID& service = overlayMsg->getService();
		if (linkReq->isReply()) {

			// find link
			PendingLinkMap::iterator i = pendingLinks.find( linkReq->getNonce() );
			if ( i == pendingLinks.end() ) {
				logging_error( "Nonce not found in link request" );
				return true;
			}

			// debug message
			logging_debug( "LinkRequest reply received. Establishing link "
				<< i->second << " to " << (linkReq->getEndpoint()->toString())
				<< " for service " << service.toString()
				<< " with nonce " << linkReq->getNonce()
			);

			// establishing link
			bc->establishLink( *linkReq->getEndpoint(), i->second );
		} else {
			OverlayMsg overlay_msg( OverlayMsg::OverlayMessageTypeLinkRequest, service, nodeId );
			LinkRequest link_request_msg(
					linkReq->getNonce(), &bc->getEndpointDescriptor(), true );
			overlay_msg.encapsulate( &link_request_msg );

			// debug message
			logging_debug( "Sending LinkRequest reply for link with nonce " <<
					linkReq->getNonce()	);

			// route message back over overlay
			overlayInterface->routeMessage(
				overlayMsg->getSourceNode(), &overlay_msg
			);
		}
	} // if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeLinkRequest))

	/* ************************************************************************
	 * unknown message type ... error!
	 */
	else {

		logging_error( "received message in invalid state! don't know " <<
				"what to do with this message of type " <<
				overlayMsg->getType() );
		return false;

	} // else

	return false;
}

void BaseOverlay::broadcastMessage(Message* message, const ServiceID& service){

	logging_debug( "broadcasting message to all known nodes " <<
			"in the overlay from service " + service.toString() );

	OverlayInterface::NodeList nodes = overlayInterface->getKnownNodes();

	OverlayInterface::NodeList::iterator i = nodes.begin();
	OverlayInterface::NodeList::iterator iend = nodes.end();

	for( ; i != iend; i++ ){
		if( *i == nodeId) continue; // don't send to ourselfs
		sendMessage( message, *i, service );
	}
}

vector<NodeID> BaseOverlay::getOverlayNeighbors() const {
	// the known nodes _can_ also include our
	// node, so we remove ourselfs

	vector<NodeID> nodes = overlayInterface->getKnownNodes();
	vector<NodeID>::iterator i = find( nodes.begin(), nodes.end(), this->nodeId );
	if( i != nodes.end() ) nodes.erase( i );

	return nodes;
}

void BaseOverlay::updateOvlVis( const NodeID& n ) {
	NodeID node = n;
/*	void visShowNodeBubble (
        NETWORK_ID network,
        NodeID& node,
        string label
        );
*/
	using namespace std;

	if (node == nodeId || node.isUnspecified()) return;

	// min/max
	if ( node < min || min.isUnspecified() ) min = node;
	if ( node > max || max.isUnspecified() ) max = node;

	// successor
	if ( succ.isUnspecified() || (node > nodeId && (succ < nodeId || (node-nodeId) < (succ-nodeId))) ) {
		if (!succ.isUnspecified() && node != succ)
			ovl.visDisconnect(ovlId, nodeId, succ, string(""));
		succ = node;
		ovl.visConnect(ovlId, nodeId, succ, string(""));
	}

	// set successor (circle-wrap)
	if (succ.isUnspecified() && !min.isUnspecified()) {
		succ = min;
		ovl.visConnect(ovlId, nodeId, succ, string(""));
	}
}

const NodeID& BaseOverlay::getNodeID(const LinkID& lid) const {

	if( lid == LinkID::UNSPECIFIED ) return nodeId;

	LinkMapping::const_iterator i = linkMapping.find( lid );
	if( i == linkMapping.end() ) return NodeID::UNSPECIFIED;
	else return i->second.node;
}

vector<LinkID> BaseOverlay::getLinkIDs( const NodeID& nid ) const {

	vector<LinkID> linkvector;

	BOOST_FOREACH( LinkPair item, linkMapping ){
		if( item.second.node == nid || nid == NodeID::UNSPECIFIED ){
			linkvector.push_back( item.second.link );
		}
	}

	return linkvector;
}

void BaseOverlay::incomingRouteMessage(Message* msg){
	// gets handled as normal data message
	receiveMessage( msg, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED );
}

void BaseOverlay::onNodeJoin(const NodeID& node){

	JoiningNodes::iterator i = std::find( joiningNodes.begin(), joiningNodes.end(), node );
	if( i == joiningNodes.end() ) return;

	logging_info( "node has successfully joined baseoverlay and overlay structure "
				<< node.toString() );

	joiningNodes.erase( i );
}

void BaseOverlay::eventFunction(){

	list<LinkID> oldlinks;
	time_t now = time(NULL);

	// first gather all the links from linkMapping that need droppin
	// don't directly drop, as the dropLink function affects the
	// linkMapping structure that we are traversing here.
	// drop links after a timeout of 30s

	BOOST_FOREACH( LinkPair item, linkMapping ){
		if( item.second.autolink && difftime(now, item.second.lastuse) > 30)
			oldlinks.push_back( item.first );
	}

	BOOST_FOREACH( const LinkID lnk, oldlinks ) {
		logging_debug( "auto-link " << lnk.toString() << " timed out and is getting dropped" );
		dropLink( lnk );
	}
}

}} // namespace ariba, overlay