RemoteModule.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_REMOTEMODULE_H_
#define _MIRA_REMOTEMODULE_H_

#include <thread/CyclicRunnable.h>
#include <serialization/adapters/std/list>
#include <serialization/adapters/boost/optional.hpp>
#include <utils/UUID.h>
#include <security/RSAKey.h>

#include <fw/DiscoverService.h>
#include <fw/PerformanceStatistics.h>
#include <fw/RemoteConnection.h>
#include <fw/RemoteServer.h>
#include <fw/RemoteConnectionPool.h>
#include <fw/ServiceLevel.h>

namespace mira {

class RemoteServer;


class MIRA_FRAMEWORK_EXPORT RemoteModule
{
public:
    typedef boost::shared_ptr<RemoteServer> RemoteServerPtr;

    typedef std::list<RemoteConnectionProxy> ConnectionList;
    typedef std::map<UUID, RemoteConnectionProxy> ConnectionMap;

    typedef std::list<KnownFramework> KnownFrameworkList;
    typedef std::list<ServiceLevel> ServiceLevelList;
    typedef std::map<std::string, ServiceLevel> ServiceLevelMap;
    typedef std::list<TypeServiceLevel> TypeServiceLevelList;
    typedef std::map<Typename, TypeServiceLevel> TypeServiceLevelMap;

    friend class RemoteConnection;
    friend class RemoteIncomingConnection;
    friend class RemoteOutgoingConnectionBase;
    friend class DiscoverService;
    friend class RemoteServer;

    enum AuthMode
    {
        AUTH_NONE=0,
        AUTH_PASSWORD=1,
        AUTH_KEY=2
    };

    struct AuthSettings
    {
        std::string group;

        template<typename Reflector>
        void reflect(Reflector& r)
        {
            r.member("Group", group,
                     "The workgroup, this framework belongs to", "");
            r.member("Password", mPassword, setter(&AuthSettings::setPassword, this),
                     "Password that is used for the weak authentication",
                     serialization::IgnoreMissing());
            r.member("Key", mKeyFile, setter(&AuthSettings::setKeyFile, this),
                     "Path to the file containing the private key "
                     "for strong authentication.",
                     serialization::IgnoreMissing());
        }

        AuthMode getMode() const {
            if(mKey)
                return AUTH_KEY;
            else if(mPassword)
                return AUTH_PASSWORD;
            else
                return AUTH_NONE;
        }

        void clearAuth();

        const std::string& getPassword() const;

        void setPassword(const boost::optional<std::string>& passwd);

        const RSAKey& getKey() const;

        void setKey(const boost::optional<std::string>& str);

        void setKeyFile(const boost::optional<std::string>& file);

    private:

        boost::optional<std::string> mPassword;
        boost::optional<std::string> mKeyFile;
        boost::optional<RSAKey> mKey;
    };

    RemoteModule();

