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

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

Last change on this file since 5285 was 5284, checked in by mies, 15 years ago

+ added new transport modules and adapted ariba to them
+ exchange endpoint descriptors an link establishment
+ clean up of base communication
+ link establishment with in the presence of multiple endpoints
+ local discovery for ipv6, ipv4 and bluetooth mac addresses
File size: 20.5 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#ifdef UNDERLAY_OMNET
42 #include "ariba/communication/modules/transport/omnet/AribaOmnetModule.h"
43 #include "ariba/communication/modules/network/omnet/OmnetNetworkProtocol.h"
44 #include "ariba/utility/system/StartupWrapper.h"
45
46 using ariba::communication::AribaOmnetModule;
47 using ariba::communication::OmnetNetworkProtocol;
48 using ariba::utility::StartupWrapper;
49#endif
50
51namespace ariba {
52namespace communication {
53
54#include "networkinfo/AddressDiscovery.hpp"
55
56use_logging_cpp(BaseCommunication);
57const BaseCommunication::LinkDescriptor
58 BaseCommunication::LinkDescriptor::UNSPECIFIED;
59
60BaseCommunication::BaseCommunication() {
61 this->transport = NULL;
62 this->started = false;
63}
64
65BaseCommunication::~BaseCommunication(){
66}
67
68void BaseCommunication::start() {
69 logging_info( "Starting up ..." );
70 currentSeqnum = 0;
71
72 // creating transports
73 logging_info( "Creating transports ..." );
74
75#ifdef UNDERLAY_OMNET
76 AribaOmnetModule* module = StartupWrapper::getCurrentModule();
77 module->setServerPort( listenport );
78
79 transport = module;
80 network = new OmnetNetworkProtocol( module );
81#else
82 transport = new transport_peer( localDescriptor.getEndpoints() );
83#endif
84
85 logging_info( "Searching for local locators ..." );
86 discoverEndpoints(localDescriptor.getEndpoints());
87 logging_info( "Done. Local endpoints = " << localDescriptor.toString() );
88
89 transport->register_listener( this );
90 transport->start();
91
92#ifndef UNDERLAY_OMNET
93 // bind to the network change detection
94 networkMonitor.registerNotification( this );
95#endif
96
97 // base comm startup done
98 started = true;
99 logging_info( "Started up." );
100}
101
102void BaseCommunication::stop() {
103 logging_info( "Stopping transports ..." );
104
105 transport->stop();
106 delete transport;
107 started = false;
108
109 logging_info( "Stopped." );
110}
111
112bool BaseCommunication::isStarted(){
113 return started;
114}
115
116/// Sets the endpoints
117void BaseCommunication::setEndpoints( string& _endpoints ) {
118 localDescriptor.getEndpoints().assign(_endpoints);
119 logging_info("Setting local end-points: "
120 << localDescriptor.getEndpoints().to_string());
121}
122
123const LinkID BaseCommunication::establishLink(
124 const EndpointDescriptor& descriptor,
125 const LinkID& link_id,
126 const QoSParameterSet& qos,
127 const SecurityParameterSet& sec) {
128
129 // copy link id
130 LinkID linkid = link_id;
131
132 // debug
133 logging_debug( "Request to establish link" );
134
135 // create link identifier
136 if (linkid.isUnspecified()) linkid = LinkID::create();
137
138 // create link descriptor
139 logging_debug( "Creating new descriptor entry with local link id=" << linkid.toString() );
140 LinkDescriptor* ld = new LinkDescriptor();
141 ld->localLink = linkid;
142 addLink( ld );
143
144 // send a message to request new link to remote
145 logging_debug( "Send messages with request to open link to " << descriptor.toString() );
146 AribaBaseMsg baseMsg( AribaBaseMsg::typeLinkRequest, linkid );
147 baseMsg.getLocalDescriptor() = localDescriptor;
148
149 // serialize and send message
150 send( &baseMsg, descriptor );
151
152 return linkid;
153}
154
155void BaseCommunication::dropLink(const LinkID link) {
156
157 logging_debug( "Starting to drop link " + link.toString() );
158
159 // see if we have the link
160 LinkDescriptor& ld = queryLocalLink( link );
161 if( ld.isUnspecified() ) {
162 logging_error( "Don't know the link you want to drop "+ link.toString() );
163 return;
164 }
165
166 // tell the registered listeners
167 BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
168 i->onLinkDown( link, ld.localLocator, ld.remoteLocator );
169 }
170
171 // create message to drop the link
172 logging_debug( "Sending out link close request. for us, the link is closed now" );
173 AribaBaseMsg msg( AribaBaseMsg::typeLinkClose, ld.localLink, ld.remoteLink );
174
175 // send message to drop the link
176 send( &msg, ld );
177
178 // remove from map
179 removeLink(link);
180}
181
182seqnum_t BaseCommunication::sendMessage( const LinkID lid, const Message* message) {
183
184 logging_debug( "Sending out message to link " << lid.toString() );
185
186 // query local link info
187 LinkDescriptor& ld = queryLocalLink(lid);
188 if( ld.isUnspecified() ){
189 logging_error( "Don't know the link with id " << lid.toString() );
190 return -1;
191 }
192
193 // link not up-> error
194 if( !ld.up ) {
195 logging_error("Can not send on link " << lid.toString() << ": link not up");
196 return -1;
197 }
198
199 // create message
200 AribaBaseMsg msg( AribaBaseMsg::typeData, ld.localLink, ld.remoteLink );
201
202 // encapsulate the payload message
203 msg.encapsulate( const_cast<Message*>(message) );
204
205 // send message
206 send( &msg, ld );
207
208 // return sequence number
209 return ++currentSeqnum;
210}
211
212const EndpointDescriptor& BaseCommunication::getEndpointDescriptor(const LinkID link) const {
213 if( link == LinkID::UNSPECIFIED){
214 return localDescriptor;
215 } else {
216 LinkDescriptor& linkDesc = queryLocalLink(link);
217 if (linkDesc.isUnspecified()) return EndpointDescriptor::UNSPECIFIED;
218 return linkDesc.remoteEndpoint;
219 }
220}
221
222void BaseCommunication::registerEventListener(CommunicationEvents* _events){
223 if( eventListener.find( _events ) == eventListener.end() )
224 eventListener.insert( _events );
225}
226
227void BaseCommunication::unregisterEventListener(CommunicationEvents* _events){
228 EventListenerSet::iterator i = eventListener.find( _events );
229 if( i != eventListener.end() )
230 eventListener.erase( i );
231}
232
233SystemEventType TransportEvent("Transport");
234SystemEventType MessageDispatchEvent("MessageDispatchEvent", TransportEvent );
235
236class DispatchMsg {
237public:
238 address_v* local;
239 address_v* remote;
240 Message* message;
241};
242
243/// called when a system event is emitted by system queue
244void BaseCommunication::handleSystemEvent(const SystemEvent& event) {
245
246 // dispatch received messages
247 if ( event.getType() == MessageDispatchEvent ){
248 logging_debug( "Forwarding message receiver" );
249 DispatchMsg* dmsg = event.getData<DispatchMsg>();
250 Message* msg = dmsg->message;
251 receiveMessage(msg, dmsg->local, dmsg->remote);
252 msg->dropPayload();
253 delete dmsg;
254 delete msg;
255 }
256}
257
258/// called when a message is received form transport_peer
259void BaseCommunication::receive_message(transport_protocol* transport,
260 const address_vf local, const address_vf remote, const uint8_t* data,
261 size_t size) {
262
263// logging_debug( "Dispatching message" );
264
265 // convert data
266 Data data_( const_cast<uint8_t*>(data), size * 8 );
267 DispatchMsg* dmsg = new DispatchMsg();
268
269 Message* msg = new Message(data_);
270 dmsg->local = local->clone();
271 dmsg->remote = remote->clone();
272 dmsg->message = msg;
273
274 SystemQueue::instance().scheduleEvent(
275 SystemEvent( this, MessageDispatchEvent, dmsg )
276 );
277}
278
279/// handles a message from the underlay transport
280void BaseCommunication::receiveMessage(const Message* message,
281 const address_v* local, const address_v* remote ){
282
283 /// decapsulate message
284 AribaBaseMsg* msg = ((Message*)message)->decapsulate<AribaBaseMsg>();
285 logging_debug( "Receiving message of type " << msg->getTypeString() );
286
287 // handle message
288 switch (msg->getType()) {
289
290 // ---------------------------------------------------------------------
291 // data message
292 // ---------------------------------------------------------------------
293 case AribaBaseMsg::typeData: {
294 logging_debug( "Received data message, forwarding to overlay" );
295 if( messageReceiver != NULL ) {
296 messageReceiver->receiveMessage(
297 msg, msg->getRemoteLink(), NodeID::UNSPECIFIED
298 );
299 }
300 break;
301 }
302
303 // ---------------------------------------------------------------------
304 // handle link request from remote
305 // ---------------------------------------------------------------------
306 case AribaBaseMsg::typeLinkRequest: {
307 logging_debug( "Received link open request" );
308
309 /// only answer the first request
310 if (!queryRemoteLink(msg->getLocalLink()).isUnspecified()) {
311 logging_debug("Link request already received. Ignore!");
312 break;
313 }
314
315 /// create link ids
316 LinkID localLink = LinkID::create();
317 LinkID remoteLink = msg->getLocalLink();
318 logging_debug( "local=" << local->to_string()
319 << " remote=" << remote->to_string()
320 );
321
322 // check if link creation is allowed by ALL listeners
323 bool allowlink = true;
324 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
325 allowlink &= i->onLinkRequest( localLink, local, remote );
326 }
327
328 // not allowed-> warn
329 if( !allowlink ){
330 logging_warn( "Overlay denied creation of link" );
331 return;
332 }
333
334 // create descriptor
335 LinkDescriptor* ld = new LinkDescriptor();
336 ld->localLink = localLink;
337 ld->remoteLink = remoteLink;
338 ld->localLocator = local->clone();
339 ld->remoteLocator = remote->clone();
340 ld->remoteEndpoint = msg->getLocalDescriptor();
341
342 // add layer 1-3 addresses
343 ld->remoteEndpoint.getEndpoints().add(
344 ld->remoteLocator, endpoint_set::Layer1_3);
345 localDescriptor.getEndpoints().add(
346 local, endpoint_set::Layer1_3
347 );
348
349 // link is now up-> add it
350 ld->up = true;
351 addLink(ld);
352
353 // link is up!
354 logging_debug( "Link (initiated from remote) is up with "
355 << "local(id=" << ld->localLink.toString() << ","
356 << "locator=" << ld->localLocator->to_string() << ") "
357 << "remote(id=" << ld->remoteLink.toString() << ", "
358 << "locator=" << ld->remoteLocator->to_string() << ")"
359 );
360
361 // sending link request reply
362 logging_debug( "Sending link request reply with ids "
363 << "local=" << localLink.toString() << ", "
364 << "remote=" << remoteLink.toString() );
365 AribaBaseMsg reply( AribaBaseMsg::typeLinkReply, localLink, remoteLink );
366 reply.getLocalDescriptor() = localDescriptor;
367 reply.getRemoteDescriptor() = ld->remoteEndpoint;
368
369 send( &reply, *ld );
370
371 // inform listeners about new open link
372 BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
373 i->onLinkUp( localLink, ld->localLocator, ld->remoteLocator);
374 }
375
376 // done
377 break;
378 }
379
380 // ---------------------------------------------------------------------
381 // handle link request reply
382 // ---------------------------------------------------------------------
383 case AribaBaseMsg::typeLinkReply: {
384 logging_debug( "Received link open reply for a link we initiated" );
385
386 // this is a reply to a link open request, so we have already
387 // a link mapping and can now set the remote link to valid
388 LinkDescriptor& ld = queryLocalLink( msg->getRemoteLink() );
389
390 // no link found-> warn!
391 if (ld.isUnspecified()) {
392 logging_warn("Failed to find local link " << msg->getRemoteLink().toString());
393 return;
394 }
395
396 // set remote locator and link id
397 ld.remoteLink = msg->getLocalLink();
398 ld.remoteLocator = remote->clone();
399 localDescriptor.getEndpoints().add(
400 msg->getRemoteDescriptor().getEndpoints(),
401 endpoint_set::Layer1_3
402 );
403 ld.up = true;
404
405 logging_debug( "Link is now up with local id "
406 << ld.localLink.toString() << " and remote id "
407 << ld.remoteLink.toString() );
408
409
410 // inform lisneters about link up event
411 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
412 i->onLinkUp( ld.localLink, ld.localLocator, ld.remoteLocator );
413 }
414
415 // done
416 break;
417 }
418
419 // ---------------------------------------------------------------------
420 // handle link close requests
421 // ---------------------------------------------------------------------
422 case AribaBaseMsg::typeLinkClose: {
423 // get remote link
424 const LinkID& localLink = msg->getRemoteLink();
425 logging_debug( "Received link close request for link " << localLink.toString() );
426
427 // searching for link, not found-> warn
428 LinkDescriptor& linkDesc = queryLocalLink( localLink );
429 if (linkDesc.isUnspecified()) {
430 logging_warn("Failed to find local link " << localLink.toString());
431 return;
432 }
433
434 // inform listeners
435 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
436 i->onLinkDown( linkDesc.localLink,
437 linkDesc.localLocator, linkDesc.remoteLocator );
438 }
439
440 // remove the link descriptor
441 removeLink( localLink );
442
443 // done
444 break;
445 }
446
447 // ---------------------------------------------------------------------
448 // handle link locator changes
449 // ---------------------------------------------------------------------
450 case AribaBaseMsg::typeLinkUpdate: {
451 const LinkID& localLink = msg->getRemoteLink();
452 logging_debug( "Received link update for link "
453 << localLink.toString() );
454
455 // find the link description
456 LinkDescriptor& linkDesc = queryLocalLink( localLink );
457 if (linkDesc.isUnspecified()) {
458 logging_warn("Failed to update local link "
459 << localLink.toString());
460 return;
461 }
462
463 // update the remote locator
464 const address_v* oldremote = linkDesc.remoteLocator;
465 linkDesc.remoteLocator = remote->clone();
466
467 // inform the listeners (local link has _not_ changed!)
468 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
469 i->onLinkChanged(
470 linkDesc.localLink, // linkid
471 linkDesc.localLocator, // old local
472 linkDesc.localLocator, // new local
473 oldremote, // old remote
474 linkDesc.remoteLocator // new remote
475 );
476 }
477
478 // done
479 break;
480 }
481 }
482}
483
484/// add a newly allocated link to the set of links
485void BaseCommunication::addLink( LinkDescriptor* link ) {
486 linkSet.push_back( link );
487}
488
489/// remove a link from set
490void BaseCommunication::removeLink( const LinkID& localLink ) {
491 for(LinkSet::iterator i=linkSet.begin(); i != linkSet.end(); i++){
492 if( (*i)->localLink != localLink) continue;
493 delete *i;
494 linkSet.erase( i );
495 break;
496 }
497}
498
499/// query a descriptor by local link id
500BaseCommunication::LinkDescriptor& BaseCommunication::queryLocalLink( const LinkID& link ) const {
501 for (int i=0; i<linkSet.size();i++)
502 if (linkSet[i]->localLink == link) return (LinkDescriptor&)*linkSet[i];
503 return (LinkDescriptor&)LinkDescriptor::UNSPECIFIED;
504}
505
506/// query a descriptor by remote link id
507BaseCommunication::LinkDescriptor& BaseCommunication::queryRemoteLink( const LinkID& link ) const {
508 for (int i=0; i<linkSet.size();i++)
509 if (linkSet[i]->remoteLink == link) return (LinkDescriptor&)*linkSet[i];
510 return (LinkDescriptor&)LinkDescriptor::UNSPECIFIED;
511}
512
513LinkIDs BaseCommunication::getLocalLinks( const address_v* addr ) const {
514 LinkIDs ids;
515 for (int i=0; i<linkSet.size(); i++){
516 if( addr == NULL ){
517 ids.push_back( linkSet[i]->localLink );
518 } else {
519 if ( *linkSet[i]->remoteLocator == *addr )
520 ids.push_back( linkSet[i]->localLink );
521 }
522 }
523 return ids;
524}
525
526void BaseCommunication::onNetworkChange(const NetworkChangeInterface::NetworkChangeInfo& info){
527
528#ifdef UNDERLAY_OMNET
529
530 // we have no mobility support for simulations
531 return
532
533#endif // UNDERLAY_OMNET
534
535/*- disabled!
536
537 // we only care about address changes, not about interface changes
538 // as address changes are triggered by interface changes, we are safe here
539 if( info.type != NetworkChangeInterface::EventTypeAddressNew &&
540 info.type != NetworkChangeInterface::EventTypeAddressDelete ) return;
541
542 logging_info( "base communication is handling network address changes" );
543
544 // get all now available addresses
545 NetworkInformation networkInformation;
546 AddressInformation addressInformation;
547
548 NetworkInterfaceList interfaces = networkInformation.getInterfaces();
549 AddressList addresses;
550
551 for( NetworkInterfaceList::iterator i = interfaces.begin(); i != interfaces.end(); i++ ){
552 AddressList newaddr = addressInformation.getAddresses(*i);
553 addresses.insert( addresses.end(), newaddr.begin(), newaddr.end() );
554 }
555
556 //
557 // get current locators for the local endpoint
558 // TODO: this code is dublicate of the ctor code!!! cleanup!
559 //
560
561 NetworkProtocol::NetworkLocatorSet locators = network->getAddresses();
562 NetworkProtocol::NetworkLocatorSet::iterator i = locators.begin();
563 NetworkProtocol::NetworkLocatorSet::iterator iend = locators.end();
564
565 //
566 // remember the old local endpoint, in case it changes
567 //
568
569 EndpointDescriptor oldLocalDescriptor( localDescriptor );
570
571 //
572 // look for local locators that we can use in communication
573 //
574 // choose the first locator that is not localhost
575 //
576
577 bool foundLocator = false;
578 bool changedLocator = false;
579
580 for( ; i != iend; i++){
581 logging_debug( "local locator found " << (*i)->toString() );
582 IPv4Locator* ipv4locator = dynamic_cast<IPv4Locator*>(*i);
583
584 if( *ipv4locator != IPv4Locator::LOCALHOST &&
585 *ipv4locator != IPv4Locator::ANY &&
586 *ipv4locator != IPv4Locator::BROADCAST ){
587
588 ipv4locator->setPort( listenport );
589 changedLocator = *localDescriptor.locator != *ipv4locator;
590 localDescriptor.locator = ipv4locator;
591 logging_info( "binding to addr = " << ipv4locator->toString() );
592 foundLocator = true;
593 break;
594 }
595 } // for( ; i != iend; i++)
596
597 //
598 // if we found no locator, bind to localhost
599 //
600
601 if( !foundLocator ){
602 changedLocator = *localDescriptor.locator != IPv4Locator::LOCALHOST;
603 localDescriptor.locator = new IPv4Locator( IPv4Locator::LOCALHOST );
604 ((IPv4Locator*)(localDescriptor.locator))->setPort( listenport );
605 logging_info( "found no good local lcoator, binding to addr = " <<
606 localDescriptor.locator->toString() );
607 }
608
609 //
610 // if we have connections that have no more longer endpoints
611 // close these. they will be automatically built up again.
612 // also update the local locator in the linkset mapping
613 //
614
615 if( changedLocator ){
616
617 logging_debug( "local endp locator has changed to " << localDescriptor.toString() <<
618 ", resettings connections that end at old locator " <<
619 oldLocalDescriptor.toString());
620
621 LinkSet::iterator i = linkSet.begin();
622 LinkSet::iterator iend = linkSet.end();
623
624 for( ; i != iend; i++ ){
625
626 logging_debug( "checking connection for locator change: " <<
627 " local " << (*i).localLocator->toString() <<
628 " old " << oldLocalDescriptor.locator->toString() );
629
630 if( *((*i).localLocator) == *(oldLocalDescriptor.locator) ){
631
632 logging_debug("terminating connection to " << (*i).remoteLocator->toString() );
633 transport->terminate( oldLocalDescriptor.locator, (*i).remoteLocator );
634
635 (*i).localLocator = localDescriptor.locator;
636 }
637 } // for( ; i != iend; i++ )
638
639 // wait 500ms to give the sockets time to shut down
640 usleep( 500000 );
641
642 } else {
643
644 logging_debug( "locator has not changed, not resetting connections" );
645
646 }
647
648 //
649 // handle the connections that have no longer any
650 // valid locator. send update messages with the new
651 // locator, so the remote node updates its locator/link mapping
652 //
653
654 LinkSet::iterator iAffected = linkSet.begin();
655 LinkSet::iterator endAffected = linkSet.end();
656
657 for( ; iAffected != endAffected; iAffected++ ){
658 LinkDescriptor descr = *iAffected;
659 logging_debug( "sending out link locator update to " << descr.remoteLocator->toString() );
660
661 AribaBaseMsg updateMsg( descr.remoteLocator,
662 AribaBaseMsg::LINK_STATE_UPDATE,
663 descr.localLink, descr.remoteLink );
664
665 transport->sendMessage( &updateMsg );
666 }
667*/
668}
669
670/// sends a message to all end-points in the end-point descriptor
671void BaseCommunication::send(Message* message, const EndpointDescriptor& endpoint) {
672 Data data = data_serialize( message, DEFAULT_V );
673 transport->send( endpoint.getEndpoints(), data.getBuffer(), data.getLength() / 8);
674}
675
676/// sends a message to the remote locator inside the link descriptor
677void BaseCommunication::send(Message* message, const LinkDescriptor& desc) {
678 Data data = data_serialize( message, DEFAULT_V );
679 transport->send( desc.remoteLocator, data.getBuffer(), data.getLength() / 8);
680}
681
682}} // namespace ariba, communication
Note: See TracBrowser for help on using the repository browser.