ChannelManager.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_CHANNELMANAGER_H_
#define _MIRA_CHANNELMANAGER_H_

#include <platform/Types.h>
#include <error/Exceptions.h>
#include <thread/Thread.h>
#include <transform/Pose.h>

#include <fw/Channel.h>

namespace mira {


MIRA_DEFINE_SERIALIZABLE_EXCEPTION(XUnknownChannel, XRuntime)


class MIRA_FRAMEWORK_EXPORT ChannelManager
{
public:
    struct ChannelInfo
    {
        ChannelInfo() {}

        ChannelInfo(const std::string& iID, bool iInternal = false) :
            id(iID),
            internal(iInternal) {}

        bool operator<(const ChannelInfo& other) const
        {
            return id < other.id;
        }

        bool operator==(const ChannelInfo& other) const
        {
            return id == other.id;
        }

        std::string id;
        bool internal;
    };

    ChannelManager();
    ~ChannelManager();

    template<typename Reflector>
    void reflect(Reflector& r)
    {
        r.interface("IChannelManager");

        r.method("createChannel",
                 [this](const std::string& channelID, const std::string& type) {
                    std::string id = ResourceName::makeFullyQualified(channelID, "/");
                    auto ch = publish<void>(id, "", true, type);                   // publish void, internal, without publisher
                    ChannelPromoteByTypename::instance().promoteChannel(ch, type); // try to automatically promote to typed
                 },
                 "Publish a channel with given type. If type is registered for automatic promotion, "
                 "the channel will be promoted to typed immediately, else remains untyped.",
                 "channelID", "ID of channel to create", "/namespace/Rect2i",
                 "type", "Typename of channel type", "mira::Rect<int,2>");

        r.method("writeChannel",
                 [this](const std::string& channelID, const json::Value& value,
                        Time timestamp, std::string frameID, uint32 sequenceID) {
                    writeChannel(channelID, value, std::move(timestamp),
                                 std::move(frameID), sequenceID);
                 },
                 "Writes a json representation to the channel, "
                 "with specified timestamp, frame and sequence ID.",
                 "channelID", "ID of channel to write to", "/robot/RobotFrame",
                 "value", "value to write", JSONSerializer().serialize(Pose2()), 
                 "time", "data timestamp", Time::now(),
                 "frameID", "data frame ID", "/robot/OdometryFrame",
                 "sequenceID", "data sequence ID", 0);
        r.method("writeChannel",
                 [this](const std::string& channelID, const json::Value& value) {
                    writeChannel(channelID, value);
                  },
                 "Writes a json representation to the channel "
                 "with timestamp taken from clock 'now', default frame ID and increasing sequence ID.",
                 "channelID", "ID of channel to write to", "/robot/RobotFrame",
                 "value", "value to write", JSONSerializer().serialize(Pose2()));
        r.method("writeChannel",
                 [this](const std::string& channelID, const json::Value& value, Time timestamp) {
                    writeChannel(channelID, value, std::move(timestamp));
                  },
                 "Writes a json representation to the channel with specified timestamp, "
                 "default frame ID and increasing sequence ID.",
                 "channelID", "ID of channel to write to", "/robot/RobotFrame",
                 "value", "value to write", JSONSerializer().serialize(Pose2()),
                 "time", "data timestamp", Time::now());
        r.method("writeChannelWithFrameID", // needs a different name to distinguish JSONRPC signature from above
                 [this](const std::string& channelID, const json::Value& value, std::string frameID) {
                    writeChannel(channelID, value, std::move(frameID));
                  },
                 "Writes a json representation to the channel with specified frame ID, "
                  "timestamp taken from clock 'now' and increasing sequence ID.",
                 "channelID", "ID of channel to write to", "/robot/RobotFrame",
                 "value", "value to write", JSONSerializer().serialize(Pose2()),
                 "frameID", "data frame ID", "/robot/OdometryFrame");

        r.method("readChannel", &ChannelManager::readChannel, this,
                 "Reads the latest channel data as json representation",
                 "channelID", "ID of channel to read", "/robot/RobotFrame");
        r.method("getChannelList", &ChannelManager::getChannelList, this,
                 "get list of channels",
                 "publishedOnly", "if true, include only channels with (local) publishers", false,
                 "subscribedOnly", "if true, include only channels with (local) subscribers", false);
    }

public:
    template<typename T>
    ConcreteChannel<T>* publish(const std::string& channelID,
                                const std::string& publisherID,
                                bool internal, const Typename& type,
                                bool noAutoPromoteCheck = false)
    {
        // If exists && id and T are not the same throw Exception
        ConcreteChannel<T>* channel = obtainConcreteChannel<T>(channelID);
        channel->fixateType(); // fixate the type

        if ( channel->isTyped() ) {
            Typename channelType = channel->getTypename();
            if ((channelType != type) &&
                // typed publish to polymorphic channel might specify a different type
                // (when the polymorphic channel is typed, its type has been fixated),
                //  so untyped publish must match it)
                ((channelType[channelType.size()-1] != '*') || (typeName<T>() == "void")))
                MIRA_THROW(XBadCast, "Invalid publish<void> to typed channel '" << channelID <<
                           "'. Typename does not match. ('" <<
                           type << "' != '" << channel->getTypename() << "')");

            if (mCheckChannelRegistrations && !noAutoPromoteCheck)
                ChannelPromoteByTypename::instance().checkForPromoter(channel);
        } else {
            if (!channel->getTypename().empty() && (channel->getTypename() != type))
                MIRA_THROW(XBadCast, "Invalid publish<void> to untyped channel '" << channelID <<
                           "' (with set typename). Typename does not match. ('" <<
                           type << "' != '" << channel->getTypename() << "')");

            channel->setTypename(type);
        }

        {
            insertPublishedChannel(channelID, ChannelInfo(publisherID, internal));
            MIRA_LOG(DEBUG) << "'" << publisherID << "' published channel '" << channelID << "'";
        }

        if (!internal)
            remotePublishChannel(channelID, type);

        return channel;
    }

