StreamConnection.h

Go to the documentation of this file.

/*
 * StreamConnection.h
 *
 * Copyright (C) 2011 IBR, TU Braunschweig
 *
 * Written-by: Johannes Morgenroth <morgenroth@ibr.cs.tu-bs.de>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#ifndef STREAMCONNECTION_H_
#define STREAMCONNECTION_H_


#include "ibrdtn/data/Bundle.h"
#include "ibrdtn/data/Exceptions.h"
#include "ibrdtn/streams/StreamContactHeader.h"
#include "ibrdtn/streams/StreamDataSegment.h"
#include <ibrcommon/thread/Mutex.h>
#include <ibrcommon/thread/MutexLock.h>
#include <ibrcommon/thread/Timer.h>
#include <ibrcommon/Exceptions.h>
#include <ibrcommon/thread/Queue.h>
#include <iostream>
#include <streambuf>
#include <vector>

namespace dtn
{
        namespace streams
        {
                class StreamConnection : public iostream
                {
                public:
                        enum ConnectionShutdownCases
                        {
                                CONNECTION_SHUTDOWN_NOTSET = 0,
                                CONNECTION_SHUTDOWN_IDLE = 1,
                                CONNECTION_SHUTDOWN_ERROR = 2,
                                CONNECTION_SHUTDOWN_SIMPLE_SHUTDOWN = 3,
                                CONNECTION_SHUTDOWN_NODE_TIMEOUT = 4,
                                CONNECTION_SHUTDOWN_PEER_SHUTDOWN = 5
                        };

                        class TransmissionInterruptedException : public ibrcommon::IOException
                        {
                                public:
                                        TransmissionInterruptedException(const dtn::data::Bundle &bundle, const dtn::data::Length &position) throw()
                                         : ibrcommon::IOException("Transmission was interrupted."), _bundle(bundle), _position(position)
                                        {
                                        };

                                        virtual ~TransmissionInterruptedException() throw ()
                                        {
                                        };

                                        const dtn::data::Bundle _bundle;
                                        const dtn::data::Length _position;
                        };

                        class StreamClosedException : public ibrcommon::IOException
                        {
                        public:
                                StreamClosedException(string what = "The stream has been closed.") throw() : IOException(what)
                                {
                                };
                        };

                        class StreamErrorException : public ibrcommon::IOException
                        {
                        public:
                                StreamErrorException(string what = "StreamError") throw() : IOException(what)
                                {
                                };
                        };

                        class StreamShutdownException : public ibrcommon::IOException
                        {
                        public:
                                StreamShutdownException(string what = "Shutdown message received.") throw() : IOException(what)
                                {
                                };
                        };

                        class Callback
                        {
                        public:
                                virtual void eventShutdown(StreamConnection::ConnectionShutdownCases csc) throw () = 0;

                                virtual void eventTimeout() throw () = 0;

                                virtual void eventError() throw () = 0;

                                virtual void eventBundleRefused() throw () = 0;
                                virtual void eventBundleForwarded() throw () = 0;
                                virtual void eventBundleAck(const dtn::data::Length &ack) throw () = 0;

                                virtual void eventConnectionUp(const StreamContactHeader &header) throw () = 0;

                                virtual void eventConnectionDown() throw () = 0;
                        };

                        StreamConnection(StreamConnection::Callback &cb, iostream &stream, const dtn::data::Length buffer_size = 4096);

                        virtual ~StreamConnection();

                        void handshake(const dtn::data::EID &eid, const dtn::data::Timeout &timeout, const dtn::data::Bitset<StreamContactHeader::HEADER_BITS> &flags);

                        void shutdown(ConnectionShutdownCases csc = CONNECTION_SHUTDOWN_SIMPLE_SHUTDOWN);

                        void reject();

                        void keepalive();

                        void enableIdleTimeout(const dtn::data::Timeout &seconds);

                        void setMonitor(bool val);
                        void resetMonitorStats();
                        size_t getMonitorStat(int index);

                private:
                        class StreamBuffer : public std::basic_streambuf<char, std::char_traits<char> >, public ibrcommon::TimerCallback
                        {
                        friend class StreamConnection;
                        public:
                                enum State
                                {
                                        INITIAL = 0,
                                        IDLE = 1,
                                        DATA_AVAILABLE = 2,
                                        DATA_TRANSFER = 3,
                                        SHUTDOWN = 4
                                };

                                StreamBuffer(StreamConnection &conn, iostream &stream, const dtn::data::Length buffer_size = 1024);
                                virtual ~StreamBuffer();

                                const StreamContactHeader handshake(const StreamContactHeader &header);

                                void close();

                                void shutdown(const StreamDataSegment::ShutdownReason reason = StreamDataSegment::MSG_SHUTDOWN_NONE);

                                void reject();

                                void wait();

                                void abort();

                                void keepalive();

                                size_t timeout(ibrcommon::Timer *timer);

                                void enableIdleTimeout(const dtn::data::Timeout &seconds);

                        protected:
                                virtual int sync();
                                virtual std::char_traits<char>::int_type overflow(std::char_traits<char>::int_type = std::char_traits<char>::eof());
                                virtual std::char_traits<char>::int_type underflow();

                        private:
                                bool __good() const;

                                void __error() const;

                                enum timerNames
                                {
                                        TIMER_IN = 1,
                                        TIMER_OUT = 2
                                };

                                enum StateBits
                                {
                                        STREAM_FAILED = 1 << 0,
                                        STREAM_BAD = 1 << 1,
                                        STREAM_EOF = 1 << 2,
                                        STREAM_HANDSHAKE = 1 << 3,
                                        STREAM_SHUTDOWN = 1 << 4,
                                        STREAM_CLOSED = 1 << 5,
                                        STREAM_REJECT = 1 << 6,
                                        STREAM_SKIP = 1 << 7,
                                        STREAM_ACK_SUPPORT = 1 << 8,
                                        STREAM_NACK_SUPPORT = 1 << 9,
                                        STREAM_SOB = 1 << 10,                   // start of bundle
                                        STREAM_TIMER_SUPPORT = 1 << 11
                                };

                                void skipData(dtn::data::Length &size);

                                bool get(const StateBits bit) const;
                                void set(const StateBits bit);
                                void unset(const StateBits bit);

                                const dtn::data::Length _buffer_size;

                                ibrcommon::Mutex _statelock;
                                int _statebits;

                                StreamConnection &_conn;

                                // Input buffer
                                std::vector<char> in_buf_;

                                // Output buffer
                                std::vector<char> out_buf_;

                                ibrcommon::Mutex _sendlock;

                                std::iostream &_stream;

                                dtn::data::Number _recv_size;

                                // this queue contains all sent data segments
                                // they are removed if an ack or nack is received
                                ibrcommon::Queue<StreamDataSegment> _segments;
                                std::queue<StreamDataSegment> _rejected_segments;

                                Length _underflow_data_remain;
                                State _underflow_state;

                                ibrcommon::Timer _idle_timer;

                                // traffic monitoring
                                bool _monitor;
                                std::vector<size_t> _monitor_stats;
                        };

                        void connectionTimeout();

                        void eventShutdown(StreamConnection::ConnectionShutdownCases csc);

                        void eventBundleAck(const dtn::data::Length &ack);
                        void eventBundleRefused();
                        void eventBundleForwarded();

                        StreamConnection::Callback &_callback;

                        dtn::streams::StreamContactHeader _peer;

                        StreamConnection::StreamBuffer _buf;

                        ibrcommon::Mutex _shutdown_reason_lock;
                        ConnectionShutdownCases _shutdown_reason;
                };
        }
}

#endif /* STREAMCONNECTION_H_ */