    template<typename Reflector>
    void reflect(Reflector& r)
    {
        serialization::VersionType version = r.version(2, this);
        r.member("Port", mPort,
                 "The port where the framework server runs on. 0 means auto port", 0);
        r.member("IOThreadCount", mIOThreadCount,
                 "How many threads are used for communication", 1);
        r.member("CycleTime", mCyclicRunnable.cycleTime,
                 "The cycle time of remote framework thread that observes connections",
                 Duration::seconds(1));

        r.property("EnablePTPSync", mEnablePTPSync, "Enable/Disable synchronization of clocks via PTP (effective for new connections)", true);
        r.property("EnablePingTimeout", mEnablePingTimeout, "Enable/Disable ping timeout (effective for new connections)", false);
        r.member("PingInterval", mPingInterval,
                 "The interval of sending ping messages.", Duration::seconds(1));
        r.member("PingTimeout", mPingTimeout,
                 "The maximum time before a connection is assumed as dead "
                 "when no ping messages are received.", Duration::seconds(10));

        r.member("ExitOnDisconnectTimeout", mExitOnDisconnectTimeout,
                 "Exits/Closes the framework if any connection could not be established or is closed for a certain time",
                 Duration::invalid());

        r.member("Authentication", mAuthSettings, "Authentication Settings",
                 AuthSettings());

        if (version >= 2)
            r.member("KnownFrameworks", mKnownFrameworks,
                     "The list of remote frameworks", KnownFrameworkList());
        else {
            // old KnownFramework serialization = delegate to just address
            std::vector<std::string> tmpFWs;
            r.member("KnownFrameworks", tmpFWs, "",
                     std::vector<std::string>(), REFLECT_CTRLFLAG_TEMP_TRACKING);
            mKnownFrameworks.clear();
            foreach (const std::string& a, tmpFWs) {
                mKnownFrameworks.emplace_back();
                mKnownFrameworks.back().address = a;
            }
        }

        r.member("ServiceLevels", mServiceLevels,
                 "The service level agreement per channel", ServiceLevelList());
        r.member("ServiceLevelsByType", mTypeServiceLevels,
                 "The service level agreement per channel type", TypeServiceLevelList());

        r.property("FrameworkConnections", mConnections,
                   "The list of active remote framework connections", serialization::IgnoreMissing());

        r.interface("IRemoteModule");
        r.method("migrateUnitToThisFramework", &RemoteModule::migrateUnitToThisFramework, this,
                 "Requests a unit from an other framework to migrate into this framework",
                 "id", "id of the unit to migrate", "/navigation/Pilot");
        r.method("isFrameworkConnected", &RemoteModule::isFrameworkConnected, this,
                 "Returns if a framework with ID is connected to this framework",
                 "frameworkID", "framework ID to check", "Framework_81ab60d5-0e9a-43d6-aced-7e33a3cd875a");
        r.method("isConnectedTo", &RemoteModule::isConnectedTo, this,
                 "Returns if a framework with address is connected to this framework",
                 "address", "address to check", "127.0.0.1:1234");
        r.method("disconnectFramework", &RemoteModule::disconnectFramework, this,
                 "Disconnects a framework with ID (if connected)",
                 "frameworkID", "framework ID to disconnect", "Framework_81ab60d5-0e9a-43d6-aced-7e33a3cd875a",
                 "autoReconnect", "if true, keep as known framework (thus try reconnecting)", false);
        r.method("disconnectFrom", &RemoteModule::disconnectFrom, this,
                 "Disconnects from a framework with address",
                 "address", "address to disconnect", "127.0.0.1:1234",
                 "autoReconnect", "if true, keep as known framework (thus try reconnecting)", false);
        r.method("connectTo",
                 (void(RemoteModule::*)(const std::string&))(&RemoteModule::addKnownFramework), this,
                 "Add an address to the list of known remote frameworks and try to connect",
                 "address", "Address of the remote framework as Host:Port", "127.0.0.1:1234");
        r.method("connectTo",
                 (void(RemoteModule::*)(const std::string&, bool, bool, bool, const Duration&))(&RemoteModule::addKnownFramework), this,
                 "Add an address to the list of known remote frameworks and try to connect",
                 "address", "Address of the remote framework as Host:Port", "127.0.0.1:1234",
                 "forcePTP", "Force PTP time sync", false,
                 "legacyBinaryFormat", "Use legacy binary serializer to encode data (connect to legacy framework)", false,
                 "monitorOnly", "Do not publish local channels, services and authorities to remote", false,
                 "delayConnect", "Wait for this duration before trying to connect", Duration::seconds(1));
    }

    static uint32 getCurrentVersion();

    void setPort(uint16 port) { mPort = port; }

    uint16 getPort() const;


    void enablePTPSync(bool enable=true) { mEnablePTPSync = enable; }

    bool isPTPSyncEnabled() const { return mEnablePTPSync; }

    void enablePingTimeout(bool enable=true) { mEnablePingTimeout = enable; }

    bool isPingTimeoutEnabled() const { return mEnablePingTimeout; }

    Duration getPingInterval() const { return mPingInterval; }

    Duration getPingTimeout() const { return mPingTimeout; }

    void setAuthGroup(const std::string& group);

    void setNoAuth();

    void setAuthPassword(const std::string& password);

    void setAuthKey(const std::string& key);

    void setAuthKeyFile(const std::string& keyfile);

