RPCManager.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_RPCMANAGER_H_
#define _MIRA_RPCMANAGER_H_

#include <set>

#include <platform/Platform.h>

#include <rpc/RPCServer.h>
#include <rpc/RPCClient.h>
#include <rpc/RPCPatternCheck.h>
#include <rpc/RPCCallDefinition.h>
#include <rpc/AbstractDeferredInvoker.h>
#include <rpc/AbstractRPCHandler.h>
#include <rpc/AbstractInterfaceCallbackHandler.h>

namespace mira {


class RPCManager : protected DeferredInvokerFinishHandler
{
public:

    class RemoteRequestHandler
    {
    public:
        virtual ~RemoteRequestHandler() {}
        virtual void onRPCrequested(Buffer<uint8>&& request) = 0;
    };
    typedef boost::shared_ptr<RemoteRequestHandler> RemoteRequestHandlerPtr;


    class RemoteFinishHandler
    {
    public:
        virtual ~RemoteFinishHandler() {}
        virtual void onRPCfinished(Buffer<uint8>&& response) = 0;
    };
    typedef boost::shared_ptr<RemoteFinishHandler> RemoteFinishHandlerPtr;

public:

    template<typename Reflector>
    void reflect(Reflector& r)
    {
        r.interface("IRPCManager");
        r.method("getLocalServices", &RPCManager::getLocalServices, this,
                 "Get all registered local services");
        r.method("getLocalService", &RPCManager::getLocalService, this,
                 "Returns a reference to the local service object with the given name",
                 "name", "name of the service", "MyService");
        r.method("getRemoteServices", &RPCManager::getRemoteServices, this,
                 "Get all registered remote services");
        r.method("getRemoteService", &RPCManager::getRemoteService, this,
                 "Returns a reference to the remote service object with the given name",
                 "name", "name of the service", "MyService");
        r.method("existsService", &RPCManager::existsService, this,
                 "Returns true, if a service with the given name exists, otherwise false",
                 "name", "name of the service", "MyService");
        r.method("queryServicesForInterface", &RPCManager::queryServicesForInterface, this,
                 "Returns a string list with the names of all registered services (local"
                 "and remote), that implement the specified interface",
                 "name", "name of the interface", "MyInterface");
        r.method("implementsInterface", &RPCManager::implementsInterface, this,
                 "Returns if the given service implements a certain interface",
                 "name", "name of the service", "MyService",
                 "interface", "name of the interface", "MyInterface");
    }

public: // local service registering

    template <typename Service>
    void addLocalService(const std::string& name, Service& service,
                         AbstractRPCHandlerPtr handler)
    {
        boost::mutex::scoped_lock lock(mServiceManagementMutex);

        // reflect the service to extend the service object and get the new methods
        // and interfaces
        std::set<RPCSignature> addedMethods;
        std::set<std::string> addedInterfaces;
        const RPCServer::Service& s = mServer.registerServiceObject(name,
                service, &addedMethods, &addedInterfaces);


        // add the service methods and their handlers
        foreach(const RPCSignature& s, addedMethods)
        {
            std::string method = name + "." + s.name;
            auto it = mLocalServiceMethods.find(method);
            if(it == mLocalServiceMethods.end())
                mLocalServiceMethods.insert(std::make_pair(method, handler));
            else {
                if(it->second != handler)
                    MIRA_THROW(XLogical, "The service method '" << method <<
                               "' was already registered with a different handler");
            }
        }

        // loop through all new interfaces of the service s
        foreach(const std::string& ifc, addedInterfaces)
            handleCallbacks(ifc, name);

        mLocalServices.insert(name);
    }

    void removeLocalService(const std::string& name);

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

    const RPCServer::Service& getLocalService(const std::string& name) const {
        return mServer.getService(name);
    }

public: // remote service registering

    typedef RPCServer::ServiceInfo<RPCServer::MethodInfoSet> RemoteService;

    void addRemoteService(const RemoteService& serviceInfo,
                          RemoteRequestHandlerPtr handler,
                          uint8 binaryFormatVersion = BinaryBufferSerializer::getSerializerFormatVersion());

