dataOStream.h

/* Copyright (c) 2007-2012, Stefan Eilemann <eile@equalizergraphics.com>
 *                    2010, Cedric Stalder <cedric.stalder@gmail.com> 
 *
 * This library is free software; you can redistribute it and/or modify it under
 * the terms of the GNU Lesser General Public License version 2.1 as published
 * by the Free Software Foundation.
 *  
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
 * details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#ifndef CO_DATAOSTREAM_H
#define CO_DATAOSTREAM_H

#include <co/api.h>
#include <co/types.h>
#include <lunchbox/buffer.h> // member
#include <lunchbox/nonCopyable.h> // base class

#include <iostream>
#include <vector>

//#define EQ_INSTRUMENT_DATAOSTREAM

namespace co
{
namespace DataStreamTest
{
    class Sender;
}

    class DataOStream : public lunchbox::NonCopyable
    {
    public:
        CO_API DataOStream();
        virtual CO_API ~DataOStream();

        CO_API void disable();

        void disable( const Packet& packet );

        void enableSave();

        void disableSave();

        bool hasSentData() const { return _dataSent; }

        template< typename T > DataOStream& operator << ( const T& value )
            { write( &value, sizeof( value )); return *this; }

        template< typename T >
        DataOStream& operator << ( const std::vector< T >& value );

        CO_API void write( const void* data, uint64_t size );

        template< typename C >
        void serializeChildren( const std::vector< C* >& children );
 
    protected:
        void _initCompressor( const uint32_t compressor );

        CO_API void _enable();

        void _flush();

        void _setupConnections( const Nodes& receivers );

        void _setupConnection( NodePtr node, const bool useMulticast );

        void _resend();

        void _send( const Packet& packet );

        void _clearConnections() { _connections.clear(); }

        virtual void sendData( const void* buffer, const uint64_t size,
                               const bool last ) = 0;

        template< typename P >
        void sendPacket( P& packet, const void* buffer, const uint64_t size,
                         const bool last );

        virtual CO_API void reset();

    private:        
        enum CompressorState
        {
            STATE_UNCOMPRESSED,
            STATE_PARTIAL,
            STATE_COMPLETE,
            STATE_UNCOMPRESSIBLE
        };
        CompressorState _compressorState;
        
        lunchbox::Bufferb  _buffer;

        uint64_t _bufferStart;

        uint64_t _dataSize;

        Connections _connections;
        friend class DataStreamTest::Sender;

        CPUCompressor* const _compressor;

        bool _enabled;

        bool _dataSent;

        bool _save;

        bool _disable();

        void _sendData( const void* data, const uint64_t size );
        
        void _resetBuffer();

        template< typename T > 
        DataOStream& _writeFlatVector( const std::vector< T >& value )
        {
            const uint64_t nElems = value.size();
            write( &nElems, sizeof( nElems ));
            if( nElems > 0 )
                write( &value.front(), nElems * sizeof( T ));
            return *this;
        }
        void _sendFooter( const void* buffer, const uint64_t size );

        CO_API uint64_t _getCompressedData( void** chunks,
                                            uint64_t* chunkSizes ) const;

        void _compress( void* src, const uint64_t size,
                        const CompressorState result );
    };

    std::ostream& operator << ( std::ostream& os,
                                const DataOStream& dataOStream );

}

#include <co/object.h>
#include <co/objectVersion.h>

namespace co
{
    template<>
    inline DataOStream& DataOStream::operator << ( const std::string& str )
    { 
        const uint64_t nElems = str.length();
        write( &nElems, sizeof( nElems ));
        if ( nElems > 0 )
            write( str.c_str(), nElems );

        return *this;
    }

    template<> inline DataOStream& 
    DataOStream::operator << ( const Object* const& object )
    {
        LBASSERT( !object || object->isAttached( ));
        (*this) << ObjectVersion( object );
        return *this;
    }
 
    template< typename T > inline DataOStream& 
    DataOStream::operator << ( const std::vector< T >& value )
    {
        const uint64_t nElems = value.size();
        (*this) << nElems;
        for( uint64_t i = 0; i < nElems; ++i )
            (*this) << value[i];
        return *this;
    }
 
    template< typename C > inline void
    DataOStream::serializeChildren( const std::vector<C*>& children )
    {
        const uint64_t nElems = children.size();
        (*this) << nElems;

        for( typename std::vector< C* >::const_iterator i = children.begin();
             i != children.end(); ++i )
        {
            C* child = *i;
            (*this) << ObjectVersion( child );
            LBASSERTINFO( !child || child->isAttached(),
                          "Found unmapped object during serialization" );
        }
    }
    template<> inline DataOStream& 
    DataOStream::operator << ( const std::vector< uint8_t >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream& 
    DataOStream::operator << ( const std::vector< uint16_t >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream&
    DataOStream::operator << ( const std::vector< int16_t >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream& 
    DataOStream::operator << ( const std::vector< uint32_t >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream&
    DataOStream::operator << ( const std::vector< int32_t >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream&
    DataOStream::operator << ( const std::vector< uint64_t >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream&
    DataOStream::operator << ( const std::vector< int64_t >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream&
    DataOStream::operator << ( const std::vector< float >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream&
    DataOStream::operator << ( const std::vector< double >& value )
    { return _writeFlatVector( value ); }

    template<> inline DataOStream&
    DataOStream::operator << ( const std::vector< ObjectVersion >& value )
    { return _writeFlatVector( value ); }
}
#endif //CO_DATAOSTREAM_H