    const AuthSettings& getAuthSettings() const;

    void setExitOnDisconnectTimeout(Duration timeout) { mExitOnDisconnectTimeout = timeout; }

    const UUID& getID() const { return mID; }

    ServiceLevel getServiceLevel(const std::string& channelID, const Typename& channelType);

    void start();

    void stop();


    ScopedAccess<const ConnectionMap, boost::recursive_mutex> getConnections() const;

    void addKnownFramework(const std::string& address);

    void addKnownFramework(const std::string& address,
                           bool forcePTP,
                           bool legacyBinaryFormat,
                           bool monitorOnly,
                           const Duration& delayConnect = Duration::seconds(0));

    void disconnectFramework(const std::string& frameworkID,
                             bool autoReconnect = false);

    void disconnectFrom(const std::string& address,
                        bool autoReconnect = false);

    bool isFrameworkConnected(const std::string& frameworkID) const;
    
    bool isConnectedTo(const std::string& address) const;

    void migrateUnitToThisFramework(const std::string& id);


    void publishAuthority(const AuthorityDescription& authority);

    void unpublishAuthority(const AuthorityDescription& authority);

    void publishService(const std::string& service);

    void unpublishService(const std::string& service);

    void publishChannel(const std::string& channelID, const Typename& type);

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

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

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


    void updateIncomingStats(std::size_t size);
    void updateOutgoingStats(std::size_t size);

    std::size_t getIncomingBytesPerSecond() const;
    std::size_t getOutgoingBytesPerSecond() const;

protected:

    RemoteConnectionProxy createIncomingConnection();

    RemoteConnectionProxy& addPendingIncomingConnection(RemoteConnectionProxy&& proxy,
                                                        bool start = true);

    void removePendingIncomingConnection(RemoteConnection* connection);

    RemoteConnectionProxy& addPendingOutgoingConnection(RemoteConnectionProxy&& proxy,
                                                        bool start = true);

    void removePendingOutgoingConnection(RemoteConnection* connection);

    void onRemoteFrameworkDiscovered(const std::string& host, uint16 port, UUID id);

    void storeConnection(RemoteConnectionProxy iConnection);

    void eraseConnection(ConnectionMap::iterator it);
    
    bool onIncomingConnected(RemoteIncomingConnection* iConnection);

    void onIncomingDisconnected(RemoteIncomingConnection* iConnection);

    bool onOutgoingConnected(RemoteOutgoingConnectionBase* iConnection);

    void onOutgoingDisconnected(RemoteOutgoingConnectionBase* iConnection);

    void process();

    void startDisconnectTimeout();
    void stopDisconnectTimeout();
    bool checkDisconnectTimeout();

protected:

    PerformanceStatistics mIncomingStats;
    PerformanceStatistics mOutgoingStats;
    uint32 mIncomingData;
    uint32 mOutgoingData;

    CyclicRunnable mCyclicRunnable;
    DiscoverServicePtr mDiscoverService;
    boost::thread mThread;
    uint16 mPort;
    bool mEnablePTPSync;
    bool mEnablePingTimeout;
    Duration mPingInterval;
    Duration mPingTimeout;
    uint16 mIOThreadCount;
    UUID mID;
    RemoteServerPtr mServer;

    ConnectionList mPendingOutgoingConnections;
    ConnectionList mPendingIncomingConnections;
    ConnectionMap mConnections;

    RemoteConnectionPool mRemoteConnectionPool;

    AuthSettings mAuthSettings;
    Duration mExitOnDisconnectTimeout;
    Time mDisconnectedSince;
    KnownFrameworkList mKnownFrameworks;
    ServiceLevelList mServiceLevels;
    TypeServiceLevelList mTypeServiceLevels;
    ServiceLevelMap mChannelServiceLevels;
    TypeServiceLevelMap mChannelTypeServiceLevels;
    mutable boost::recursive_mutex mConnectionMutex;
};


typedef boost::shared_ptr<RemoteModule> RemoteModulePtr;


template <>
class IsCopyAssignable<RemoteModule::ConnectionMap> : public std::false_type {};


}

#endif