Index: /source/ariba/communication/BaseCommunication.cpp =================================================================== --- /source/ariba/communication/BaseCommunication.cpp (revision 3710) +++ /source/ariba/communication/BaseCommunication.cpp (revision 3712) @@ -217,10 +217,4 @@ } - // warn if this link has some queued messages attached - if( descriptor.waitingmsg.size() > 0 ){ - logging_warn( "dropping link " << link.toString() << - " that has " << descriptor.waitingmsg.size() << " waiting messages" ); - } - // create message to drop the link logging_debug( "sending out link close request. for us, the link is closed now" ); @@ -266,19 +260,12 @@ msg.encapsulate( const_cast(message) ); - if( linkDesc.linkup ){ - - // send message - transport->sendMessage( &msg ); - return ++currentSeqnum; - - } else { - - // queue message - logging_info( "link " << lid.toString() << " is not up yet, queueing message" ); - linkDesc.waitingmsg.push_back( new Message(msg) ); // TODO ooooo - - return 0; - - } // if( linkDesc.linkup ) + if( !linkDesc.linkup ){ + logging_error("cant send message on link " << lid.toString() << ", link not up"); + return -1; + } + + // send message + transport->sendMessage( &msg ); + return ++currentSeqnum; } @@ -466,15 +453,4 @@ } - if( linkDesc.waitingmsg.size() > 0 ){ - logging_info( "sending out queued messages on link " << linkDesc.localLink.toString() ); - - BOOST_FOREACH( Message* msg, linkDesc.waitingmsg ){ - sendMessage( linkDesc.localLink, msg ); - delete msg; - } - - linkDesc.waitingmsg.clear(); - } - } // LINK_STATE_OPEN_REPLY @@ -503,14 +479,4 @@ i->onLinkDown( linkDesc.localLink, linkDesc.localLocator, linkDesc.remoteLocator ); } - - // - // delete all pending messages for the link that has been closed - // - - BOOST_FOREACH( Message* msg, linkDesc.waitingmsg ){ - delete msg; - } - - linkDesc.waitingmsg.clear(); // @@ -579,11 +545,5 @@ if( (*i).localLink != localLink) continue; - BOOST_FOREACH( Message* msg, i->waitingmsg ){ - delete msg; - } - - i->waitingmsg.clear(); linkSet.erase( i ); - break; } Index: /source/ariba/communication/BaseCommunication.h =================================================================== --- /source/ariba/communication/BaseCommunication.h (revision 3710) +++ /source/ariba/communication/BaseCommunication.h (revision 3712) @@ -46,5 +46,4 @@ #include #include -#include #include #include @@ -78,5 +77,4 @@ using std::map; using std::vector; -using std::deque; using std::pair; using std::make_pair; @@ -256,8 +254,4 @@ remoteEndpoint(desc.remoteEndpoint), linkup(desc.linkup) { - - BOOST_FOREACH( Message* msg, desc.waitingmsg ){ - waitingmsg.push_back( msg ); - } } @@ -266,12 +260,11 @@ } - LinkID localLink; + LinkID localLink; const NetworkLocator* localLocator; - LinkID remoteLink; + LinkID remoteLink; const NetworkLocator* remoteLocator; - EndpointDescriptor remoteEndpoint; - - bool linkup; - deque waitingmsg; + EndpointDescriptor remoteEndpoint; + + bool linkup; }; Index: /source/ariba/overlay/BaseOverlay.cpp =================================================================== --- /source/ariba/overlay/BaseOverlay.cpp (revision 3710) +++ /source/ariba/overlay/BaseOverlay.cpp (revision 3712) @@ -270,4 +270,5 @@ 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() ); @@ -275,9 +276,20 @@ } + 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 ); - - LinkItem item = i->second; 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 ); @@ -287,4 +299,8 @@ logging_debug( "baseoverlay is sending data message on link " << link.toString() ); + + // + // get the mapping for this link + // LinkMapping::iterator i = linkMapping.find( link ); @@ -294,9 +310,29 @@ } - OverlayMsg overmsg( - OverlayMsg::OverlayMessageTypeData, i->second.service, nodeId ); + 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" ); + i->second.waitingmsg.push_back( new Message(*message) ); + } 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) ); - i->second.markused(); return bc->sendMessage( link, &overmsg ); } @@ -308,4 +344,8 @@ 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++ ){ @@ -316,4 +356,8 @@ } + // + // if we found no link, create an auto link + // + if( link == LinkID::UNSPECIFIED ){ @@ -322,6 +366,10 @@ ". 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 ){ @@ -330,6 +378,4 @@ return -1; } - - i->second.autolink = true; logging_debug( "establishing autolink in progress to node " @@ -531,4 +577,13 @@ 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 ); } @@ -788,7 +843,5 @@ // update our link mapping information for this link - bool changed = - ( i->second.node != sourcenode ) || - ( i->second.service != service ); + bool changed = ( i->second.node != sourcenode ) || ( i->second.service != service ); i->second.node = sourcenode; i->second.service = service; @@ -818,14 +871,5 @@ // ask the service whether it wants to accept this link - if( iface->onLinkRequest(sourcenode) ){ - - logging_debug("link " << link.toString() << - " has been accepted by service " << service.toString()); - - // call the notification functions - iface->onLinkUp( link, sourcenode ); - sideport->onLinkUp( link, nodeId, sourcenode, this->spovnetId ); - - } else { + if( !iface->onLinkRequest(sourcenode) ){ logging_debug("link " << link.toString() << @@ -836,5 +880,30 @@ // 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; Index: /source/ariba/overlay/BaseOverlay.h =================================================================== --- /source/ariba/overlay/BaseOverlay.h (revision 3710) +++ /source/ariba/overlay/BaseOverlay.h (revision 3712) @@ -46,4 +46,5 @@ #include #include +#include #include @@ -80,4 +81,5 @@ using std::pair; using std::find; +using std::deque; using ariba::NodeListener; @@ -420,24 +422,53 @@ LinkItem() : - link(LinkID::UNSPECIFIED), node(NodeID::UNSPECIFIED), service( - ServiceID::UNSPECIFIED), interface( - &CommunicationListener::DEFAULT), autolink(false), lastuse( - 0) { + link(LinkID::UNSPECIFIED), + node(NodeID::UNSPECIFIED), + service(ServiceID::UNSPECIFIED), + interface(&CommunicationListener::DEFAULT), + autolink(false), + lastuse(0), + linkup(false){ } LinkItem(const LinkID& _link, const NodeID& _node, const ServiceID& _service, CommunicationListener* _interface) : - link(_link), node(_node), service(_service), interface(_interface), - autolink(false), lastuse(time(NULL)) { + link(_link), + node(_node), + service(_service), + interface(_interface), + autolink(false), + lastuse(time(NULL)), + linkup(false){ assert( _interface != NULL ); } + LinkItem(const LinkItem& rh) : + link(rh.link), + node(rh.node), + service(rh.service), + interface(rh.interface), + autolink(rh.autolink), + lastuse(rh.lastuse), + linkup(rh.linkup){ + + BOOST_FOREACH( Message* msg, rh.waitingmsg ){ + waitingmsg.push_back( msg ); + } + } + + void deleteWaiting(){ + BOOST_FOREACH( Message* msg, waitingmsg ){ + delete msg; + } + waitingmsg.clear(); + } + // general information about the link - const LinkID link; NodeID node; ServiceID service; CommunicationListener* interface; + bool linkup; // information needed for auto links @@ -448,4 +479,5 @@ bool autolink; time_t lastuse; + deque waitingmsg; }; @@ -467,6 +499,5 @@ }; -} -} // namespace ariba, overlay +}} // namespace ariba, overlay #endif /*BASEOVERLAY_H_*/ Index: /source/ariba/utility/system/StartupWrapper.cpp =================================================================== --- /source/ariba/utility/system/StartupWrapper.cpp (revision 3710) +++ /source/ariba/utility/system/StartupWrapper.cpp (revision 3712) @@ -108,6 +108,10 @@ // set up again an individual level if you like { - // log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseOverlay")); - // logger->setLevel(log4cxx::Level::getInfo()); +// log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseOverlay")); +// logger->setLevel(log4cxx::Level::getDebug()); + } + { +// log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseCommunication")); +// logger->setLevel(log4cxx::Level::getDebug()); }