source: source/ariba/communication/BaseCommunication.h@ 12750

// [License]

// The Ariba-Underlay Copyright

//

// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH)

//

// Institute of Telematics

// UniversitÀt Karlsruhe (TH)

// Zirkel 2, 76128 Karlsruhe

// Germany

//

// Redistribution and use in source and binary forms, with or without

// modification, are permitted provided that the following conditions are

// met:

//

// 1. Redistributions of source code must retain the above copyright

// notice, this list of conditions and the following disclaimer.

// 2. Redistributions in binary form must reproduce the above copyright

// notice, this list of conditions and the following disclaimer in the

// documentation and/or other materials provided with the distribution.

//

// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND

// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE

// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR

// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,

// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,

// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR

// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF

// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING

// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

//

// The views and conclusions contained in the software and documentation

// are those of the authors and should not be interpreted as representing

// official policies, either expressed or implied, of the Institute of

// Telematics.

// [License]



#ifndef BASECOMMUNICATION_H_

#define BASECOMMUNICATION_H_



// boost & std includes

#include <boost/unordered_map.hpp>

#include <boost/unordered_set.hpp>

#include <map>

#include <set>

#include <vector>

#include <iostream>

#include <algorithm>

#include <boost/foreach.hpp>



#ifdef ECLIPSE_PARSER

    #define foreach(a, b) for(a : b)

#else

    #define foreach(a, b) BOOST_FOREACH(a, b)

#endif



// utilities

#include "ariba/utility/types.h"

#include "ariba/utility/messages/MessageReceiver.h"

#include "ariba/utility/logging/Logging.h"

#include "ariba/utility/misc/Demultiplexer.hpp"

#include "ariba/utility/system/SystemEventListener.h"



// new transport and addressing

#include "ariba/utility/transport/transport_peer.hpp"

#include "ariba/utility/transport/interfaces/transport_connection.hpp"

#include "ariba/utility/transport/interfaces/transport_listener.hpp"

#include "ariba/utility/addressing2/endpoint.hpp"



// communication

#include "ariba/communication/CommunicationEvents.h"

#include "ariba/communication/EndpointDescriptor.h"

#include "ariba/communication/messages/AribaBaseMsg.h"



// network changes

#include "ariba/communication/networkinfo/NetworkChangeInterface.h"

#include "ariba/communication/networkinfo/NetworkChangeDetection.h"

#include "ariba/communication/networkinfo/NetworkInformation.h"



namespace ariba {

    class SideportListener;

}



namespace ariba {

namespace communication {





class communication_message_not_sent: public std::runtime_error

{

public:

    /** Takes a character string describing the error.  */

    explicit communication_message_not_sent(const string& __arg)  :

        std::runtime_error(__arg)

    {

    }

    

    virtual ~communication_message_not_sent() throw() {}

};







using namespace std;

using namespace ariba::transport;

using namespace ariba::utility;



// use base ariba types (clarifies multiple definitions)

using ariba::utility::Message;

using ariba::utility::seqnum_t;



/**

 * This class implements the Ariba Base Communication<br />

 *

 * Its primary task is to provide an abstraction to existing

 * protocols and addressing schemes.

 *

 * @author Sebastian Mies, Christoph Mayer, Mario Hock

 */

class BaseCommunication:

        public NetworkChangeInterface,

