daemon/src/net/TCPConvergenceLayer.h

Go to the documentation of this file.

/*
 * TCPConvergenceLayer.h
 *
 *  Created on: 05.08.2009
 *      Author: morgenro
 */

#ifndef TCPCONVERGENCELAYER_H_
#define TCPCONVERGENCELAYER_H_

#include "config.h"
#include "Component.h"

#include "core/NodeEvent.h"
#include "core/Node.h"
#include "net/ConvergenceLayer.h"
#include "net/DiscoveryService.h"
#include "net/DiscoveryServiceProvider.h"

#include <ibrdtn/data/Bundle.h>
#include <ibrdtn/data/EID.h>
#include <ibrdtn/data/MetaBundle.h>
#include <ibrdtn/data/Serializer.h>
#include <ibrdtn/streams/StreamConnection.h>
#include <ibrdtn/streams/StreamContactHeader.h>

#include <ibrcommon/net/tcpserver.h>
#include <ibrcommon/net/tcpstream.h>
#include <ibrcommon/net/vinterface.h>
#include <ibrcommon/thread/Queue.h>
#include <memory>

#ifdef WITH_TLS
#include <ibrcommon/net/TLSStream.h>
#endif

using namespace dtn::streams;

namespace dtn
{
        namespace net
        {
                class TCPConvergenceLayer;

                class TCPConnection : public StreamConnection::Callback, public ibrcommon::DetachedThread
                {
                public:
                        TCPConnection(TCPConvergenceLayer &tcpsrv, ibrcommon::tcpstream *stream, const dtn::data::EID &name, const size_t timeout = 10);

                        TCPConnection(TCPConvergenceLayer &tcpsrv, const dtn::core::Node &node, const dtn::data::EID &name, const size_t timeout = 10);

                        virtual ~TCPConnection();

                        virtual void initialize();

                        void shutdown();

                        const StreamContactHeader& getHeader() const;

                        const dtn::core::Node& getNode() const;

                        virtual void eventShutdown(StreamConnection::ConnectionShutdownCases csc);
                        virtual void eventTimeout();
                        virtual void eventError();
                        virtual void eventConnectionUp(const StreamContactHeader &header);
                        virtual void eventConnectionDown();

                        virtual void eventBundleRefused();
                        virtual void eventBundleForwarded();
                        virtual void eventBundleAck(size_t ack);

                        dtn::core::Node::Protocol getDiscoveryProtocol() const;

                        void queue(const dtn::data::BundleID &bundle);

                        bool match(const dtn::core::Node &n) const;
                        bool match(const dtn::data::EID &destination) const;
                        bool match(const dtn::core::NodeEvent &evt) const;

                        friend TCPConnection& operator>>(TCPConnection &conn, dtn::data::Bundle &bundle);
                        friend TCPConnection& operator<<(TCPConnection &conn, const dtn::data::Bundle &bundle);

#ifdef WITH_TLS

                        void enableTLS();
#endif

                protected:
                        void rejectTransmission();

                        void setup();
                        void connect();
                        void run();
                        void finally();
                        void __cancellation();

                        void clearQueue();

                        void keepalive();
                        bool good() const;

                private:
                        class KeepaliveSender : public ibrcommon::JoinableThread
                        {
                        public:
                                KeepaliveSender(TCPConnection &connection, size_t &keepalive_timeout);
                                ~KeepaliveSender();

                                void run();

                                void __cancellation();

                        private:
                                ibrcommon::Conditional _wait;
                                TCPConnection &_connection;
                                size_t &_keepalive_timeout;
                        };

                        class Sender : public ibrcommon::JoinableThread, public ibrcommon::Queue<dtn::data::BundleID>
                        {
                        public:
                                Sender(TCPConnection &connection);
                                virtual ~Sender();

                        protected:
                                void run();
                                void finally();
                                void __cancellation();

                        private:
                                TCPConnection &_connection;
                                //dtn::data::BundleID _current_transfer;
                        };

                        StreamContactHeader _peer;
                        dtn::core::Node _node;

                        std::auto_ptr<ibrcommon::tcpstream> _tcpstream;

#ifdef WITH_TLS
                        ibrcommon::TLSStream _tlsstream;
#endif

                        StreamConnection _stream;

                        // This thread gets awaiting bundles of the queue
                        // and transmit them to the peer.
                        Sender _sender;

                        // Keepalive sender
                        KeepaliveSender _keepalive_sender;

                        // handshake variables
                        dtn::data::EID _name;
                        size_t _timeout;

                        ibrcommon::Queue<dtn::data::MetaBundle> _sentqueue;
                        size_t _lastack;
                        size_t _keepalive_timeout;

                        TCPConvergenceLayer &_callback;

                        /* flags to be used in this nodes StreamContactHeader */
                        char _flags;

                        /* with this boolean the connection is marked as aborted */
                        bool _aborted;
                };

                class TCPConvergenceLayer : public dtn::daemon::IndependentComponent, public ConvergenceLayer, public DiscoveryServiceProvider
                {
                        friend class TCPConnection;
                public:
                        TCPConvergenceLayer();

                        virtual ~TCPConvergenceLayer();

                        void bind(const ibrcommon::vinterface &net, int port);

                        void queue(const dtn::core::Node &n, const ConvergenceLayer::Job &job);

                        void open(const dtn::core::Node &n);

                        virtual const std::string getName() const;

                        dtn::core::Node::Protocol getDiscoveryProtocol() const;

                        void update(const ibrcommon::vinterface &iface, std::string &name, std::string &data) throw(dtn::net::DiscoveryServiceProvider::NoServiceHereException);

                protected:
                        void __cancellation();

                        void componentUp();
                        void componentRun();
                        void componentDown();

                private:
                        void closeAll();

                        void connectionUp(TCPConnection *conn);

                        void connectionDown(TCPConnection *conn);

                        static const int DEFAULT_PORT;
                        bool _running;

                        ibrcommon::tcpserver _tcpsrv;

                        ibrcommon::Conditional _connections_cond;
                        std::list<TCPConnection*> _connections;
                        std::list<ibrcommon::vinterface> _interfaces;
                        std::map<ibrcommon::vinterface, unsigned int> _portmap;
                };
        }
}

#endif /* TCPCONVERGENCELAYER_H_ */