ChannelBuffer.h

Go to the documentation of this file.

/*
 * Copyright (C) 2012 by
 *   MetraLabs GmbH (MLAB), GERMANY
 * and
 *   Neuroinformatics and Cognitive Robotics Labs (NICR) at TU Ilmenau, GERMANY
 * All rights reserved.
 *
 * Contact: info@mira-project.org
 *
 * Commercial Usage:
 *   Licensees holding valid commercial licenses may use this file in
 *   accordance with the commercial license agreement provided with the
 *   software or, alternatively, in accordance with the terms contained in
 *   a written agreement between you and MLAB or NICR.
 *
 * GNU General Public License Usage:
 *   Alternatively, this file may be used under the terms of the GNU
 *   General Public License version 3.0 as published by the Free Software
 *   Foundation and appearing in the file LICENSE.GPL3 included in the
 *   packaging of this file. Please review the following information to
 *   ensure the GNU General Public License version 3.0 requirements will be
 *   met: http://www.gnu.org/copyleft/gpl.html.
 *   Alternatively you may (at your option) use any later version of the GNU
 *   General Public License if such license has been publicly approved by
 *   MLAB and NICR (or its successors, if any).
 *
 * IN NO EVENT SHALL "MLAB" OR "NICR" BE LIABLE TO ANY PARTY FOR DIRECT,
 * INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF
 * THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF "MLAB" OR
 * "NICR" HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * "MLAB" AND "NICR" SPECIFICALLY DISCLAIM ANY WARRANTIES, INCLUDING,
 * BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 * FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND "MLAB" AND "NICR" HAVE NO OBLIGATION TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS OR MODIFICATIONS.
 */

#ifndef _MIRA_CHANNELBUFFER_H_
#define _MIRA_CHANNELBUFFER_H_

#include <list>
#include <string>
#include <type_traits>

#ifndef Q_MOC_RUN
#include <boost/noncopyable.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/shared_mutex.hpp>
#endif

#include <error/LoggingCore.h>

#include <factory/TypeId.h>

#include <serialization/BinarySerializer.h>
#include <serialization/JSONSerializer.h>
#include <serialization/MetaSerializer.h>
#include <serialization/XMLSerializer.h>

#include <platform/Typename.h>
#include <utils/Stamped.h>
#include <utils/Time.h>

#include <math/Eigen.h>

namespace mira {


template<typename TargetType>
struct ChannelBufferPromoterCommon;

template<typename TargetType>
struct ChannelBufferPromoter;


enum SlotQueryMode
{
    OLDER_SLOT = 0,

    NEWER_SLOT,

    NEAREST_SLOT
};

enum IntervalFillMode
{
    PREFER_OLDER = 0,
    PREFER_NEWER
};


class ChannelBufferBase : boost::noncopyable
{
    template<typename TargetType>
    friend struct ChannelBufferPromoterCommon;

    template<typename TargetType>
    friend struct ChannelBufferPromoter;

public:
    struct Slot;

    struct ListItem : boost::noncopyable
    {
        ListItem()
        {
            // creates empty list
            next = (Slot*)this;
            prev = (Slot*)this;
        }

        // both will be locked by mutex in ChannelBuffer
        Slot* next;
        Slot* prev;
    };

    struct Slot : public ListItem
    {
        boost::shared_mutex lock; // locks the following data

        boost::mutex serializedValueLock;

        const Time* timestampPtr;

        Buffer<uint8> serializedValue;

        uint32 flags;

        const Time& timestamp() const
        {
            assert(timestampPtr != nullptr); // pointer is set in ChannelBuffers allocateSlot
            return *timestampPtr;
        }

        // required to allow dynamic_cast for subclasses
        virtual ~Slot() = default;
    };

public:
    ChannelBufferBase() = default;

    virtual ~ChannelBufferBase();

public:
    void cloneParameters(const ChannelBufferBase& other)
    {
        mMinSlots = other.mMinSlots;
        mMaxSlots = other.mMaxSlots;
        mStorageDuration = other.mStorageDuration;
        mAutoIncreaseStorageDuration = other.mAutoIncreaseStorageDuration;
    }

public:
    void setStorageDuration(const Duration& storageDuration);