    void enableAutoPromoteChannels(bool enable) { mAutoPromoteChannels = enable; }

    bool autoPromoteChannelsEnabled() { return mAutoPromoteChannels; }

    void enableCheckChannelRegistrations(bool enable) { mCheckChannelRegistrations = enable; }

    bool checkChannelRegistrationsEnabled() { return mCheckChannelRegistrations; }

public:

    template<typename T>
    ConcreteChannel<T>* subscribe(const std::string& channelID,
                                  const std::string& subscriberID,
                                  const Duration& storageDuration, bool internal)
    {
        ConcreteChannel<T>* channel = obtainConcreteChannel<T>(channelID);

        assert(channel->getBuffer()!=NULL);
        channel->getBuffer()->setStorageDuration(storageDuration);
        insertSubscribedChannel(channelID, ChannelInfo(subscriberID, internal));
        MIRA_LOG(DEBUG) << "'" << subscriberID << "' subscribed to channel '" << channelID << "'";

        if (!internal)
            remoteSubscribeChannel(channelID);

        return channel;
    }

    void removePublisherFromAllChannels(const std::string& publisherID, bool internal);

    void removePublisher(const std::string& channelID, const std::string& publisherID, bool internal);

    void removeSubscriberFromAllChannels(const std::string& subscriberID);

    void removeSubscriber(const std::string& channelID, const std::string& subscriberID);

    void deductNumSubscribers(const std::string& channelID);

    template<typename T>
    ConcreteChannel<T>* getConcreteChannel(const std::string& channelID)
    {
        return channel_cast<T>(getAbstractChannel(channelID));
    }

public:

    std::set<std::string> getChannels() const;

    std::set<std::string> getChannelList(bool publishedOnly = false,
                                         bool subscribedOnly = false) const;

    std::set<std::string> getSubscribedChannels(bool includeInternal = false) const;

    std::set<std::string> getSubscribedChannelsBy(const std::string& subscriberID) const;

    bool isSubscribedOn(const std::string& subscriberID,
                        const std::string& channelID) const;

    std::map<std::string, Typename> getPublishedChannels(bool includeInternal = false) const;

    std::set<std::string> getPublishedChannelsBy(const std::string& publisherID) const;

    bool hasPublished(const std::string& publisherID, const std::string& channelID) const;

