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

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

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

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