close Warning: Can't use blame annotator: 
No changeset 2259 in the repository

source: source/ariba/communication/BaseCommunication.cpp@ 9314

Last change on this file since 9314 was 8620, checked in by Christoph Mayer, 14 years ago

-mem leaks
File size: 22.8 KB
RevLine 
1// [License]
2// The Ariba-Underlay Copyright
3//
4// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH)
5//
6// Institute of Telematics
7// UniversitÀt Karlsruhe (TH)
8// Zirkel 2, 76128 Karlsruhe
9// Germany
10//
11// Redistribution and use in source and binary forms, with or without
12// modification, are permitted provided that the following conditions are
13// met:
14//
15// 1. Redistributions of source code must retain the above copyright
16// notice, this list of conditions and the following disclaimer.
17// 2. Redistributions in binary form must reproduce the above copyright
18// notice, this list of conditions and the following disclaimer in the
19// documentation and/or other materials provided with the distribution.
20//
21// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND
22// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
23// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
24// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR
25// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
26// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
27// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
28// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
29// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
30// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
31// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32//
33// The views and conclusions contained in the software and documentation
34// are those of the authors and should not be interpreted as representing
35// official policies, either expressed or implied, of the Institute of
36// Telematics.
37// [License]
38
39#include "BaseCommunication.h"
40
41#include "networkinfo/AddressDiscovery.h"
42#include "ariba/utility/types/PeerID.h"
43
44#ifdef UNDERLAY_OMNET
45 #include "ariba/communication/modules/transport/omnet/AribaOmnetModule.h"
46 #include "ariba/communication/modules/network/omnet/OmnetNetworkProtocol.h"
47 #include "ariba/utility/system/StartupWrapper.h"
48
49 using ariba::communication::AribaOmnetModule;
50 using ariba::communication::OmnetNetworkProtocol;
51 using ariba::utility::StartupWrapper;
52#endif
53
54namespace ariba {
55namespace communication {
56
57using ariba::utility::PeerID;
58
59use_logging_cpp(BaseCommunication);
60
61/// adds an endpoint to the list
62void BaseCommunication::add_endpoint( const address_v* endpoint ) {
63 if (endpoint==NULL) return;
64 BOOST_FOREACH( endpoint_reference& ref, remote_endpoints ) {
65 if (ref.endpoint->type_id() == endpoint->type_id() && *ref.endpoint == *endpoint) {
66 ref.count++;
67 return;
68 }
69 }
70 endpoint_reference ref;
71 ref.endpoint = endpoint->clone();
72 ref.count = 1;
73 remote_endpoints.push_back(ref);
74}
75
76/// removes an endpoint from the list
77void BaseCommunication::remove_endpoint( const address_v* endpoint ) {
78 if (endpoint==NULL) return;
79 for (vector<endpoint_reference>::iterator i = remote_endpoints.begin();
80 i != remote_endpoints.end(); i++) {
81 if ((*i->endpoint).type_id() == endpoint->type_id() && (*i->endpoint) == *endpoint) {
82 i->count--;
83 if (i->count==0) {
84 logging_info("No more links to " << i->endpoint->to_string() << ": terminating transports!");
85 transport->terminate(i->endpoint);
86 delete i->endpoint;
87 remote_endpoints.erase(i);
88 }
89 return;
90 }
91 }
92}
93
94BaseCommunication::BaseCommunication() {
95 this->transport = NULL;
96 this->started = false;
97}
98
99BaseCommunication::~BaseCommunication(){
100}
101
102void BaseCommunication::start() {
103 logging_info( "Starting up ..." );
104 currentSeqnum = 0;
105
106 // set local peer id
107 localDescriptor.getPeerId() = PeerID::random();
108 logging_info( "Using PeerID: " << localDescriptor.getPeerId() );
109
110 // creating transports
111 logging_info( "Creating transports ..." );
112
113#ifdef UNDERLAY_OMNET
114 AribaOmnetModule* module = StartupWrapper::getCurrentModule();
115 module->setServerPort( listenport );
116
117 transport = module;
118 network = new OmnetNetworkProtocol( module );
119#else
120 transport = new transport_peer( localDescriptor.getEndpoints() );
121#endif
122
123 logging_info( "Searching for local locators ..." );
124 /**
125 * DONT DO THAT: if(localDescriptor.getEndpoints().to_string().length() == 0)
126 * since addresses are used to initialize transport addresses
127 */
128 AddressDiscovery::discover_endpoints( localDescriptor.getEndpoints() );
129 logging_info( "Done. Local endpoints = " << localDescriptor.toString() );
130
131 transport->register_listener( this );
132 transport->start();
133
134#ifndef UNDERLAY_OMNET
135 // bind to the network change detection
136 networkMonitor.registerNotification( this );
137#endif
138
139 // base comm startup done
140 started = true;
141 logging_info( "Started up." );
142}
143
144void BaseCommunication::stop() {
145 logging_info( "Stopping transports ..." );
146
147 transport->stop();
148 delete transport;
149 started = false;
150
151 logging_info( "Stopped." );
152}
153
154bool BaseCommunication::isStarted(){
155 return started;
156}
157
158/// Sets the endpoints
159void BaseCommunication::setEndpoints( string& _endpoints ) {
160 localDescriptor.getEndpoints().assign(_endpoints);
161 logging_info("Setting local end-points: "
162 << localDescriptor.getEndpoints().to_string());
163}
164
165const LinkID BaseCommunication::establishLink(
166 const EndpointDescriptor& descriptor,
167 const LinkID& link_id,
168 const QoSParameterSet& qos,
169 const SecurityParameterSet& sec) {
170
171 // copy link id
172 LinkID linkid = link_id;
173
174 // debug
175 logging_debug( "Request to establish link" );
176
177 // create link identifier
178 if (linkid.isUnspecified()) linkid = LinkID::create();
179
180 // create link descriptor
181 logging_debug( "Creating new descriptor entry with local link id=" << linkid.toString() );
182 LinkDescriptor* ld = new LinkDescriptor();
183 ld->localLink = linkid;
184 addLink( ld );
185
186 // send a message to request new link to remote
187 logging_debug( "Send messages with request to open link to " << descriptor.toString() );
188 AribaBaseMsg baseMsg( AribaBaseMsg::typeLinkRequest, linkid );
189 baseMsg.getLocalDescriptor() = localDescriptor;
190 baseMsg.getRemoteDescriptor().getPeerId() = descriptor.getPeerId();
191
192 // serialize and send message
193 send( &baseMsg, descriptor );
194
195 return linkid;
196}
197
198void BaseCommunication::dropLink(const LinkID link) {
199
200 logging_debug( "Starting to drop link " + link.toString() );
201
202 // see if we have the link
203 LinkDescriptor& ld = queryLocalLink( link );
204 if( ld.isUnspecified() ) {
205 logging_error( "Don't know the link you want to drop "+ link.toString() );
206 return;
207 }
208
209 // tell the registered listeners
210 BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
211 i->onLinkDown( link, ld.localLocator, ld.remoteLocator );
212 }
213
214 // create message to drop the link
215 logging_debug( "Sending out link close request. for us, the link is closed now" );
216 AribaBaseMsg msg( AribaBaseMsg::typeLinkClose, ld.localLink, ld.remoteLink );
217
218 // send message to drop the link
219 send( &msg, ld );
220
221 // remove from map
222 removeLink(link);
223}
224
225seqnum_t BaseCommunication::sendMessage( const LinkID lid, const Message* message) {
226
227 logging_debug( "Sending out message to link " << lid.toString() );
228
229 // query local link info
230 LinkDescriptor& ld = queryLocalLink(lid);
231 if( ld.isUnspecified() ){
232 logging_error( "Don't know the link with id " << lid.toString() );
233 return -1;
234 }
235
236 // link not up-> error
237 if( !ld.up ) {
238 logging_error("Can not send on link " << lid.toString() << ": link not up");
239 return -1;
240 }
241
242 // create message
243 AribaBaseMsg msg( AribaBaseMsg::typeData, ld.localLink, ld.remoteLink );
244
245 // encapsulate the payload message
246 msg.encapsulate( const_cast<Message*>(message) );
247
248 // send message
249 send( &msg, ld );
250
251 // return sequence number
252 return ++currentSeqnum;
253}
254
255const EndpointDescriptor& BaseCommunication::getEndpointDescriptor(const LinkID link) const {
256 if( link.isUnspecified() ){
257 return localDescriptor;
258 } else {
259 LinkDescriptor& linkDesc = queryLocalLink(link);
260 if (linkDesc.isUnspecified()) return EndpointDescriptor::UNSPECIFIED();
261 return linkDesc.remoteEndpoint;
262 }
263}
264
265void BaseCommunication::registerEventListener(CommunicationEvents* _events){
266 if( eventListener.find( _events ) == eventListener.end() )
267 eventListener.insert( _events );
268}
269
270void BaseCommunication::unregisterEventListener(CommunicationEvents* _events){
271 EventListenerSet::iterator i = eventListener.find( _events );
272 if( i != eventListener.end() )
273 eventListener.erase( i );
274}
275
276SystemEventType TransportEvent("Transport");
277SystemEventType MessageDispatchEvent("MessageDispatchEvent", TransportEvent );
278
279class DispatchMsg {
280public:
281 DispatchMsg() : local(NULL), remote(NULL), message(NULL) {}
282 address_v* local;
283 address_v* remote;
284 Message* message;
285};
286
287/// called when a system event is emitted by system queue
288void BaseCommunication::handleSystemEvent(const SystemEvent& event) {
289
290 // dispatch received messages
291 if ( event.getType() == MessageDispatchEvent ){
292 logging_debug( "Forwarding message receiver" );
293 DispatchMsg* dmsg = event.getData<DispatchMsg>();
294 Message* msg = dmsg->message;
295 receiveMessage(msg, dmsg->local, dmsg->remote);
296 msg->dropPayload();
297 delete dmsg->local;
298 delete dmsg->remote;
299 delete msg;
300 delete dmsg;
301 }
302}
303
304/// called when a message is received from transport_peer
305void BaseCommunication::receive_message(transport_protocol* transport,
306 const address_vf local, const address_vf remote, const uint8_t* data,
307 size_t size) {
308
309// logging_debug( "Dispatching message" );
310
311 // convert data
312 Data data_( const_cast<uint8_t*>(data), size * 8 );
313 DispatchMsg* dmsg = new DispatchMsg();
314
315 Message* msg = new Message(data_);
316 dmsg->local = local->clone();
317 dmsg->remote = remote->clone();
318 dmsg->message = msg;
319
320 SystemQueue::instance().scheduleEvent(
321 SystemEvent( this, MessageDispatchEvent, dmsg )
322 );
323}
324
325/// handles a message from the underlay transport
326void BaseCommunication::receiveMessage(const Message* message,
327 const address_v* local, const address_v* remote ){
328
329 /// decapsulate message
330 AribaBaseMsg* msg = ((Message*)message)->decapsulate<AribaBaseMsg>();
331 logging_debug( "Receiving message of type " << msg->getTypeString() );
332
333 // handle message
334 switch (msg->getType()) {
335
336 // ---------------------------------------------------------------------
337 // data message
338 // ---------------------------------------------------------------------
339 case AribaBaseMsg::typeData: {
340 logging_debug( "Received data message, forwarding to overlay" );
341 if( messageReceiver != NULL ) {
342 messageReceiver->receiveMessage(
343 msg, msg->getRemoteLink(), NodeID::UNSPECIFIED
344 );
345 }
346 break;
347 }
348
349 // ---------------------------------------------------------------------
350 // handle link request from remote
351 // ---------------------------------------------------------------------
352 case AribaBaseMsg::typeLinkRequest: {
353 logging_debug( "Received link open request" );
354
355 /// not the correct peer id-> skip request
356 if (!msg->getRemoteDescriptor().getPeerId().isUnspecified()
357 && msg->getRemoteDescriptor().getPeerId() != localDescriptor.getPeerId()) {
358 logging_info("Received link request for "
359 << msg->getRemoteDescriptor().getPeerId().toString()
360 << "but i'm "
361 << localDescriptor.getPeerId()
362 << ": Ignoring!");
363 break;
364 }
365
366 /// only answer the first request
367 if (!queryRemoteLink(msg->getLocalLink()).isUnspecified()) {
368 logging_debug("Link request already received. Ignore!");
369 break;
370 }
371
372 /// create link ids
373 LinkID localLink = LinkID::create();
374 LinkID remoteLink = msg->getLocalLink();
375 logging_debug( "local=" << local->to_string()
376 << " remote=" << remote->to_string()
377 );
378
379 // check if link creation is allowed by ALL listeners
380 bool allowlink = true;
381 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
382 allowlink &= i->onLinkRequest( localLink, local, remote );
383 }
384
385 // not allowed-> warn
386 if( !allowlink ){
387 logging_warn( "Overlay denied creation of link" );
388 delete msg;
389 return;
390 }
391
392 // create descriptor
393 LinkDescriptor* ld = new LinkDescriptor();
394 ld->localLink = localLink;
395 ld->remoteLink = remoteLink;
396 ld->localLocator = local->clone();
397 ld->remoteLocator = remote->clone();
398 ld->remoteEndpoint = msg->getLocalDescriptor();
399 add_endpoint(ld->remoteLocator);
400
401 // add layer 1-3 addresses
402 ld->remoteEndpoint.getEndpoints().add(
403 ld->remoteLocator, endpoint_set::Layer1_3 | endpoint_set::NoLoopback);
404 localDescriptor.getEndpoints().add(
405 local, endpoint_set::Layer1_3 | endpoint_set::NoLoopback);
406
407 // link is now up-> add it
408 ld->up = true;
409 addLink(ld);
410
411 // link is up!
412 logging_debug( "Link (initiated from remote) is up with "
413 << "local(id=" << ld->localLink.toString() << ","
414 << "locator=" << ld->localLocator->to_string() << ") "
415 << "remote(id=" << ld->remoteLink.toString() << ", "
416 << "locator=" << ld->remoteLocator->to_string() << ")"
417 );
418
419 // sending link request reply
420 logging_debug( "Sending link request reply with ids "
421 << "local=" << localLink.toString() << ", "
422 << "remote=" << remoteLink.toString() );
423 AribaBaseMsg reply( AribaBaseMsg::typeLinkReply, localLink, remoteLink );
424 reply.getLocalDescriptor() = localDescriptor;
425 reply.getRemoteDescriptor() = ld->remoteEndpoint;
426
427 send( &reply, *ld );
428
429 // inform listeners about new open link
430 BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
431 i->onLinkUp( localLink, ld->localLocator, ld->remoteLocator);
432 }
433
434 // done
435 break;
436 }
437
438 // ---------------------------------------------------------------------
439 // handle link request reply
440 // ---------------------------------------------------------------------
441 case AribaBaseMsg::typeLinkReply: {
442 logging_debug( "Received link open reply for a link we initiated" );
443
444 // this is a reply to a link open request, so we have already
445 // a link mapping and can now set the remote link to valid
446 LinkDescriptor& ld = queryLocalLink( msg->getRemoteLink() );
447
448 // no link found-> warn!
449 if (ld.isUnspecified()) {
450 logging_warn("Failed to find local link " << msg->getRemoteLink().toString());
451 delete msg;
452 return;
453 }
454
455 // set remote locator and link id
456 ld.remoteLink = msg->getLocalLink();
457 ld.remoteLocator = remote->clone();
458 ld.remoteEndpoint.getEndpoints().add(
459 msg->getLocalDescriptor().getEndpoints(),
460 endpoint_set::Layer1_4
461 );
462
463 localDescriptor.getEndpoints().add(
464 msg->getRemoteDescriptor().getEndpoints(),
465 endpoint_set::Layer1_3
466 );
467 ld.up = true;
468 add_endpoint(ld.remoteLocator);
469
470 logging_debug( "Link is now up with local id "
471 << ld.localLink.toString() << " and remote id "
472 << ld.remoteLink.toString() );
473
474
475 // inform lisneters about link up event
476 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
477 i->onLinkUp( ld.localLink, ld.localLocator, ld.remoteLocator );
478 }
479
480 // done
481 break;
482 }
483
484 // ---------------------------------------------------------------------
485 // handle link close requests
486 // ---------------------------------------------------------------------
487 case AribaBaseMsg::typeLinkClose: {
488 // get remote link
489 const LinkID& localLink = msg->getRemoteLink();
490 logging_debug( "Received link close request for link " << localLink.toString() );
491
492 // searching for link, not found-> warn
493 LinkDescriptor& linkDesc = queryLocalLink( localLink );
494 if (linkDesc.isUnspecified()) {
495 logging_warn("Failed to find local link " << localLink.toString());
496 delete msg;
497 return;
498 }
499
500 // inform listeners
501 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
502 i->onLinkDown( linkDesc.localLink,
503 linkDesc.localLocator, linkDesc.remoteLocator );
504 }
505
506 // remove the link descriptor
507 removeLink( localLink );
508
509 // done
510 break;
511 }
512
513 // ---------------------------------------------------------------------
514 // handle link locator changes
515 // ---------------------------------------------------------------------
516 case AribaBaseMsg::typeLinkUpdate: {
517 const LinkID& localLink = msg->getRemoteLink();
518 logging_debug( "Received link update for link "
519 << localLink.toString() );
520
521 // find the link description
522 LinkDescriptor& linkDesc = queryLocalLink( localLink );
523 if (linkDesc.isUnspecified()) {
524 logging_warn("Failed to update local link "
525 << localLink.toString());
526 delete msg;
527 return;
528 }
529
530 // update the remote locator
531 const address_v* oldremote = linkDesc.remoteLocator;
532 linkDesc.remoteLocator = remote->clone();
533
534 // inform the listeners (local link has _not_ changed!)
535 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
536 i->onLinkChanged(
537 linkDesc.localLink, // linkid
538 linkDesc.localLocator, // old local
539 linkDesc.localLocator, // new local
540 oldremote, // old remote
541 linkDesc.remoteLocator // new remote
542 );
543 }
544
545 // done
546 break;
547 }
548 }
549
550 delete msg;
551}
552
553/// add a newly allocated link to the set of links
554void BaseCommunication::addLink( LinkDescriptor* link ) {
555 linkSet.push_back( link );
556}
557
558/// remove a link from set
559void BaseCommunication::removeLink( const LinkID& localLink ) {
560 for(LinkSet::iterator i=linkSet.begin(); i != linkSet.end(); i++){
561 if( (*i)->localLink != localLink) continue;
562 remove_endpoint((*i)->remoteLocator);
563 delete *i;
564 linkSet.erase( i );
565 break;
566 }
567}
568
569/// query a descriptor by local link id
570BaseCommunication::LinkDescriptor& BaseCommunication::queryLocalLink( const LinkID& link ) const {
571 for (size_t i=0; i<linkSet.size();i++)
572 if (linkSet[i]->localLink == link) return (LinkDescriptor&)*linkSet[i];
573
574 return LinkDescriptor::UNSPECIFIED();
575}
576
577/// query a descriptor by remote link id
578BaseCommunication::LinkDescriptor& BaseCommunication::queryRemoteLink( const LinkID& link ) const {
579 for (size_t i=0; i<linkSet.size();i++)
580 if (linkSet[i]->remoteLink == link) return (LinkDescriptor&)*linkSet[i];
581
582 return LinkDescriptor::UNSPECIFIED();
583}
584
585LinkIDs BaseCommunication::getLocalLinks( const address_v* addr ) const {
586 LinkIDs ids;
587 for (size_t i=0; i<linkSet.size(); i++){
588 if( addr == NULL ){
589 ids.push_back( linkSet[i]->localLink );
590 } else {
591 if ( *linkSet[i]->remoteLocator == *addr )
592 ids.push_back( linkSet[i]->localLink );
593 }
594 }
595 return ids;
596}
597
598void BaseCommunication::onNetworkChange(const NetworkChangeInterface::NetworkChangeInfo& info){
599
600#ifdef UNDERLAY_OMNET
601
602 // we have no mobility support for simulations
603 return
604
605#endif // UNDERLAY_OMNET
606
607/*- disabled!
608
609 // we only care about address changes, not about interface changes
610 // as address changes are triggered by interface changes, we are safe here
611 if( info.type != NetworkChangeInterface::EventTypeAddressNew &&
612 info.type != NetworkChangeInterface::EventTypeAddressDelete ) return;
613
614 logging_info( "base communication is handling network address changes" );
615
616 // get all now available addresses
617 NetworkInformation networkInformation;
618 AddressInformation addressInformation;
619
620 NetworkInterfaceList interfaces = networkInformation.getInterfaces();
621 AddressList addresses;
622
623 for( NetworkInterfaceList::iterator i = interfaces.begin(); i != interfaces.end(); i++ ){
624 AddressList newaddr = addressInformation.getAddresses(*i);
625 addresses.insert( addresses.end(), newaddr.begin(), newaddr.end() );
626 }
627
628 //
629 // get current locators for the local endpoint
630 // TODO: this code is dublicate of the ctor code!!! cleanup!
631 //
632
633 NetworkProtocol::NetworkLocatorSet locators = network->getAddresses();
634 NetworkProtocol::NetworkLocatorSet::iterator i = locators.begin();
635 NetworkProtocol::NetworkLocatorSet::iterator iend = locators.end();
636
637 //
638 // remember the old local endpoint, in case it changes
639 //
640
641 EndpointDescriptor oldLocalDescriptor( localDescriptor );
642
643 //
644 // look for local locators that we can use in communication
645 //
646 // choose the first locator that is not localhost
647 //
648
649 bool foundLocator = false;
650 bool changedLocator = false;
651
652 for( ; i != iend; i++){
653 logging_debug( "local locator found " << (*i)->toString() );
654 IPv4Locator* ipv4locator = dynamic_cast<IPv4Locator*>(*i);
655
656 if( *ipv4locator != IPv4Locator::LOCALHOST &&
657 *ipv4locator != IPv4Locator::ANY &&
658 *ipv4locator != IPv4Locator::BROADCAST ){
659
660 ipv4locator->setPort( listenport );
661 changedLocator = *localDescriptor.locator != *ipv4locator;
662 localDescriptor.locator = ipv4locator;
663 logging_info( "binding to addr = " << ipv4locator->toString() );
664 foundLocator = true;
665 break;
666 }
667 } // for( ; i != iend; i++)
668
669 //
670 // if we found no locator, bind to localhost
671 //
672
673 if( !foundLocator ){
674 changedLocator = *localDescriptor.locator != IPv4Locator::LOCALHOST;
675 localDescriptor.locator = new IPv4Locator( IPv4Locator::LOCALHOST );
676 ((IPv4Locator*)(localDescriptor.locator))->setPort( listenport );
677 logging_info( "found no good local lcoator, binding to addr = " <<
678 localDescriptor.locator->toString() );
679 }
680
681 //
682 // if we have connections that have no more longer endpoints
683 // close these. they will be automatically built up again.
684 // also update the local locator in the linkset mapping
685 //
686
687 if( changedLocator ){
688
689 logging_debug( "local endp locator has changed to " << localDescriptor.toString() <<
690 ", resettings connections that end at old locator " <<
691 oldLocalDescriptor.toString());
692
693 LinkSet::iterator i = linkSet.begin();
694 LinkSet::iterator iend = linkSet.end();
695
696 for( ; i != iend; i++ ){
697
698 logging_debug( "checking connection for locator change: " <<
699 " local " << (*i).localLocator->toString() <<
700 " old " << oldLocalDescriptor.locator->toString() );
701
702 if( *((*i).localLocator) == *(oldLocalDescriptor.locator) ){
703
704 logging_debug("terminating connection to " << (*i).remoteLocator->toString() );
705 transport->terminate( oldLocalDescriptor.locator, (*i).remoteLocator );
706
707 (*i).localLocator = localDescriptor.locator;
708 }
709 } // for( ; i != iend; i++ )
710
711 // wait 500ms to give the sockets time to shut down
712 usleep( 500000 );
713
714 } else {
715
716 logging_debug( "locator has not changed, not resetting connections" );
717
718 }
719
720 //
721 // handle the connections that have no longer any
722 // valid locator. send update messages with the new
723 // locator, so the remote node updates its locator/link mapping
724 //
725
726 LinkSet::iterator iAffected = linkSet.begin();
727 LinkSet::iterator endAffected = linkSet.end();
728
729 for( ; iAffected != endAffected; iAffected++ ){
730 LinkDescriptor descr = *iAffected;
731 logging_debug( "sending out link locator update to " << descr.remoteLocator->toString() );
732
733 AribaBaseMsg updateMsg( descr.remoteLocator,
734 AribaBaseMsg::LINK_STATE_UPDATE,
735 descr.localLink, descr.remoteLink );
736
737 transport->sendMessage( &updateMsg );
738 }
739*/
740}
741
742/// sends a message to all end-points in the end-point descriptor
743void BaseCommunication::send(Message* message, const EndpointDescriptor& endpoint) {
744 Data data = data_serialize( message, DEFAULT_V );
745 transport->send( endpoint.getEndpoints(), data.getBuffer(), data.getLength() / 8);
746 data.release();
747}
748
749/// sends a message to the remote locator inside the link descriptor
750void BaseCommunication::send(Message* message, const LinkDescriptor& desc) {
751 if (desc.remoteLocator==NULL) return;
752 Data data = data_serialize( message, DEFAULT_V );
753 transport->send( desc.remoteLocator, data.getBuffer(), data.getLength() / 8);
754 data.release();
755}
756
757}} // namespace ariba, communication
Note: See TracBrowser for help on using the repository browser.