    void setAutoIncreaseStorageDuration(bool increase);

    void setMinSlots(std::size_t minSlots);

    void setMaxSlots(std::size_t maxSlots);

public:
    std::size_t getSize() const;

    std::size_t getMaxSlots() const;

    std::size_t getMinSlots() const;

    Duration getStorageDuration() const;

    bool isAutoIncreasingStorageDuration() const;

public:
    virtual int getTypeId() const = 0;

    virtual bool isTyped() const = 0;

    virtual bool isPolymorphic() const = 0;

    virtual Typename getTypename() const = 0;

    virtual void setTypename(const Typename& name) = 0;

    virtual TypeMetaPtr createTypeMeta(Slot* s, MetaTypeDatabase& ioDB) = 0;

    TypeMetaPtr getTypeMeta() const
    {
        return mTypeMeta;
    }

    void setTypeMeta(TypeMetaPtr meta)
    {
        mTypeMeta = meta;
    }

    virtual void fixateType() = 0;

    virtual Slot* allocateSlot() = 0;

    virtual void freeSlot(Slot* s) = 0;

    virtual StampedHeader& getStampedHeader(Slot* s) = 0;

    virtual const Buffer<uint8>& readSerializedValue(Slot* s) = 0;

    virtual const Buffer<uint8>& readSerializedValue(Slot* s, uint8 formatVersion, bool orLower) = 0;

    virtual Buffer<uint8> readSerializedValue(Slot* s, std::list<BinarySerializerCodecPtr>& codecs) = 0;

    virtual Buffer<uint8> readSerializedValue(Slot* s, std::list<BinarySerializerCodecPtr>& codecs,
                                              uint8 formatVersion, bool orLower) = 0;

    virtual void writeSerializedValue(Slot* s, Buffer<uint8> data) = 0;

    virtual void readJSON(Slot* s, JSONValue& oValue) = 0;

    virtual void readJSON(Slot* s, JSONValue& oValue, JSONSerializer& serializer) = 0;

    virtual void writeJSON(Slot* s, const JSONValue& value) = 0;

    virtual void writeJSON(Slot* s, JSONDeserializer& value) = 0;

    virtual void writeXML(Slot* s, const XMLDom::const_iterator& node) = 0;

public:
    // write access

    Slot* requestWriteSlot();

    bool finishWriteSlot(Slot* n, bool* dropped = nullptr);

    void discardWriteSlot(Slot* n);

    // read access

    Slot* readNewestSlot();

    Slot* readSlotAtTime(const Time& timestamp, SlotQueryMode mode, bool keepStorageDuration);

    void readInterval(const Time& timestamp, std::size_t nrSlots, std::size_t olderSlots,
                      std::size_t newerSlots, IntervalFillMode fillMode, std::list<Slot*>& oSlots,
                      bool keepStorageDuration);

    void readInterval(const Time& from, const Time& to, std::list<Slot*>& oSlots, bool keepStorageDuration);

public:
    template<typename TargetType>
    ChannelBufferBase* promote();

private:
    Slot* findFirstOlderSlot(const Time& timestamp, Slot* start = nullptr);

private:
    void splice(Slot* first, Slot* last, Slot* position) const;

    void splice(Slot* item, Slot* position) const
    {
        splice(item, item, position);
    }

    static bool isEmpty(const ListItem& list)
    {
        return list.next == &list;
    }
    static Slot* begin(ListItem& list)
    {
        return list.next;
    }
    static Slot* last(ListItem& list)
    {
        return list.prev;
    }
    static Slot* end(ListItem& list)
    {
        return (Slot*)&list;
    }

    void deleteSlots(ListItem& list);

protected:
    void clear();

private:
    void increaseStorageDuration(const Time& requested);

    void resetSlot(Slot* s);

    // parameters:
    Duration mStorageDuration = Duration::seconds(0);
    bool mAutoIncreaseStorageDuration = true;
    std::size_t mMinSlots = 0;
    std::size_t mMaxSlots = 100;

