source: source/ariba/utility/transport/messages/message.hpp@ 12060

//-----------------------------------------------------------------------------
// Part of reboost (http://reboost.org).  Released under the
// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php).
// Copyright 2012, Sebastian Mies <mies@reboost.org> --- All rights reserved.
//-----------------------------------------------------------------------------

#ifndef REBOOST_MESSAGE_HPP_
#define REBOOST_MESSAGE_HPP_

#include<boost/thread.hpp>
#include<boost/shared_ptr.hpp>
#include<cstring>

#include "shared_buffer.hpp"

namespace reboost {

/// message size type
//typedef signed char mlength_t;  // <--- don't do this!!
//typedef size_t mlength_t;
typedef int mlength_t;  // signed int seems necessary

/// maximum number of buffers per message (default is 8)
const mlength_t message_max_buffers = (1L << 3);
//const mlength_t message_max_buffers = (1L << 4);

//! A Copy-on-Write Message with Shared Buffers.
/**
 * A Copy-on-Write Message with Shared Buffers.
 *
 * A message holds a limited (defined by <code>message_max_buffers</code>)
 * number of shared buffers. One can add new buffers and messages in front and
 * at the end of a message. If the no. of buffers exceed
 * <code>message_max_buffers</code>, then the two smallest successive buffers
 * are compacted to one buffer.
 *
 * @author Sebastian Mies <mies@reboost.org>
 */
class message_t {
private:
        // read sub-message
        struct sub_message {
                message_t* msg;
                inline void operator()(shared_buffer_t buf) {
                        msg->push_back(buf);
                }
        };

        // read from buffer
        struct read_buffer {
                boctet_t* buffer;
                inline void operator()(buffer_t buf) {
                        memcpy((void*) buffer, (void*) buf.data(), buf.size());
                        buffer += buf.size();
                }
        };

        // write to buffer
        struct write_buffer {
                const boctet_t* buffer;
                inline void operator()(buffer_t buf) {
                        memcpy((void*) buf.data(), (void*) buffer, buf.size());
                        buffer += buf.size();
                }
        };

public:
        /// Create a new message
        inline message_t() :
                imsg() {
        }

        /// Copy message
        inline message_t(const message_t& msg) :
                imsg(msg.imsg)
        {
            if ( imsg )
                imsg->owner = NULL;
        }

        /// Linearize message
        inline operator shared_buffer_t() const {
                return linearize();
        }

        /// Assign another message
        inline message_t& operator=(const message_t& msg) {
                msg.imsg->owner = NULL;
                imsg = msg.imsg;
                return *this;
        }

        /// Adds a shared buffer of given site at the end
        inline shared_buffer_t& push_back( bsize_t size ) {
                shared_buffer_t b(size); push_back(b);
                return imsg->at(-1);
        }

        /// Adds a buffer at the end of the message
        inline void push_back(const shared_buffer_t& buf) {
                own().push_back(buf);
        }

        /// Adds a message at the end of the message
        inline void push_back(const message_t& msg) {
                own();
                for (mlength_t i = 0; i < msg.length(); i++)
                        push_back(msg[i]);
        }

        /// Adds a shared buffer of given size at the front
        inline shared_buffer_t& push_front( bsize_t size ) {
                shared_buffer_t b(size); push_front(b);
                return imsg->at(0);
        }

        /// Adds a buffer at the front of the messsage
        inline void push_front(const shared_buffer_t& buf) {
                own().push_front(buf);
        }

        /// Adds a message at the end of the message
        inline void push_front(const message_t& msg) {
                own();
                for (mlength_t i = msg.length() - 1; i != 0; i--)
                        push_front(msg[i]);
        }

        /// Removes a buffer from the end of the message
        inline shared_buffer_t pop_back() {
                return own().pop_back();
        }

        /// Removes a buffer from the beginning of this message.
        inline shared_buffer_t pop_front() {
                return own().pop_front();
        }

        /// Returns the size of the message in bytes (or octets).
        inline size_t size() const {
                size_t s = 0;
                for (mlength_t i = 0; i < length(); i++)
                        s += operator[](i).size();
                return (s);
        }

        /// Returns the number of buffers inside this message.
        inline mlength_t length() const {
            if ( ! imsg )
                return 0;
            
                return (imsg->length);
        }

        /// Returns the buffer at the given index.
        inline shared_buffer_t& operator[](mlength_t idx) {
                return at(idx);
        }

        /// Returns the buffer at the given index.
        inline shared_buffer_t& at(mlength_t idx) {
                return imsg->at(idx);
        }

        /// Returns the constant buffer at the given index.
        inline const shared_buffer_t& operator[](mlength_t idx) const {
                return at(idx);
        }

        /// Returns the buffer at the given index
        inline const shared_buffer_t& at(mlength_t idx) const {
                return imsg->at(idx);
        }

        /// Iterates over a partial set of buffers.
        template<typename T>
        inline void msg_foreach(const T& work, size_t index_ = 0, size_t size_ = 0) const {
                T op = work;
                if (size_ == 0) size_ = size() - index_;

                // get first buffer
                mlength_t f = 0, pf = 0;
                for (; (pf + at(f).size()) <= index_ && f < imsg->length;
                                pf += at(f).size(), f++);
                // get last buffer
                mlength_t l = f, pl = pf;
                for (; (pl + at(l).size()) < (index_ + size_) && l < imsg->length;
                                pl += at(l).size(), l++);

                // same buffer? yes-> get sub-buffer
                if (l == f) op(at(l)(index_ - pf, size_));
                else { // no-> get sub-buffers :)
                        op(at(f)(index_ - pf));
                        for (mlength_t i = f + 1; i < l; i++) op(at(i));
                        op(at(l)(0, index_ + size_ - pl));
                }
        }

