dataOStream.h

/* Copyright (c) 2007-2017, Stefan Eilemann <eile@equalizergraphics.com>
 *                          Cedric Stalder <cedric.stalder@gmail.com>
 *                          Daniel Nachbaur <danielnachbaur@gmail.com>
 *
 * This file is part of Collage <https://github.com/Eyescale/Collage>
 *
 * 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/global.h>
#include <co/types.h>

#include <lunchbox/array.h> // used inline

#include <boost/noncopyable.hpp>
#include <boost/type_traits.hpp>
#include <map>
#include <set>
#include <unordered_map>
#include <unordered_set>
#include <vector>

namespace co
{
namespace detail
{
class DataOStream;
}
namespace DataStreamTest
{
class Sender;
}

class DataOStream : public boost::noncopyable
{
public:
    CO_API void disable();

    void enableSave();

    void disableSave();

    CO_API bool hasSentData() const;

    CO_API const Connections& getConnections() const;

    DataOStream& streamDataHeader(DataOStream& os);

    void sendBody(ConnectionPtr connection, const void* data,
                  const uint64_t dataSize);

    uint64_t getCompressedDataSize() const;

    template <class T>
    DataOStream& operator<<(const T& value)
    {
        _write(value, boost::has_trivial_copy<T>());
        return *this;
    }

    template <class T>
    DataOStream& operator<<(const Array<T> array)
    {
        _writeArray(array, boost::has_trivial_copy<T>());
        return *this;
    }

    template <class T>
    DataOStream& operator<<(const lunchbox::RefPtr<T>& ptr);

    template <class T>
    DataOStream& operator<<(const lunchbox::Buffer<T>& buffer);

    template <class T>
    DataOStream& operator<<(const lunchbox::Request<T>& request)
    {
        return (*this) << request.getID();
    }

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

    template <class K, class V>
    DataOStream& operator<<(const std::map<K, V>& value);

    template <class T>
    DataOStream& operator<<(const std::set<T>& value);

    template <class K, class V>
    DataOStream& operator<<(const std::unordered_map<K, V>& value);

    template <class T>
    DataOStream& operator<<(const std::unordered_set<T>& value);

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

    CO_API static std::ostream& printStatistics(std::ostream&); 
    CO_API static void clearStatistics();                       

protected:
    CO_API explicit DataOStream(
        size_t chunkSize = Global::getObjectBufferSize()); 
    explicit DataOStream(DataOStream& rhs);                
    virtual CO_API ~DataOStream();                         

    CO_API lunchbox::Bufferb& getBuffer();

    void _setCompressor(const pression::data::CompressorInfo& info);

    CO_API void _enable();

    void flush(const bool last);

    void _setupConnections(const Nodes& receivers);

    void _setupConnections(const Connections& connections);

    void _setupConnection(NodePtr node, const bool useMulticast);

    CO_API void _setupConnection(ConnectionPtr connection);
    friend class DataStreamTest::Sender;

    void _resend();

    void _clearConnections(); 

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

    virtual CO_API void reset();

private:
    detail::DataOStream* const _impl;

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

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

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

    void _resetBuffer();

    template <class 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;
    }

    template <class T>
    void _write(const T& value, const boost::true_type&)
    {
        _write(&value, sizeof(value));
    }

    template <class T>
    void _write(const T& value, const boost::false_type&)
    {
        _writeSerializable(value, boost::is_base_of<servus::Serializable, T>());
    }

    template <class T>
    void _writeSerializable(const T& value, const boost::true_type&);

    template <class T>
    void _writeArray(const Array<T> array, const boost::true_type&)
    {
        _write(array.data, array.getNumBytes());
    }

    template <class T>
    void _writeArray(const Array<T> array, const boost::false_type&)
    {
        for (size_t i = 0; i < array.num; ++i)
            *this << array.data[i];
    }

    void _sendFooter(const void* buffer, const uint64_t size);
};
}

#include "dataOStream.ipp" // template implementation

#endif // CO_DATAOSTREAM_H