    // mutex that locks all actions when we change our lists
    mutable boost::mutex mMutex;

    std::size_t mSize = 0; // current number of slots we are using

    ListItem mBuffer;

    ListItem mWaitingOrFree;

    ListItem mWriting;

    TypeMetaPtr mTypeMeta;

private:
    std::string dbgDumpList(ListItem& list, bool brief);

public:
    void dbgDump(const std::string& prefix, bool brief = true);

    static void dbgCheckListIntegrity(const ListItem& list);
    void dbgCheckIntegrity();
};


template<typename T>
class TypedChannelBufferBase : public ChannelBufferBase
{
public:
    typedef Stamped<T> ValueType;

public:
    virtual ~TypedChannelBufferBase()
    {
        this->clear();
    }

public:
    struct Slot : public ChannelBufferBase::Slot
    {
        ValueType data;
        EIGEN_MAKE_ALIGNED_OPERATOR_NEW

        template<bool IsCheap = IsCheapToCopy<T>::value>
        std::unique_ptr<Slot> clone()
        {
            static_assert(IsCheap,
                          "TypedChannelBufferBase<T>::Slot::clone() must not be called "
                          "with IsCheapToCopy<T>=false");
            auto slot = std::make_unique<Slot>();
            {
                boost::mutex::scoped_lock lock(this->serializedValueLock);
                slot->serializedValue = this->serializedValue;
            }
            slot->timestampPtr = &(static_cast<StampedHeader&>(slot->data).timestamp);
            slot->flags = this->flags;
            slot->data = this->data;
            return slot;
        }
    };

protected:
    Slot* castSlot(ChannelBufferBase::Slot* s)
    {
        // cast abstract ChannelBufferBase::Slot to TypedChannelBufferBase<T>::Slot.
        // this is safe since all slots were created by us in allocateSlot()
        // as TypedChannelBufferBase<T>::Slots
        return static_cast<Slot*>(s);
    }

public:
    int getTypeId() const override
    {
        return typeId<T>();
    }
    bool isTyped() const override
    {
        return true;
    }
    bool isPolymorphic() const override
    {
        return false;
    }
    Typename getTypename() const override
    {
        return typeName<T>();
    }
    void setTypename(const Typename& name) override
    {
        MIRA_THROW(XRuntime, "Cannot set Typename for typed channels");
    }
    TypeMetaPtr createTypeMeta(ChannelBufferBase::Slot* s, MetaTypeDatabase& ioDB) override
    {
        MetaSerializer ms(ioDB);
        Slot* slot = castSlot(s);
        TypeMetaPtr meta = ms.addMeta(slot->data.internalValueRep());
        return meta;
    }