        /// Read bytes (gather).
        inline void read(boctet_t* mem, size_t idx = 0, size_t size_ = 0) const {
                struct read_buffer rb = { mem };
                msg_foreach(rb, idx, size_);
        }

        /// write bytes
        inline void write(const boctet_t* mem, size_t idx = 0, size_t size_ = 0) {
                struct write_buffer wb = { mem };
                msg_foreach(wb, idx, size_);
        }

        /// Read an arbitrary, binary object.
        template<class T>
        inline T read(size_t index) {
                T obj;
                read((boctet_t*) &obj, index, sizeof(T));
                return obj;
        }

        /// Write an arbitrary, binary object.
        template<class T>
        inline void write(const T& value, size_t index) {
                write((boctet_t*) &value, index, sizeof(T));
        }

        /// Calculate a (ELF-like) hash.
        inline size_t hash() const {
                size_t h = 0;
                for (mlength_t i = 0; i < length(); i++)
                        h += at(i).hash() * (i + 1);
                return h;
        }

        /// Returns a sub-message.
        message_t operator()(size_t index, size_t size = 0) const {
                message_t m;
                struct sub_message sm = { &m };
                msg_foreach(sm, index, size);
                return m;
        }

        /// Linearizes the complete/partial message into one shared buffer.
        inline shared_buffer_t linearize(size_t index = 0, size_t size_ = 0) const {
                shared_buffer_t b(size_ == 0 ? size() : size_);
                read(b.mutable_data(), index, size_);
                return b;
        }

private:
        class imsg_t {
        public:
                volatile message_t* owner;
                shared_buffer_t buffers[message_max_buffers];
                mlength_t index, length;
        public:
                inline imsg_t() :
                        index(0), length(0) {
                }
                inline imsg_t(const imsg_t& imsg) :
                        index(imsg.index), length(imsg.length) {
                        for (mlength_t i = 0; i < length; i++)
                                at(index + i) = imsg.at(index + i);
                }
                inline shared_buffer_t& at(mlength_t idx) {
                        if (idx < 0) idx += length;
                        return buffers[(idx + index) & (message_max_buffers - 1)];
                }
                inline const shared_buffer_t& at(mlength_t idx) const {
                        if (idx < 0) idx += length;
                        return buffers[(idx + index) & (message_max_buffers - 1)];
                }

                inline void push_back(const shared_buffer_t& buf) {
                        if (buf.size() == 0) return;
                        if (length == message_max_buffers) compact();
                        at(length) = buf;
                        length++;
                }

                inline void push_front(const shared_buffer_t& buf) {
                        if (buf.size() == 0) return;
                        if (length == message_max_buffers) compact();
                        index--;
                        length++;
                        at(0) = buf;
                }

                inline shared_buffer_t pop_back() {
                        shared_buffer_t& buf = at(-1);
                        shared_buffer_t ret = buf;
                        buf.reset();
                        length--;
                        return ret;
                }

                inline shared_buffer_t pop_front() {
                        shared_buffer_t& buf = at(0);
                        shared_buffer_t ret = buf;
                        buf.reset();
                        length--;
                        index++;
                        return ret;
                }

                /// compacts the buffers, so one more buffer is available
                inline void compact() {

                        // find compacting candidate
                        bsize_t min_size=~0, min_pos=0;
                        for (mlength_t i=0; i<length; i++) {
                                bsize_t c = at(i).size() + at(i+1).size();
                                if (c < min_size || min_size == ~(bsize_t)0 ) {
                                        min_size = c;
                                        min_pos = i;
                                }
                        }

                        // compact buffers
                        shared_buffer_t nb(min_size);
                        at(min_pos).copy_to( nb, 0 );
                        at(min_pos+1).copy_to( nb, at(min_pos).size() );

                        // move buffers and assign new buffer
                        for (mlength_t i=min_pos+1; i<length; i++) at(i) = at(i+1);
                        at(min_pos) = nb;

                        length--;
                }
        };
        /// own a new message
        inline imsg_t& own() {
                if (imsg.get() != NULL && imsg->owner == this) return *imsg;
                if (imsg.get() == NULL) imsg = boost::shared_ptr<imsg_t>(new imsg_t());
                else imsg = boost::shared_ptr<imsg_t>(new imsg_t(*imsg));
                imsg->owner = this;
                return *imsg;
        }
        boost::shared_ptr<imsg_t> imsg;
};

inline message_t operator+(const message_t& lhs, const message_t& rhs) {
        message_t m = lhs;
        m.push_back(rhs);
        return m;
}

inline message_t operator+(const message_t& lhs, const buffer_t& rhs) {
        message_t m = lhs;
        m.push_back(rhs);
        return m;
}

inline message_t operator+(const shared_buffer_t& lhs, const message_t& rhs) {
        message_t m = rhs;
        m.push_front(lhs);
        return m;
}

inline message_t operator+(const shared_buffer_t& lhs,
        const shared_buffer_t& rhs) {
        message_t m;
        m.push_back(lhs);
        m.push_back(rhs);
        return m;
}

std::ostream& operator<<(std::ostream&, const message_t);

} /* namespace reboost */

#endif /* REBOOST_MESSAGE_HPP_ */