An Overlay-based
Virtual Network Substrate
Home
Features
Download
Installation
Documentation
Publications
Demos
Team
License
SpoVNet

source: source/ariba/overlay/BaseOverlay.cpp @ 2803

Last change on this file since 2803 was 2803, checked in by Christoph Mayer, 15 years ago

-drop all auto-links when leaving the spovnet
File size: 26.2 KB
Line 
1// [Licence]
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// [Licence]
38
39#include "BaseOverlay.h"
40
41#include "ariba/utility/misc/OvlVis.h"
42using ariba::utility::OvlVis;
43
44namespace ariba {
45namespace overlay {
46
47use_logging_cpp(BaseOverlay);
48
49BaseOverlay::BaseOverlay( BaseCommunication& _basecomm, const NodeID& _nodeid )
50        :       bc( _basecomm ), nodeId( _nodeid ), spovnetId( SpoVNetID::UNSPECIFIED), overlayInterface( NULL ),
51                initiatorLink( LinkID::UNSPECIFIED ), state( BaseOverlayStateInvalid ) {
52
53        logging_info("creating base overlay");
54
55        bc.registerMessageReceiver( this );
56        bc.registerEventListener( this );
57
58        ovl.visCreate( ovlId, nodeId, string(""), string("") );
59        ovl.visChangeNodeColor(ovlId, nodeId, OvlVis::NODE_COLORS_GREY);
60
61//      if (Identifier(Configuration::instance().read<unsigned long>("BASE_nodeid")) ==
62//          Identifier(Configuration::instance().read<unsigned long>("SOURCE"))) {
63//              ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CAMERA);
64//      } else if (Identifier(Configuration::instance().read<unsigned long>("BASE_nodeid")) ==
65//          Identifier(Configuration::instance().read<unsigned long>("MR_A"))) {
66//              ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CHARACTER_A);
67//      } else if (Identifier(Configuration::instance().read<unsigned long>("BASE_nodeid")) ==
68//          Identifier(Configuration::instance().read<unsigned long>("MR_W"))) {
69//              ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CHARACTER_W);
70//      }
71
72        // timer for auto link management
73        Timer::setInterval( 5000 );
74        Timer::start();
75}
76
77BaseOverlay::~BaseOverlay() {
78
79        logging_info("deleting base overlay");
80
81        Timer::stop();
82        bc.unregisterMessageReceiver( this );
83        bc.unregisterEventListener( this );
84}
85
86void BaseOverlay::joinSpoVNet(const SpoVNetID& id, const EndpointDescriptor& bootstrapEp){
87
88        ovl.visShowNodeBubble ( ovlId, nodeId, "joining..." );
89        logging_info( "starting to join spovnet " << id.toString() << "...");
90
91        //
92        // contact the spovnet initiator and request
93        // to join. if the join is granted we will
94        // receive further information on the structure
95        // of the overlay that is used in the spovnet
96        //
97        // but first, we have to establish a link to the initiator...
98        //
99
100        spovnetId = id;
101        state = BaseOverlayStateJoinInitiated;
102
103        initiatorLink = bc.establishLink( bootstrapEp );
104}
105
106void BaseOverlay::leaveSpoVNet(){
107
108        logging_info( "leaving spovnet " << spovnetId );
109
110        logging_debug( "dropping all auto-links ..." );
111       
112        BOOST_FOREACH( LinkPair item, linkMapping ){
113                if( item.second.autolink ) 
114                        dropLink( item.first );
115        }
116
117        logging_debug( "leaving overlay" );
118        // first, leave the overlay interface
119        overlayInterface->leaveOverlay();
120
121        if( state != BaseOverlayStateInitiator ){
122
123                // then, leave the spovnet baseoverlay
124                OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeBye, nodeId );
125                bc.sendMessage( initiatorLink, &overMsg );
126
127                // drop the link and set to correct state
128                bc.dropLink( initiatorLink );
129                initiatorLink = LinkID::UNSPECIFIED;
130        }
131
132        state = BaseOverlayStateInvalid;
133        ovl.visShutdown( ovlId, nodeId, string("") );
134
135        // inform all registered services of the event
136        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
137                i->onOverlayDestroy( spovnetId );
138        }
139}
140
141void BaseOverlay::createSpoVNet(const SpoVNetID& id, const OverlayParameterSet& param, const SecurityParameterSet& sec, const QoSParameterSet& qos){
142
143        //
144        // set the state that we are an initiator,
145        // this way incoming messages are handled correctly
146        //
147
148        logging_info("creating spovnet " + id.toString());
149
150        spovnetId = id;
151        state = BaseOverlayStateInitiator;
152
153        overlayInterface = OverlayFactory::create( *this, param, nodeId, this );
154        if( overlayInterface == NULL ){
155                logging_fatal( "overlay structure not supported" );
156                state = BaseOverlayStateInvalid;
157                return;
158        }
159
160        //
161        // create the overlay
162        //
163
164        overlayInterface->createOverlay();
165        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
166                i->onOverlayCreate( spovnetId );
167        }
168
169        //
170        // bootstrap against ourselfs
171        //
172
173        overlayInterface->joinOverlay();
174        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
175                i->onJoinSuccess( spovnetId );
176        }
177
178        ovl.visChangeNodeIcon ( ovlId, nodeId, OvlVis::ICON_ID_CAMERA );
179        ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN);
180}
181
182const EndpointDescriptor& BaseOverlay::getEndpointDescriptor(const LinkID& link) const {
183
184        return bc.getEndpointDescriptor( link );
185}
186
187const EndpointDescriptor& BaseOverlay::getEndpointDescriptor(const NodeID& node) const {
188
189        if( node == nodeId || node == NodeID::UNSPECIFIED )
190                return bc.getEndpointDescriptor();
191
192        if( overlayInterface == NULL ){
193                logging_error( "overlay interface not set, cannot resolve endpoint" );
194                return EndpointDescriptor::UNSPECIFIED;
195        }
196
197        // TODO: if this is not a onehop overlay the operation will go asynchronously
198        return overlayInterface->resolveNode( node );
199}
200
201const LinkID BaseOverlay::establishLink(const NodeID& node, const ServiceID& service){
202
203        // TODO: if this is not a onehop overlay the operation will go asynchronously
204        const EndpointDescriptor& endpoint = overlayInterface->resolveNode( node );
205        if( endpoint == EndpointDescriptor::UNSPECIFIED ){
206                logging_error( "could not resolve node to endpoint. unable to establish link" );
207                return LinkID::UNSPECIFIED;
208        }
209
210        logging_debug( "baseoverlay called to establish link between node " <<
211                        node.toString() << " on endpoint " << endpoint.toString() <<
212                        " for service " << service.toString() );
213
214        return establishLink( endpoint, service );
215}
216
217const LinkID BaseOverlay::establishLink(const EndpointDescriptor& ep, const ServiceID& service){
218
219        if( !listenerMux.contains( service ) ){
220                logging_error( "no registered listener on serviceid " << service.toString() );
221                return LinkID::UNSPECIFIED;
222        }
223
224        ServiceInterface* receiver = listenerMux.get( service );
225        const LinkID link = bc.establishLink( ep );
226
227        LinkItem item (link, NodeID::UNSPECIFIED, service, receiver);
228        linkMapping.insert( make_pair(link, item) );
229
230        return link;
231}
232
233void BaseOverlay::dropLink(const LinkID& link){
234
235        logging_debug( "baseoverlay dropping link " << link.toString() );
236        LinkMapping::iterator i = linkMapping.find( link );
237
238        if( i == linkMapping.end() ){
239                logging_warn( "can't drop link, mapping unknown " << link.toString() );
240                return;
241        }
242
243        linkMapping.erase( i );
244
245        LinkItem item = i->second;
246        bc.dropLink( link );
247
248        if( item.interface != NULL )
249                item.interface->onLinkDown( spovnetId, nodeId, item.node );
250}
251
252seqnum_t BaseOverlay::sendMessage(const Message* message, const LinkID& link ){
253
254        logging_debug( "baseoverlay is sending message on link " << link.toString() );
255
256        LinkMapping::iterator i = linkMapping.find( link );
257        if( i == linkMapping.end() ){
258                logging_error( "could not send message. link not found " << link.toString() );
259                return -1;
260        }
261
262        OverlayMsg overmsg( OverlayMsg::OverlayMessageTypeData, i->second.service, nodeId );
263        overmsg.encapsulate( const_cast<Message*>(message) );
264
265        i->second.markused();
266        return bc.sendMessage( link, &overmsg );
267}
268
269seqnum_t BaseOverlay::sendMessage(const Message* message, const NodeID& node, const ServiceID& service){
270
271        LinkID link = LinkID::UNSPECIFIED;
272
273        LinkMapping::iterator i = linkMapping.begin();
274        LinkMapping::iterator iend = linkMapping.end();
275
276        for( ; i != iend; i++ ){
277                if( i->second.node == node && i->second.service == service ){
278                        link = i->second.link;
279                        break;
280                }
281        }
282
283        if( link == LinkID::UNSPECIFIED ){
284
285                logging_info( "no link could be found to send message to node " <<
286                                node.toString() << " for service " << service.toString() <<
287                                ". creating auto link ...");
288               
289                const LinkID link = establishLink( node, service );
290                LinkMapping::iterator i = linkMapping.find( link );
291
292                if( i == linkMapping.end() ){
293                        logging_error( "failed to establish auto link to node " << node.toString() <<
294                                        " for service " << service.toString() );
295                        return -1;
296                }
297
298                i->second.autolink = true;
299
300        } // if( link != LinkID::UNSPECIFIED )
301
302        i->second.markused();
303        return sendMessage( message, link );
304}
305
306bool BaseOverlay::bind(ServiceInterface* service, const ServiceID& sid) {
307
308        logging_debug( "binding service " << service << " on serviceid " << sid.toString() );
309
310        if( listenerMux.contains( sid ) ){
311                logging_error( "some service already registered for service id " << sid.toString() );
312                return false;
313        }
314
315        listenerMux.registerItem( service, sid );
316        return true;
317}
318
319ServiceInterface* BaseOverlay::unbind(const ServiceID& sid){
320
321        logging_debug( "unbinding service from serviceid " << sid.toString() );
322
323        if( !listenerMux.contains( sid ) ){
324                logging_warn( "cannot unbind service. no service registered on service id " << sid.toString() );
325                return NULL;
326        }
327
328        ServiceInterface* iface = listenerMux.get( sid );
329        listenerMux.unregisterItem( sid );
330       
331        return NULL; //iface;
332}
333
334void BaseOverlay::onLinkUp(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){
335
336        logging_debug( "base overlay received linkup event " + id.toString() );
337        // TODO: updateOvlVis( getNodeID(id) );
338
339        //
340        // if we get up a link while we are in the
341        // join phase and this is the link that
342        // we have initiated towards the spovnet owner
343        // continue the join process by sending
344        // a join request message through the link
345        //
346
347        if( state == BaseOverlayStateJoinInitiated && id == initiatorLink){
348
349                logging_info( "join has been initiated by me and the link is now up. " <<
350                                "sending out join request for spovnet " <<
351                                spovnetId.toString() );
352
353                OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeJoinRequest, nodeId );
354                JoinRequest joinmsg( spovnetId, nodeId );
355                overMsg.encapsulate( &joinmsg );
356
357                state = BaseOverlayStateJoinInitiated; // state remains in JoinInitiated
358                bc.sendMessage( id, &overMsg );
359
360                return;
361
362        } // if( state == BaseOverlayStateJoinInitiated && id == initiatorLink)
363
364        //
365        // otherwise this is a link initiated by a service
366        // then we exchange update messages to exchange the
367        // service id and node id for the link. in this case
368        // we should have a link mapping for this link. if
369        // we have no link mapping this link was initiated by
370        // the remote side.
371        //
372
373        LinkMapping::iterator i = linkMapping.find( id );
374
375        if( i == linkMapping.end() ){
376
377                LinkItem item (id, NodeID::UNSPECIFIED, ServiceID::UNSPECIFIED, NULL );
378                linkMapping.insert( make_pair(id, item) );
379
380        } else {
381
382                logging_debug( "sending out OverlayMessageTypeUpdate" <<
383                                " for service " << i->second.service.toString() <<
384                                " with local node id " << nodeId.toString() <<
385                                " on link " << id.toString() );
386
387                OverlayMsg overMsg( 
388                        OverlayMsg::OverlayMessageTypeUpdate, 
389                        i->second.service, 
390                        nodeId
391                        );
392
393                bc.sendMessage( id, &overMsg );
394                i->second.markused();
395
396        } // if( i == linkMapping.end() )
397
398        // the link is only valid for the service when we receive
399        // the OverlayMessageTypeUpdate from the remote node and
400        // have the nodeid and serviceid for the link!
401}
402
403void BaseOverlay::onLinkDown(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){
404
405        logging_debug( "link went down " << id.toString() );
406
407        //
408        // tell the service that the link went
409        // down and remove the mapping
410        //
411
412        LinkMapping::iterator i = linkMapping.find( id );
413        if( i == linkMapping.end() ) {
414                // this can also be one of the baseoverlay links that
415                // no mapping is stored for. therefore we issue no warning
416                return;
417        }
418
419        if( i->second.interface != NULL )
420                i->second.interface->onLinkDown( id, nodeId, i->second.node );
421
422        linkMapping.erase( i );
423}
424
425void BaseOverlay::onLinkChanged(const LinkID& id, const NetworkLocator* oldlocal, const NetworkLocator* newlocal, const NetworkLocator* oldremote, const NetworkLocator* newremote){
426
427        logging_debug( "link changed " << id.toString() );
428
429        //
430        // tell the service that the link changed
431        //
432
433        LinkMapping::iterator i = linkMapping.find( id );
434        if( i == linkMapping.end() ) return;
435
436        if( i->second.interface != NULL )
437                i->second.interface->onLinkChanged( id, nodeId, i->second.node );
438
439        i->second.markused();
440}
441
442void BaseOverlay::onLinkFail(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){
443
444        logging_debug( "link failed " << id.toString() );
445
446        //
447        // tell the service that the link failed
448        //
449
450        LinkMapping::iterator i = linkMapping.find( id );
451        if( i == linkMapping.end() ) return;
452
453        if( i->second.interface != NULL )
454                i->second.interface->onLinkFail( id, nodeId, i->second.node );
455
456        i->second.markused();
457}
458
459void BaseOverlay::onLinkQoSChanged(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote, const QoSParameterSet& qos) {
460
461        logging_debug( "link qos changed " << id.toString() );
462
463        //
464        // tell the service that the link qos has changed
465        //
466
467        LinkMapping::iterator i = linkMapping.find( id );
468        if( i == linkMapping.end() ) return;
469
470        if( i->second.interface != NULL )
471                i->second.interface->onLinkQoSChanged( id, nodeId, i->second.node, qos );
472
473        i->second.markused();
474}
475
476bool BaseOverlay::receiveMessage(const Message* message,
477        const LinkID& link, const NodeID& /*the nodeid is invalid in this case! removed var to prevent errors*/ ){
478
479        OverlayMsg* overlayMsg = ((Message*)message)->decapsulate<OverlayMsg>();
480        if( overlayMsg == NULL ) return false;
481
482        // mark the link as in action
483        LinkMapping::iterator item = linkMapping.find( link );
484        if( item != linkMapping.end() ) item->second.markused();
485
486        //
487        // handle user date that we forward to the
488        // appropriate service using the service id
489        // in the message. as we don't know the class
490        // of message that the service handles, we
491        // forward it as a pure Message*
492        //
493
494        if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeData) ) {
495
496                logging_debug( "baseoverlay received message of type OverlayMessageTypeData" );
497
498                const ServiceID& service = overlayMsg->getService();
499                ServiceInterface* serviceListener = listenerMux.get( service );
500
501                logging_debug( "received data for service " << service.toString() );
502
503                if( serviceListener != NULL )
504                        serviceListener->receiveMessage( overlayMsg, link, overlayMsg->getSourceNode() );
505
506                return true;
507
508        } // if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeData) )
509
510        //
511        // handle spovnet instance join requests
512        //
513
514        else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinRequest) &&
515                                                state == BaseOverlayStateInitiator){
516
517                logging_debug( "baseoverlay received message of type OverlayMessageTypeJoinRequest" );
518
519                JoinRequest* joinReq = overlayMsg->decapsulate<JoinRequest>();
520                logging_info( "received join request for spovnet " <<
521                                         joinReq->getSpoVNetID().toString() );
522
523                //
524                // make sure that the node actually wants to join
525                // the correct spovnet id that we administrate
526                //
527
528                if( joinReq->getSpoVNetID() != spovnetId ){
529                        logging_error( "received join request for spovnet we don't handle " <<
530                                        joinReq->getSpoVNetID().toString() );
531                        return false;
532                }
533
534                //
535                // only if all services allow the node to join it is allowed
536                // using the isJoinAllowed interface security policies can be
537                // implemented by higher layer services
538                //
539
540                bool allow = true;
541
542                BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
543                        allow &= i->isJoinAllowed( overlayMsg->getSourceNode(), spovnetId );
544                }
545
546                logging_info( "sending back join reply for spovnet " <<
547                                spovnetId.toString() << " to node " <<
548                                overlayMsg->getSourceNode().toString() <<
549                                ". result: " << (allow ? "allowed" : "denied") );
550
551                joiningNodes.push_back( overlayMsg->getSourceNode() );
552
553                //
554                // send back our spovnetid, default overlay parameters,
555                // join allow result, and ourself as the endpoint
556                // to bootstrap the overlay against
557                //
558
559                OverlayMsg retmsg( OverlayMsg::OverlayMessageTypeJoinReply, nodeId );
560                JoinReply replyMsg( spovnetId, OverlayParameterSet::DEFAULT,
561                                        allow, getEndpointDescriptor() );
562
563                retmsg.encapsulate(&replyMsg);
564                bc.sendMessage( link, &retmsg );
565
566                return true;
567
568        } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinRequest) && state == BaseOverlayStateInitiator)
569
570        //
571        // handle replies to spovnet instance join requests
572        //
573
574        else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinReply) &&
575                                                state == BaseOverlayStateJoinInitiated){
576
577                logging_debug( "baseoverlay received message of type OverlayMessageTypeJoinReply" );
578
579                JoinReply* replyMsg = overlayMsg->decapsulate<JoinReply>();
580                logging_info( "received spovnet join reply" );
581
582                //
583                // make sure that we actually wanted to get
584                // into the spovnet whose id is in the message
585                //
586
587                if( replyMsg->getSpoVNetID() != spovnetId ){
588                        logging_error( "received spovnet join reply for spovnet " <<
589                                        replyMsg->getSpoVNetID().toString() <<
590                                        " but we wanted to join spovnet " <<
591                                        spovnetId.toString() );
592
593                        // state does not change here, maybe
594                        // the reply does come in later
595                        return false;
596                }
597
598                //
599                // if we did not get access to the spovnet
600                // notify of the failure and
601                // close the link to the initiator
602                //
603
604                if( ! replyMsg->getJoinAllowed() ){
605
606                        logging_warn( "our join request has been denied" );
607
608                        bc.dropLink( initiatorLink );
609                        initiatorLink = LinkID::UNSPECIFIED;
610                        state = BaseOverlayStateInvalid;
611
612                        // inform all registered services of the event
613                        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
614                                i->onJoinFail( spovnetId );
615                        }
616
617                        return true;
618                }
619
620                logging_info( "join request has been accepted for spovnet " << spovnetId.toString() );
621
622                //
623                // if we did get access to the spovnet
624                // we try to create the overlay structure
625                // as given in the reply message
626                //
627
628                overlayInterface = OverlayFactory::create( *this, replyMsg->getParam(), nodeId, this );
629
630                if( overlayInterface == NULL ){
631                        logging_error( "overlay structure not supported" );
632
633                        bc.dropLink( initiatorLink );
634                        initiatorLink = LinkID::UNSPECIFIED;
635                        state = BaseOverlayStateInvalid;
636
637                        // inform all registered services of the event
638                        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
639                                i->onJoinFail( spovnetId );
640                        }
641
642                        return true;
643                }
644
645                //
646                // now start the join process for the overlay.
647                // the join process for the spovnet baseoverlay
648                // is now complete. we use the endpoint for
649                // overlay structure bootstrapping that the
650                // initiator provided in his reply message
651                //
652
653                state = BaseOverlayStateCompleted;
654                ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN);
655
656                overlayInterface->createOverlay();
657
658                // inform all registered services of the event
659                BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
660                        i->onOverlayCreate( spovnetId );
661                }
662
663                overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() );
664
665                // inform all registered services of the event
666                BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
667                        i->onJoinSuccess( spovnetId );
668                }
669
670                return true;
671
672        } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinReply) && state == BaseOverlayStateJoinInitiated)
673
674
675        //
676        // handle update messages for link establishment
677        //
678
679        else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeUpdate) ){
680
681                logging_debug( "baseoverlay received message of type OverlayMessageTypeUpdate" );
682
683                const NodeID& sourcenode = overlayMsg->getSourceNode();
684                const ServiceID& service = overlayMsg->getService();
685
686                //
687                // we should have a linkmapping for the link, otherwise
688                // we ignore update messages
689                //
690
691                LinkMapping::iterator i = linkMapping.find( link );
692                if( i == linkMapping.end() ){
693                        logging_warn( "received overlay update message for link " <<
694                                        link.toString() << " for which we have no mapping" );
695                        return false;
696                }
697
698                //
699                // update our link mapping information for this link
700                //
701
702                bool changed = ( i->second.node != sourcenode ) || ( i->second.service != service );
703
704                i->second.node = sourcenode;
705                i->second.service = service;
706
707                //
708                // if our link information changed, we send out an update, too
709                //
710
711                if( changed ){
712                        OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeUpdate, i->second.service, nodeId );
713                        bc.sendMessage( link, &overMsg );
714                }
715
716                //
717                // set the correct listener service for the linkitem
718                // now we can tell the registered service of the linkup event
719                //
720
721                if( !listenerMux.contains( service ) ){
722                        logging_warn( "linkup event for service that has not been registered" );
723                        return false;
724                }
725
726                ServiceInterface* iface = listenerMux.get( service );
727                if( iface == NULL ){
728                        logging_warn( "linkup event for service that has been registered with a NULL interface" );
729                        return true;
730                }
731
732                i->second.interface = iface;
733                iface->onLinkUp( link, nodeId, sourcenode );
734                i->second.markused();
735
736                return true ;
737
738        } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeUpdate) )
739
740        //
741        // bye messages to say goodbye
742        //
743
744        else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeBye)){
745
746                logging_debug( "baseoverlay received message of type OverlayMessageTypeBye" );
747
748                logging_debug( "received bye message from " <<
749                                overlayMsg->getSourceNode().toString() );
750
751                //
752                // if we are the initiator and receive a bye from a node
753                // the node just left. if we are a node and receive a bye
754                // from the initiator, we have to close, too.
755                //
756
757                if( overlayMsg->getSourceNode() == spovnetInitiator ){
758
759                        bc.dropLink( initiatorLink );
760                        initiatorLink = LinkID::UNSPECIFIED;
761                        state = BaseOverlayStateInvalid;
762
763                        logging_fatal( "initiator ended spovnet" );
764
765                        // inform all registered services of the event
766                        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
767                                i->onOverlayDestroy( spovnetId );
768                        }
769
770                } else {
771
772                        // a node that said goodbye and we are the initiator
773                        // don't have to do much here, as the node also
774                        // will go out of the overlay structure
775                        logging_info( "node left " << overlayMsg->getSourceNode() );
776
777                        // inform all registered services of the event
778                        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
779                                i->onNodeLeave( overlayMsg->getSourceNode(), spovnetId );
780                        }
781                }
782
783                return true;
784
785        } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeBye))
786
787        //
788        // something wrong ...
789        //
790
791        else {
792
793                logging_error( "received message in invalid state! don't know " <<
794                                "what to do with this message of type " <<
795                                overlayMsg->getType() );
796                return false;
797
798        } // else
799
800        return false;
801}
802
803void BaseOverlay::broadcastMessage(Message* message, const ServiceID& service){
804
805        logging_debug( "broadcasting message to all known nodes " <<
806                        "in the overlay from service " + service.toString() );
807
808        OverlayInterface::NodeList nodes = overlayInterface->getKnownNodes();
809
810        OverlayInterface::NodeList::iterator i = nodes.begin();
811        OverlayInterface::NodeList::iterator iend = nodes.end();
812
813        for( ; i != iend; i++ ){
814                if( *i == nodeId) continue; // don't send to ourselfs
815                sendMessage( message, *i, service ); // TODO: sollte auto links aufbauen fÃŒr sowas
816        }
817}
818
819void BaseOverlay::updateOvlVis( const NodeID& n ) {
820        NodeID node = n;
821/*      void visShowNodeBubble (
822        NETWORK_ID network,
823        NodeID& node,
824        string label
825        );
826*/
827        using namespace std;
828
829        if (node == nodeId || node.isUnspecified()) return;
830
831        // min/max
832        if ( node < min || min.isUnspecified() ) min = node;
833        if ( node > max || max.isUnspecified() ) max = node;
834
835        // successor
836        if ( succ.isUnspecified() || (node > nodeId && (succ < nodeId || (node-nodeId) < (succ-nodeId))) ) {
837                if (!succ.isUnspecified() && node != succ)
838                        ovl.visDisconnect(ovlId, nodeId, succ, string(""));
839                succ = node;
840                ovl.visConnect(ovlId, nodeId, succ, string(""));
841        }
842
843        // set successor (circle-wrap)
844        if (succ.isUnspecified() && !min.isUnspecified()) {
845                succ = min;
846                ovl.visConnect(ovlId, nodeId, succ, string(""));
847        }
848}
849
850const NodeID& BaseOverlay::getNodeID(const LinkID& lid) const {
851
852        if( lid == LinkID::UNSPECIFIED ) return nodeId;
853
854        LinkMapping::const_iterator i = linkMapping.find( lid );
855        if( i == linkMapping.end() ) return NodeID::UNSPECIFIED;
856        else return i->second.node;
857}
858
859void BaseOverlay::incomingRouteMessage(Message* msg){
860        // gets handled as normal data message
861        // TODO: passt das so?
862        receiveMessage( msg, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED );
863}
864
865void BaseOverlay::onNodeJoin(const NodeID& node){
866
867        JoiningNodes::iterator i = std::find( joiningNodes.begin(), joiningNodes.end(), node );
868        if( i == joiningNodes.end() ) return;
869
870        logging_info( "node has successfully joined baseoverlay and overlay structure "
871                                << node.toString() );
872
873        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
874                i->onNodeJoin( node, spovnetId );
875        }
876
877        joiningNodes.erase( i );
878}
879
880void BaseOverlay::eventFunction(){
881
882        list<LinkID> oldlinks;
883        time_t now = time(NULL);
884
885        // first gather all the links from linkMapping that need droppin
886        // don't directly drop, as the dropLink function affects the
887        // linkMapping structure that we are traversing here.
888        // drop links after a timeout of 30s
889
890        BOOST_FOREACH( LinkPair item, linkMapping ){
891                if( item.second.autolink && difftime(now, item.second.lastuse) > 30)
892                        oldlinks.push_back( item.first );
893        }
894
895        BOOST_FOREACH( const LinkID lnk, oldlinks ){
896                logging_debug( "auto-link " << lnk.toString() << " timed out and is getting dropped" );
897                dropLink( lnk );
898        }
899}
900
901}} // namespace ariba, overlay
Note: See TracBrowser for help on using the repository browser.