    void fixateType() override {} // doesn't do anything
    ChannelBufferBase::Slot* allocateSlot() override
    {
        Slot* slot = new Slot;
        slot->timestampPtr =
            &(static_cast<StampedHeader&>(slot->data).timestamp); // "connect" the timestamp ptr
        return slot;
    }
    void freeSlot(ChannelBufferBase::Slot* s) override
    {
        // casting s to our slot type ensures that the correct destructor is called
        Slot* slot = castSlot(s);
        delete slot;
    }

public:
    StampedHeader& getStampedHeader(ChannelBufferBase::Slot* s) override
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);
        return stampedHeader(slot->data);
    }

    const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s) override
    {
        // lock the serialization mutex to protect serializedValueLock
        // (we or someone else may write it here)
        boost::mutex::scoped_lock lock(s->serializedValueLock);

        // no serialized data yet, so serialize it now (lazy evaluation)
        if (s->serializedValue.empty()) {
            // we can safely cast here, since s is an object of our Slot type
            Slot* slot = static_cast<Slot*>(s);

            // serialize the data
            BinaryBufferSerializer bs(&s->serializedValue);
            bs.serialize(slot->data.internalValueRep(), false);
        }

        return s->serializedValue;
    }

    const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s, uint8 formatVersion,
                                             bool orLower) override
    {
        uint8 serializerVersion = BinaryBufferSerializer::getSerializerFormatVersion();

        if (formatVersion == serializerVersion)
            return readSerializedValue(s);

        if (formatVersion == 0) {
            // lock the serialization mutex to protect serializedValueLock
            // (we or someone else may write it here)
            boost::mutex::scoped_lock lock(s->serializedValueLock);

            // no serialized data yet, so serialize it now (lazy evaluation)
            if (s->serializedValue.empty()) {
                // we can safely cast here, since s is an object of our Slot type
                Slot* slot = static_cast<Slot*>(s);

                // serialize the data
                BinaryBufferSerializerLegacy bs(&s->serializedValue);
                bs.serialize(slot->data.internalValueRep(), false);
            }

            return s->serializedValue;
        }

        MIRA_THROW(XIO,
                   "Requested serialized data of binary format version "
                       << (int)formatVersion << ". Only implemented for versions 0, "
                       << (int)serializerVersion << ".");
    }

    Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s,
                                      std::list<BinarySerializerCodecPtr>& codecs) override
    {
        Buffer<uint8> serializedValue;

        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        BinaryBufferSerializer bs(&serializedValue);

        // register codecs
        foreach (BinarySerializerCodecPtr codec, codecs)
            bs.registerCodec(codec);

        // serialize the data
        bs.serialize(slot->data.internalValueRep(), false);

        return std::move(serializedValue);
    }

    Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s, std::list<BinarySerializerCodecPtr>& codecs,
                                      uint8 formatVersion, bool orLower) override
    {
        uint8 serializerVersion = BinaryBufferSerializer::getSerializerFormatVersion();

        if (formatVersion == serializerVersion)
            return readSerializedValue(s, codecs);

        if (formatVersion == 0) {
            Buffer<uint8> serializedValue;

            // we can safely cast here, since s is an object of our Slot type
            Slot* slot = static_cast<Slot*>(s);

            BinaryBufferSerializerLegacy bs(&serializedValue);

            // register codecs
            foreach (BinarySerializerCodecPtr codec, codecs)
                bs.registerCodec(codec);

            // serialize the data
            bs.serialize(slot->data.internalValueRep(), false);

            return std::move(serializedValue);
        }

        MIRA_THROW(XIO,
                   "Requested serialized data of binary format version "
                       << (int)formatVersion << ". Only implemented for versions 0, "
                       << (int)serializerVersion << ".");
    }

    void writeSerializedValue(ChannelBufferBase::Slot* s, Buffer<uint8> data) override
    {
        s->serializedValue = std::move(data);

        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        // deserialize into our data
        BinaryBufferDeserializer bs(&s->serializedValue);

        {
            boost::mutex::scoped_lock lock(mCodecsMutex);
            bs.setCodecs(mCodecs);
        }
        bs.deserialize(slot->data.internalValueRep(), false);
        {
            boost::mutex::scoped_lock lock(mCodecsMutex);
            mCodecs = bs.getCodecs();
        }
    }

    void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue) override
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        JSONSerializer json;
        oValue = json.serialize(slot->data.internalValueRep());
    }

    void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue, JSONSerializer& serializer) override
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        oValue = serializer.serialize(slot->data.internalValueRep());
    }

    void writeJSON(ChannelBufferBase::Slot* s, const JSONValue& value) override
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        JSONDeserializer json(value);
        json.deserialize(slot->data.internalValueRep());
    }

    void writeJSON(ChannelBufferBase::Slot* s, JSONDeserializer& deserializer) override
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        deserializer.deserialize(slot->data.internalValueRep());
    }

    void writeXML(ChannelBufferBase::Slot* s, const XMLDom::const_iterator& node) override
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);
        XMLDeserializer xml(node);
        const std::string rootTag = *node;
        xml.deserializeFromNode(rootTag.c_str(), slot->data.internalValueRep());
    }

private:
    BinaryBufferDeserializer::CodecsMap mCodecs;
    boost::mutex mCodecsMutex;
};


template<typename T>
class ChannelBuffer : public TypedChannelBufferBase<T>
{};