        public transport_listener {



        use_logging_h(BaseCommunication);

        friend class ariba::SideportListener;



public:

        /// Default ctor that just creates an non-functional base communication

        BaseCommunication();



        /// Default dtor that does nothing

        virtual ~BaseCommunication();



        /// Startup the base communication, start modules etc.

        void start(addressing2::EndpointSetPtr listen_on);



        /// Stops the base communication, stop modules etc.

        void stop();



        /// Check whether the base communication has been started up

        bool isStarted();



        /// Establishes a link to another end-point.

        const LinkID establishLink(const EndpointDescriptor& descriptor,

                const LinkID& linkid = LinkID::UNSPECIFIED, const QoSParameterSet& qos =

                                QoSParameterSet::DEFAULT, const SecurityParameterSet& sec =

                                SecurityParameterSet::DEFAULT);



        /// Drops a link

        void dropLink(const LinkID link);



        /**

         * Sends a message though an existing link to an end-point.

         *

         * @param lid The link identifier

         * @param message The message to be sent

         * @return A sequence number for this message

         */

        seqnum_t sendMessage(const LinkID& lid,

                reboost::message_t message,

                uint8_t priority,

                bool bypass_overlay = false) throw(communication_message_not_sent);



        /**

         * Returns the end-point descriptor

         *

         * @param link the link id of the requested end-point

         * @return The end-point descriptor of the link's end-point

         */

        const EndpointDescriptor& getEndpointDescriptor(const LinkID link =

                        LinkID::UNSPECIFIED) const;



        /**

         * Get local links to the given endpoint of all local link

         * using the default parameter EndpointDescriptor::UNSPECIFIED

         * @param ep The remote endpoint to get all links to.

         * @return List of LinkID

         */

//      LinkIDs getLocalLinks(const address_v* addr) const;  // XXX aktuell



        /**

         * Registers a receiver.

         *

         * @param _receiver The receiving side

         */

        void registerMessageReceiver(MessageReceiver* receiver) {

                messageReceiver = receiver;

        }



        /**

         * Unregister a receiver.

         *

         * @param _receiver The receiving side

         */

        void unregisterMessageReceiver(MessageReceiver* receiver) {

                messageReceiver = NULL;

        }



        void registerEventListener(CommunicationEvents* _events);



        void unregisterEventListener(CommunicationEvents* _events);



        /**

         * called within the ASIO thread

         * when a message is received from underlay transport

         */ 

        virtual void receive_message(transport_connection::sptr connection,

                reboost::shared_buffer_t msg);



    /**

     * called within the ASIO thread

     * when a connection is terminated (e.g. TCP close)

     */ 

    virtual void connection_terminated(transport_connection::sptr connection);



    addressing2::EndpointPtr get_local_endpoint_of_link(const LinkID& linkid);

    addressing2::EndpointPtr get_remote_endpoint_of_link(const LinkID& linkid);



        

protected:



        /**

         * called within the ARIBA thread (System Queue)

         * when a message is received from underlay transport

         */ 

        void receiveMessage(transport_connection::sptr connection,

                reboost::shared_buffer_t message);

        

    /**

     * called within the ARIBA thread (System Queue)

     * when a connection is terminated (e.g. TCP close)

     */ 

    void connectionTerminated(transport_connection::sptr connection);

        



        /// called when a network interface change happens

        virtual void onNetworkChange(

                const NetworkChangeInterface::NetworkChangeInfo& info);



private:

        /**

         * A link descriptor consisting of the end-point descriptor and currently

         * used underlay address.

         */

        class LinkDescriptor {

        public:



                /// default constructor

                LinkDescriptor() :

                        localLink(LinkID::UNSPECIFIED),

                        remoteLink(LinkID::UNSPECIFIED),

                        up(false) {

                }



                ~LinkDescriptor()

                {

                        if ( connection )

                        {

                            connection->unregister_communication_link(&localLink);

                        }

                }



                bool isUnspecified() const {

                        return (this == &UNSPECIFIED());

                }



                static LinkDescriptor& UNSPECIFIED(){

                        static LinkDescriptor* unspec = NULL;

                        if(unspec == NULL) unspec = new LinkDescriptor();

                        return *unspec;

                }

                

                

                transport_connection::sptr get_connection() const

                {

                    return connection;

                }

                

                void set_connection(const transport_connection::sptr& conn)

                {

                    // unregister from old connection,

                    // if any (but normally there shouldn't..)

                    if ( connection )

                    {

                        connection->unregister_communication_link(&localLink);

                    }



                    // * set_connection *

                    connection = conn;

                    

                    // register this link with the connection

                    conn->register_communication_link(&localLink);

                }



                bool unspecified;



                /// link identifiers

                LinkID localLink;

                addressing2::EndpointPtr localLocator;



                /// used underlay addresses for the link

                LinkID remoteLink;

                addressing2::EndpointPtr remoteLocator;



                /// the remote end-point descriptor

                EndpointDescriptor remoteDescriptor;



                /// flag, whether this link is up

                bool up;

                

                

        private:

                /// connection if link is up

                transport_connection::sptr connection;

        };



        /// Link management: list of links

        typedef vector<LinkDescriptor*> LinkSet;



        /// Link management: the set of currently managed links

        LinkSet linkSet;



        /// Link management: add a link

        void addLink( LinkDescriptor* link );



        /// Link management: remove a link

        void removeLink(const LinkID& localLink);



        /// Link management: get link information using the local link

        LinkDescriptor& queryLocalLink(const LinkID& localLink) const;



        /// Link management: get link information using the remote link

        LinkDescriptor& queryRemoteLink(const LinkID& remoteLink) const;



        /// The local end-point descriptor

        EndpointDescriptor localDescriptor;

        

        /**

         * endpoint_set holding the addresses of the "server"-sockets,

         * ---> that should be opened

         * 

         * (e.g. 0.0.0.0:41322)

         */

        addressing2::EndpointSetPtr listenOn_endpoints;

        

    /**

     * endpoint_set holding the addresses of the "server"-sockets,

     * ---> that are actually open

     * 

     * (e.g. 0.0.0.0:41322)

     * 

     * XXX should only be in transport_peer

     */

        addressing2::EndpointSetPtr active_listenOn_endpoints;

        

    /**

     * endpoint_set holding the addresses of the "server"-sockets,

     * ---> here the discovered "addressable" addresses are stored

     * 

     * (e.g. 192.168.0.5:41322)

     * 

     * XXX should only be in localDescriptor

     */

        addressing2::EndpointSetPtr local_endpoints;



        /// network change detector

        NetworkChangeDetection networkMonitor;



        /// list of all remote addresses of links to end-points

        // XXX DEPRECATED

//      class endpoint_reference {

//      public:

//              int count; ///< the number of open links to this end-point

//              const address_v* endpoint; ///< the end-point itself

//      };

//      vector<endpoint_reference> remote_endpoints;



        // XXX DEPRECATED

//      /// adds an end-point to the list

//      void add_endpoint( const address_v* endpoint );

//

//      /// removes an end-point from the list

//      void remove_endpoint( const address_v* endpoint );



        /// event listener

        typedef set<CommunicationEvents*> EventListenerSet;

        EventListenerSet eventListener;



        /// sequence numbers

        seqnum_t currentSeqnum;



        /// transport peer

        transport_peer* transport;



        /// the base overlay message receiver

        MessageReceiver* messageReceiver;



        

        /* 

         * Sends a message over an existing link.

         *   ---> Adds »Link Message« Header

         */

        bool send_over_link(

                const uint8_t type,

                reboost::message_t message,

                const LinkDescriptor& desc,

                const uint8_t priority);



    /* 

     * Sends a message to a known peer. (To all known endpoints.)

     *   ---> Adds »Peer Message« Header

     */

        void send_to_peer(

                const uint8_t type,

                const PeerID& peer_id,

                reboost::message_t message,

                const EndpointDescriptor& endpoint,

                const uint8_t priority );

        



        /// state of the base communication

        bool started;



};



}} // namespace ariba, communication



#endif /* BASECOMMUNICATION_H_ */