    std::list<std::pair<std::string, Typename>> getChannelsOfType(const Typename& type) const;

    template <typename T>
    std::list<std::pair<std::string, Typename>> getChannelsOfType() const {
        return getChannelsOfType(typeName<T>());
    }

    bool hasSubscriber(const std::string& channelID) const;

    bool hasPublisher(const std::string& channelID, bool includeInternal) const;

    uint32 getNrPublishers(const std::string& channelID) const;

    std::set<std::string> getPublishers(const std::string& channelID,
                                        bool includeInternal = false) const;

    uint32 getNrSubscribers(const std::string& channelID) const;

    std::set<std::string> getSubscribers(const std::string& channelID,
                                         bool includeInternal = false) const;

    bool hasChannel(const std::string& channelID) const;

    int getTypeId(const std::string& channelID) const;

    void setTypename(const std::string& channelID, const Typename& typenam);

    Typename getTypename(const std::string& channelID) const;

    void setTypeMeta(const std::string& channelID, TypeMetaPtr meta);

    TypeMetaPtr getTypeMeta(const std::string& channelID) const;

    Time getLastSlotTime(const std::string& channelID) const;

    std::size_t getNrOfSlots(const std::string& channelID) const;

    uint64 getNrOfDataChanges(const std::string& channelID) const;

public:

    void ensureChannel(const std::string& channelID)
    {
        obtainConcreteChannel<void>(channelID);
    }

    void setStorageDuration(const std::string& channelID, const Duration& storageDuration);

    void setAutoIncreaseStorageDuration(const std::string& channelID, bool increase);

    void setMinSlots(const std::string& channelID, std::size_t minSlots);

    void setMaxSlots(const std::string& channelID, std::size_t maxSlots);

    Duration getStorageDuration(const std::string& channelID);

    bool isAutoIncreasingStorageDuration(const std::string& channelID);

    std::size_t getMinSlots(const std::string& channelID);

    std::size_t getMaxSlots(const std::string& channelID);

public:

    void writeChannel(const std::string& channelID,
                      const json::Value& value,
                      Time time,
                      std::string frameID,
                      uint32 sequenceID);

    void writeChannel(const std::string& channelID,
                      const json::Value& value,
                      Time time = Time::now());

    void writeChannel(const std::string& channelID,
                      const json::Value& value,
                      std::string frameID,
                      Time time = Time::now());

    json::Value readChannel(const std::string& channelID);

private:

    const AbstractChannel* getAbstractChannel(const std::string& channelID) const;

    AbstractChannelPtr getAbstractChannel(const std::string& channelID);

    AbstractChannelPtr getAbstractChannelNoLocking(const std::string& channelID);

    const AbstractChannel* getAbstractChannelNoLocking(const std::string& channelID) const;

    template<typename T>
    ConcreteChannel<T>* obtainConcreteChannel(const std::string& channelID)
    {
        boost::mutex::scoped_lock lock(mChannelsMutex);

        ConcreteChannel<T>* channel = NULL;
        try
        {
            channel = channel_cast<T>(getAbstractChannelNoLocking(channelID));
        }
        catch (XUnknownChannel&)
        {
            // if we reach here, the channel does not exist, so create a new one
            channel = new ConcreteChannel<T>(channelID);
            mChannels.insert(ChannelMap::value_type(channelID, channel));
        }
        return channel;
    }

    void remotePublishChannel(const std::string& channelID, const Typename& type);
    void remoteUnpublishChannel(const std::string& channelID);
    void remoteSubscribeChannel(const std::string& channelID);
    void remoteUnsubscribeChannel(const std::string& channelID);


private:

    void insertPublishedChannel(const std::string& channelID,  const ChannelInfo& info);
    void insertSubscribedChannel(const std::string& channelID,  const ChannelInfo& info);

private:

    typedef std::map<std::string, AbstractChannelPtr > ChannelMap;
    ChannelMap mChannels; 
    mutable boost::mutex mChannelsMutex;

    class Pimpl;
    Pimpl* p;

    mutable boost::mutex mSubscriberMutex;
    mutable boost::mutex mPublisherMutex;

    bool mAutoPromoteChannels;
    bool mCheckChannelRegistrations;

};


}

#endif