template<>
class ChannelBuffer<void> : public ChannelBufferBase
{
public:
    typedef StampedHeader ValueType;

public:
    struct Slot : public ChannelBufferBase::Slot
    {
        ValueType header;
        EIGEN_MAKE_ALIGNED_OPERATOR_NEW
    };

protected:
    Slot* castSlot(ChannelBufferBase::Slot* s)
    {
        return static_cast<Slot*>(s);
    }

public:
    ChannelBuffer() : mTypename("") {}

    virtual ~ChannelBuffer()
    {
        this->clear();
    }

public:
    int getTypeId() const override
    {
        return -1;
    }
    bool isTyped() const override
    {
        return false;
    }
    bool isPolymorphic() const override
    {
        return false;
    }
    Typename getTypename() const override
    {
        return mTypename;
    }
    void setTypename(const Typename& name) override;
    TypeMetaPtr createTypeMeta(ChannelBufferBase::Slot* s, MetaTypeDatabase& ioDB) override
    {
        MIRA_THROW(XRuntime, "Cannot create type meta for untyped channels");
        // keep compiler happy
        return TypeMetaPtr();
    }
    void fixateType() override {} // doesn't do anything

    ChannelBufferBase::Slot* allocateSlot() override
    {
        Slot* slot = new Slot;
        slot->timestampPtr = &slot->header.timestamp; // "connect" the timestamp ptr
        return slot;
    }

    void freeSlot(ChannelBufferBase::Slot* s) override
    {
        // casting s to our slot type ensures that the correct destructor is called
        Slot* slot = castSlot(s);
        delete slot;
    }

public:
    StampedHeader& getStampedHeader(ChannelBufferBase::Slot* s) override
    {
        // casting here is not safe, if we actually are a typed buffer,
        // however, if we are, then the TypedChannelBuffer::getStampedHeader()
        // of that class would have been called instead of us.
        Slot* slot = static_cast<Slot*>(s);
        return slot->header;
    }

    const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s) override
    {
        return s->serializedValue;
    }

    const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s, uint8 formatVersion,
                                             bool orLower) override
    {
        BinaryBufferIstream stream(&(s->serializedValue));
        uint8 dataVersion = BinaryBufferDeserializer::getDataFormatVersion(stream);
        if ((!orLower && (dataVersion != formatVersion)) || (dataVersion > formatVersion)) {
            MIRA_THROW(XIO,
                       "Untyped channel contains serialized data of version "
                           << (int)dataVersion << ", cannot return data of version " << (int)formatVersion
                           << ".");
        }

        return readSerializedValue(s);
    }

    Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s,
                                      std::list<BinarySerializerCodecPtr>& codecs) override
    {
        // we must ignore the codecs
        return s->serializedValue;
    }

    Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s, std::list<BinarySerializerCodecPtr>& codecs,
                                      uint8 formatVersion, bool orLower) override
    {
        // we must ignore the codecs
        return readSerializedValue(s, formatVersion, orLower);
    }

    void writeSerializedValue(ChannelBufferBase::Slot* s, Buffer<uint8> data) override
    {
        s->serializedValue = std::move(data);
    }

    void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue) override;
    void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue, JSONSerializer& serializer) override;

    void writeJSON(ChannelBufferBase::Slot* s, const JSONValue& value) override;
    void writeJSON(ChannelBufferBase::Slot* s, JSONDeserializer& deserializer) override;
    void writeXML(ChannelBufferBase::Slot* s, const XMLDom::const_iterator& node) override;


private:
    Typename mTypename; 
};


class PolymorphicChannelBuffer : public TypedChannelBufferBase<Object*>
{
    typedef TypedChannelBufferBase<Object*> Base;

public:
    PolymorphicChannelBuffer() : mMostDerivedClass(nullptr), mFixatedClass(nullptr) {}

public:
    bool isPolymorphic() const override
    {
        return true;
    }

    int getTypeId() const override;

    Typename getTypename() const override;

    ChannelBufferBase::Slot* allocateSlot() override;

    void freeSlot(ChannelBufferBase::Slot* s) override;

    void fixateType() override;

    void writeSerializedValue(ChannelBufferBase::Slot* s, Buffer<uint8> data) override;