    void removeRemoteService(const std::string& name);

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

    const RemoteService& getRemoteService(const std::string& name) const;

public: // interface handling

    void registerCallback(const std::string& interface,
                          AbstractInterfaceCallbackHandlerPtr callback);

    void unregisterCallback(const std::string& interface,
                            AbstractInterfaceCallbackHandlerPtr callback);

    void unregisterCallback(AbstractInterfaceCallbackHandlerPtr callback);

    bool existsService(const std::string& name) const
    {
        boost::mutex::scoped_lock lock(mServiceManagementMutex);

        // check local services first
        if(mServer.existsService(name))
            return true;

        // otherwise examine all remote services
        for(auto it=mRemoteServices.begin(); it!=mRemoteServices.end(); ++it)
            if(it->second.name==name)
                return true; // found a remote service with the desired name

        return false; // nothing found
    }

    bool waitForService(const std::string& name,
                        Duration timeout = Duration::invalid()) const;

    std::string waitForServiceInterface(const std::string& interface,
                                        Duration timeout = Duration::invalid()) const;

    std::list<std::string> queryServicesForInterface(const std::string& interface) const
    {
        boost::mutex::scoped_lock lock(mServiceManagementMutex);

        // get the local interfaces first
        std::list<std::string> res = mServer.queryServicesForInterface(interface);

        // now examine all remote services
        for(auto it=mRemoteServices.begin(); it!=mRemoteServices.end(); ++it)
            // if remote service has the interface ...
            if(it->second.interfaces.count(interface)!=0)
                res.push_back(it->second.name); // ... add it

        return res;
    }

    bool implementsInterface(const std::string& name, const std::string& interface) const
    {
        boost::mutex::scoped_lock lock(mServiceManagementMutex);
        // get the local interfaces first
        if (mServer.implementsInterface(name, interface))
            return true;

        // now examine all remote services
        auto it = mRemoteServices.find(name);
        if (it == mRemoteServices.end())
            return false;
        return it->second.interfaces.count(interface) > 0;
    }

public: // for handling remote requests and their corresponding responses

    void handleRemoteRequest(Buffer<uint8> buffer,
                             RemoteFinishHandlerPtr handler = RemoteFinishHandlerPtr());

    void handleRemoteRequest(Buffer<uint8> buffer,
                             RemoteFinishHandlerPtr handler, uint8 binaryFormatVersion);

    void handleRemoteResponse(Buffer<uint8> buffer);

    void handleRemoteResponse(Buffer<uint8> buffer, uint8 binaryFormatVersion);

public:

    template<typename R, typename... ARGS>
    RPCFuture<R> call(const std::string& service, const std::string& method, ARGS&&... args)
    {
        Buffer<uint8> buffer;
        boost::mutex::scoped_lock lock(mServiceManagementMutex);
        if (mLocalServices.count(service) > 0) {
            lock.unlock();
            BinaryRPCBackend::ClientRequest request(&buffer);
            RPCFuture<R> future = mClient.call<BinaryRPCBackend, R>(
                    request, service, method, forwardRPCParams(std::forward<ARGS>(args))...);
            handleLocalRequest<BinaryRPCBackend>(std::move(buffer), service);
            return std::move(future);
        }
        else {
            auto it = mRemoteServices.find(service);
            if (it == mRemoteServices.end()) {
                MIRA_THROW(XRPC,
                           "Unable to call method '" << method << "' for service '" << service
                                                     << "'. The requested service is not known.");
            };

            RPCFuture<R> future;
            if (it->second.binaryFormatVersion == BinaryBufferSerializer::getSerializerFormatVersion()) {
                BinaryRPCBackend::ClientRequest request(&buffer);
                future = mClient.call<BinaryRPCBackend, R>(request, service, method,
                                                           forwardRPCParams(std::forward<ARGS>(args))...);
            }
            else if (it->second.binaryFormatVersion == 0) {
                BinaryRPCBackendLegacy::ClientRequest request(&buffer);
                future = mClient.call<BinaryRPCBackendLegacy, R>(
                        request, service, method, forwardRPCParams(std::forward<ARGS>(args))...);
            }
            else {
                MIRA_THROW(XRPC, "Requested service expects binary format version " 
                                 << (int)it->second.binaryFormatVersion        
                                 << ". RPCManager::call() only implemented for versions 0, " 
                                 << (int)BinaryBufferSerializer::getSerializerFormatVersion()); 
            };
            it->second.handler->onRPCrequested(std::move(buffer));
            lock.unlock();
            boost::mutex::scoped_lock lock2(mPendingRequestsMutex);
            mPendingRemoteRequests[future.callId()] = service;
            return std::move(future);
        }
    }