    void promote(const Class* promotionClass);

protected:
    const Class* mMostDerivedClass;
    const Class* mFixatedClass;
};

// declare template extern to speed up compilation time by several seconds
extern template class TypedChannelBufferBase<Object*>;


template<typename T>
class ChannelBuffer<T*> : public PolymorphicChannelBuffer
{
    static_assert(std::is_base_of<mira::Object, T>::value,
                  "In channels you can only use pointers of polymorphic "
                  "classes that are derived from mira::Object. Pointers to "
                  "other classes cannot be stored in a channel.");

public:
    typedef Stamped<T*> ValueType;

public:
    ChannelBuffer()
    {
        mMostDerivedClass = &T::CLASS();
    }
};

// Channel buffer promotion code

template<typename U>
struct ChannelBufferPromoterCommon
{
    static ChannelBufferBase* promoteUntyped(ChannelBufferBase* buffer)
    {
        assert(!buffer->isTyped() && "Can promote untyped channels to typed channels only");

        if (!buffer->getTypename().empty() && buffer->getTypename() != typeName<U>())
            MIRA_THROW(XBadCast,
                       "Invalid promotion from untyped to typed "
                       "ChannelBuffer. Typename does not match. ('"
                           << buffer->getTypename() << "' != '" << typeName<U>() << "'");

        typedef ChannelBufferBase::Slot Slot;

        // create a new typed buffer
        ChannelBuffer<U>* typedBuffer = new ChannelBuffer<U>;

        // copy all parameters
        typedBuffer->cloneParameters(*buffer);

        // copy all the serialized data from the untyped buffer into the
        // typed buffer and deserialize it into the value

        // now lock the whole buffer for reading ...
        for (Slot* s = ChannelBufferBase::begin(buffer->mBuffer);
             s != ChannelBufferBase::end(buffer->mBuffer); s = s->next)
            s->lock.lock_shared();

        // ... and deserialize the content of each slot into the new buffer
        for (Slot* s = ChannelBufferBase::begin(buffer->mBuffer);
             s != ChannelBufferBase::end(buffer->mBuffer); s = s->next) {
            Slot* n = typedBuffer->requestWriteSlot(); // obtain a new slot

            // ... copy the headers
            typedBuffer->getStampedHeader(n) = buffer->getStampedHeader(s);

            // ... and write the value into the new slot (this will deserialize the data there)
            typedBuffer->writeSerializedValue(n, s->serializedValue);

            typedBuffer->finishWriteSlot(n);
        }

        // free the read lock from above ...
        for (Slot* s = ChannelBufferBase::begin(buffer->mBuffer);
             s != ChannelBufferBase::end(buffer->mBuffer); s = s->next)
            s->lock.unlock_shared();

        return typedBuffer;
    }
};


template<typename U>
struct ChannelBufferPromoter : public ChannelBufferPromoterCommon<U>
{
    typedef ChannelBufferPromoterCommon<U> Base;
    static ChannelBufferBase* promote(ChannelBufferBase* buffer)
    {
        if (buffer->isTyped())
            return buffer; // we are promoted already, do nothing
        return Base::promoteUntyped(buffer);
    }
};


template<typename U>
struct ChannelBufferPromoter<U*>
{
    typedef ChannelBufferPromoterCommon<U> Base;
    static ChannelBufferBase* promote(ChannelBufferBase* buffer)
    {
        // if untyped, promote to typed first
        if (!buffer->isTyped())
            buffer = Base::promoteUntyped(buffer);

        PolymorphicChannelBuffer* polymorphicBuffer = dynamic_cast<PolymorphicChannelBuffer*>(buffer);

        assert(polymorphicBuffer != nullptr
               && "We should never reach here if the buffer to promote is not "
                  "polymorphic");

        // get class of U
        const Class* classU = &U::CLASS();

        // promote us (this will do error checking and might throw XBadCast)
        polymorphicBuffer->promote(classU);

        return polymorphicBuffer;
    }
};


template<typename U>
ChannelBufferBase* ChannelBufferBase::promote()
{
    return ChannelBufferPromoter<U>::promote(this);
}


} // namespace mira

#endif