    RPCFuture<JSONRPCResponse> callJSON(const json::Value& jsonCall);

    RPCFuture<JSONRPCResponse> callJSON(const std::string& jsonCall);

    RPCFuture<JSONRPCResponse> callJSON(const std::string& service,
                                        const std::string& method,
                                        const json::Value& params = json::Value());

    RPCFuture<JSONRPCResponse> callJSON(const std::string& service,
                                        const std::string& method,
                                        const std::string& params);

    RPCFuture<JSONRPCResponse> callJSON(const RPCCallDefinition& rpc);

protected:

    RPCFuture<JSONRPCResponse> callJSONIntern(const std::string& service,
                                              const std::string& callID,
                                              const json::Value& request);

    template <typename Backend>
    void handleLocalRequest(typename Backend::ContainerType buffer,
                            const std::string& service);


    template <typename Backend>
    void handleRequest(typename Backend::ContainerType buffer,
                       RemoteFinishHandlerPtr finishHandler, bool remote);

private:

    void handleCallbacks(const std::string& interface,
                         const std::string& servicename);

protected:

    // implements DeferredInvokerFinishHandler
    virtual void onRPCfinished(AbstractDeferredInvoker* invoker);

private:

    struct AbstractRequestContext
    {
        AbstractRequestContext(RemoteFinishHandlerPtr iHandler, bool iRemote) :
            handler(iHandler), remote(iRemote){}

        virtual ~AbstractRequestContext() {}

        virtual void onFinished(RPCClient& client) = 0;

    public:
        RemoteFinishHandlerPtr handler;
        bool remote;
    };
    typedef boost::shared_ptr<AbstractRequestContext> AbstractRequestContextPtr;

    template<typename Backend>
    struct RequestContext;

private:

    // everything is just passed through except const char* (string literal),
    // which is explicitly converted to std::string
    // thus we get the correct RPC signature without the need for explicit cast as in
    // call<void>("Service", "method", std::string("param"));

    template<typename T>
    T&& forwardRPCParams(T&& p) { return std::forward<T>(p); }

    std::string forwardRPCParams(const char* p) { return std::string(p); }

private:

    RPCServer mServer;

    RPCClient mClient;

    struct RemoteServiceDetail : public RemoteService
    {
        RemoteServiceDetail(const RemoteService& serviceInfo,
                            RemoteRequestHandlerPtr iHandler,
                            uint8 iBinaryFormatVersion = BinaryBufferSerializer::getSerializerFormatVersion()) :
            RemoteService(serviceInfo), handler(iHandler),
            binaryFormatVersion(iBinaryFormatVersion) {}

        RemoteRequestHandlerPtr handler;
        uint8 binaryFormatVersion;
    };

    mutable boost::mutex mServiceManagementMutex;

    std::map<std::string, AbstractRPCHandlerPtr> mLocalServiceMethods;

    std::set<std::string> mLocalServices;

    std::map<std::string, RemoteServiceDetail> mRemoteServices;

    mutable boost::mutex mPendingRequestsMutex;
    std::map<AbstractDeferredInvoker*, AbstractRequestContextPtr> mPendingRequests;
    std::map<std::string, std::string> mPendingRemoteRequests;

    mutable boost::mutex mCallbacksMutex;
    std::multimap<std::string, AbstractInterfaceCallbackHandlerPtr> mInterfaceCallbacks;
};


}

#endif