RPCServer.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_RPCSERVER_H_
#define _MIRA_RPCSERVER_H_

#include <functional>

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

#include <platform/Environment.h>

#include <serialization/ReflectorInterface.h>

#include <error/LoggingCore.h>
#include <error/LoggingAux.h>

#include <rpc/RPCSignature.h>
#include <rpc/RPCInvoker.h>
#include <rpc/RPCError.h>
#include <rpc/BinaryRPCBackend.h>
#include <rpc/JSONRPCBackend.h>

#include <rpc/AbstractDeferredInvoker.h>

#include <utils/ToString.h>

namespace mira {


class RPCServer
{
public:

    struct ParameterInfo
    {
        ParameterInfo() {}

        ParameterInfo(const std::string& iName, const std::string& iDescription)
            : name(iName), description(iDescription) {}

        template<typename Reflector>
        void reflect(Reflector& r)
        {
            r.member("Name", name, "");
            r.member("Description", description, "");
        }

        std::string name;

        std::string description;
        
    public:

        friend std::ostream& operator<<(std::ostream& s, const ParameterInfo& v)
        {
            s << v.name << " \t: " << v.description; 
            return s;
        }
    };

    typedef std::vector<ParameterInfo> ParameterDescriptions;

    struct MethodInfo
    {
        class ReflectMethodParamsDocumentation : public Singleton<ReflectMethodParamsDocumentation>
        {
            friend class singleton::CreateUsingNew<ReflectMethodParamsDocumentation>;

        protected:
            ReflectMethodParamsDocumentation() : mReflectMethodParamsDocumentation(true)
            {
                try {
                    resolveEnvironmentVariable("MIRA_RPC_METHODS_REFLECT_NO_PARAMS_DOCUMENTATION");
                    // if resolve succeeds (--> variable exists), do not reflect version and parameter documentation
                    mReflectMethodParamsDocumentation = false;
                    MIRA_LOG(WARNING) << "Found definition of environment variable "
                                      << "MIRA_RPC_METHODS_REFLECT_NO_PARAMS_DOCUMENTATION "
                                      << "- RPC parameter documentation will not be shared with "
                                      << "remote frameworks (ensuring backward compatibility)";
                }
                catch(...) {}
            }

        public:
            static bool enabled() { return instance().mReflectMethodParamsDocumentation; }

        private:
            bool mReflectMethodParamsDocumentation;
        };

        MethodInfo() : parameterSamplesDefault(true) {}

        MethodInfo(const RPCSignature& iSignature, const std::string& iComment) :
            signature(iSignature), comment(iComment), parameterSamplesDefault(true) {}

        MethodInfo(const RPCSignature& iSignature, const std::string& iComment,
                   const ParameterDescriptions& iParameterDescriptions) :
            signature(iSignature), comment(iComment),
            parameterDesc(iParameterDescriptions), parameterSamplesDefault(true) {}

        template<typename Reflector>
        void reflect(Reflector& r)
        {
            serialization::VersionType version = r.version(3, this);

            r.member("Signature", signature, "");
            r.member("Comment", comment, "");

            if (version >= 2) {
                r.member("ParameterDescriptions", parameterDesc, "");
                r.member("ParameterSamples", parameterSamples, "");

                if (version >= 3)
                    r.member("ParameterSamplesDefault", parameterSamplesDefault, "");
            }
        }

        template<typename Reflector>
        void reflectBinaryV0(Reflector& r)
        {
            serialization::VersionType version = 0;

            if (ReflectMethodParamsDocumentation::enabled())
                version = r.version(2, this); // version 3 did not exist with serialization format v0

            r.member("Signature", signature, "");
            r.member("Comment", comment, "");

            if (version >= 2) {
                r.member("ParameterDescriptions", parameterDesc, "");
                r.member("ParameterSamples", parameterSamples, "");
            }
        }

        // Specialization for serialization format v0
        template<typename Stream>
        void reflect(ConcreteBinarySerializer<Stream, 0>& r)
        {
            reflectBinaryV0(r);
        }

        // Specialization for deserialization format v0
        template<typename Stream>
        void reflect(ConcreteBinaryDeserializer<Stream, 0>& r)
        {
            reflectBinaryV0(r);
        }

        bool operator<(const MethodInfo& rhs) const {
            return signature < rhs.signature;
        }

        bool operator==(const MethodInfo& rhs) const {
            return signature==rhs.signature;
        }

        std::string extendedSignature() const;

        std::string parameterDescriptions(const std::string& prefix = "") const;

        std::string sampleParametersSet(bool formatted = false) const;

        RPCSignature signature;

        std::string comment;

        ParameterDescriptions parameterDesc;

        bool parameterSamplesDefault;
        std::list<json::Value> parameterSamples;
    };

    struct Method : public MethodInfo
    {
        Method(const RPCSignature& signature, const std::string& comment) :
            MethodInfo(signature, comment) {}

        Method(const RPCSignature& signature, const std::string& comment,
               const ParameterDescriptions& parameterDescriptions) :
            MethodInfo(signature, comment, parameterDescriptions) {}

        template<typename Backend>
        void addInvoker(TRPCInvoker<Backend>* invoker)
        {
            invokers.insert(std::make_pair(typeId<Backend>(),
                                           boost::shared_ptr<RPCInvoker>(invoker)));
        }

        std::map<int, boost::shared_ptr<RPCInvoker>> invokers;
    };

    typedef std::set<Method> MethodSet;

    typedef std::set<MethodInfo> MethodInfoSet;


    template <typename TMethodSet>
    struct ServiceInfo
    {
        ServiceInfo(const std::string& iName="") :
            name(iName) {}

        template<typename Reflector>
        void reflect(Reflector& r)
        {
            r.member("Name", name, "");
            r.member("Methods", methods, "");
            r.member("Interfaces", interfaces, "");
        }

        std::string name;
        TMethodSet methods;
        std::set<std::string> interfaces;
    };

    typedef ServiceInfo<MethodSet> Service;

    typedef std::map<std::string, Service> ServiceMap;


    class RPCReflector : public ReflectorInterface<RPCReflector>
    {
    public:

        RPCReflector(Service& iService) :
            mService(iService) {}


    public:

        template <typename Base>
        void reflectBase(Base& base) {
            // simplified (compared to AbstractReflector), but should be enough for the RPCServer
            base.reflect(*this);
        }

    public:

        const std::set<RPCSignature>& getAddedMethods() const {
            return mAddedMethods;
        }

        const std::set<std::string>& getAddedInterfaces() const {
            return mAddedInterfaces;
        }

    public:

        void interface(const char* name) {
            if(mService.interfaces.insert(std::string(name)).second)
                mAddedInterfaces.insert(std::string(name));
        }

        template<typename R, typename... Args, typename F, typename... Description>
        InvalidRPCDescription<R(Args...), Description...>
        method(const char*, F&&, Description&&...)
        {
            Private::rpc::invalidAssertion<R(Args...), Description...>();
        }

        template<typename F, typename... Description>
        InvalidRPCDescription<F, Description...>
        method(const char*, F&&, Description&&...)
        {
            Private::rpc::invalidAssertion<F, Description...>();
        }

        template<typename R, typename Class, typename... Args, typename... Description>
        InvalidRPCDescription<R (Class::*)(Args...), Description...>
        method(const char*, R (Class::*)(Args...), Class*, Description&&...)
        {
            Private::rpc::invalidAssertion<R (Class::*)(Args...), Description...>();
        }

        template<typename R, typename Class, typename... Args, typename... Description>
        InvalidRPCDescription<R (Class::*)(Args...) const, Description...>
        method(const char*, R (Class::*)(Args...) const, Class*, Description&&...)
        {
            Private::rpc::invalidAssertion<R (Class::*)(Args...) const, Description...>();
        }

        template<typename R, typename... Args, typename F, typename... Description>
        ValidRPCDescription<R(Args...), Description...>
        method(const char* name, F&& fn, Description&&... descriptions)
        {
            concreteMethod<R, Args...>(name, std::forward<F>(fn), std::forward<Description>(descriptions)...);
        }

        template<typename F, typename... Description>
        ValidRPCDescription<F, Description...>
        method(const char* name, F&& fn, Description&&... descriptions)
        {
            concreteMethodHelper<typename Private::FunctionTraits<F>::ReturnValue>(
                name, std::forward<F>(fn), typename Private::FunctionTraits<F>::Arguments{},
                std::forward<Description>(descriptions)...);
        }

        template<typename R, typename Class, typename... Args, typename... Description>
        ValidRPCDescription<R (Class::*)(Args...), Description...>
        method(const char* name, R (Class::*fn)(Args...), Class* This, Description&&... descriptions)
        {
            auto func = Private::MemberInvoker<decltype(std::mem_fn(fn)), Class>(std::mem_fn(fn), This);
            concreteMethod<R, Args...>(name, std::move(func), std::forward<Description>(descriptions)...);
        }

        template<typename R, typename Class, typename... Args, typename... Description>
        ValidRPCDescription<R (Class::*)(Args...) const, Description...>
        method(const char* name, R (Class::*fn)(Args...) const, Class* This, Description&&... descriptions)
        {
            auto func = Private::MemberInvoker<decltype(std::mem_fn(fn)), Class>(std::mem_fn(fn), This);
            concreteMethod<R, Args...>(name, std::move(func), std::forward<Description>(descriptions)...);
        }

    private:
        template<typename R, typename... Args, typename F, typename... Description>
        void concreteMethodHelper(const char* name, F&& fn, Private::ArgumentTuple<Args...> args,
                            Description&&... descriptions)
        {
            concreteMethod<R, Args...>(name, std::forward<F>(fn), std::forward<Description>(descriptions)...);
        }

        template<typename R, typename... Args, typename F, typename Comment, typename... Description>
        void concreteMethod(const char* name, F&& fn, Comment&& comment, Description&&... descriptions)
        {
            Method m(makeRPCSignature<R, Args...>(name), std::forward<Comment>(comment));
            m.parameterDesc.reserve(sizeof...(Args));
            addParameterDescription<Args...>(m, std::forward<Description>(descriptions)...);
            m.addInvoker(make_RPCInvoker<BinaryRPCBackend, R, Args...>(fn));
            m.addInvoker(make_RPCInvoker<BinaryRPCBackendLegacy, R, Args...>(fn));
            m.addInvoker(make_RPCInvoker<JSONRPCBackend, R, Args...>(fn));
            addMethod(m);
        }

    private:

        void addMethod(const Method& method) {
            foreach (const MethodInfo& m, mService.methods) {
                // more than 1 method with same name and number of parameters?
                if ((m.signature.name == method.signature.name) &&
                    (m.signature.parameterTypes.size() == method.signature.parameterTypes.size())) {
                    MIRA_LOG(WARNING) << mService.name << "." << m.signature  << " and "
                                      << mService.name << "." << method.signature
                                      << " are ambiguous when called through JSON RPC"
                                      << " - choosing different method names is strongly recommended!";
                }
            }

            if(mService.methods.insert(method).second)
                mAddedMethods.insert(method.signature);
            // adding the same method multiple times, is handled correctly
        }

        template<typename... RECURSIONSTOP>
        typename std::enable_if<(sizeof...(RECURSIONSTOP)) == 0>::type addParameterDescription(Method& m)
        {}

        template<typename ParameterType, typename... Args>
        MIRA_DEPRECATED("_________________Please provide parameter descriptions (and sample values, if appropriate) for RPC methods!_________________",
        void addParameterDescription(Method& m)
        {
            m.parameterSamples.emplace_back(createParameterSample<Private::StrippedType<ParameterType>>());
            addParameterDescription<Args...>(m);
        }
        )

        template<typename ParameterType, typename... Args, class Name, class Description, typename... Tail,
                 typename = typename std::enable_if<(sizeof...(Args)) * 2 == sizeof...(Tail)>::type>
        void addParameterDescription(Method& m, Name&& name, Description&& description, Tail&&... tail)
        {
            m.parameterDesc.emplace_back(std::forward<Name>(name), std::forward<Description>(description));
            m.parameterSamples.emplace_back(createParameterSample<Private::StrippedType<ParameterType>>());
            addParameterDescription<Args...>(m, std::forward<Tail>(tail)...);
        }

        template<typename ParameterType, typename... Args, class Name, class Description, class Example,
                 typename... Tail,
                 typename = typename std::enable_if<(sizeof...(Args)) * 3 == sizeof...(Tail)>::type>
        void addParameterDescription(Method& m, Name&& name, Description&& description, Example&& example,
                                     Tail&&... tail)
        {
            m.parameterDesc.emplace_back(std::forward<Name>(name), std::forward<Description>(description));
            m.parameterSamples.emplace_back(
                createParameterSample<Private::StrippedType<ParameterType>>(std::forward<Example>(example)));
            m.parameterSamplesDefault = false; // this will be assigned multiple times in recursive calls,
                                               // but the alternative seems to be to double the number of methods needed for this 
            addParameterDescription<Args...>(m, std::forward<Tail>(tail)...);
        }

        template<typename T>
        json::Value createParameterSample(const T& v = T())
        {
            return mJSONSerializer.serialize(v);
        }

        JSONSerializer mJSONSerializer;

    private:

        Service& mService;

        std::set<RPCSignature> mAddedMethods;

        std::set<std::string> mAddedInterfaces;
    };

public:

    template <typename T>
    const Service& registerServiceObject(const std::string& serviceName,
                                         T& serviceObject,
                                         std::set<RPCSignature>* addedMethods = NULL,
                                         std::set<std::string>* addedInterfaces = NULL)
    {
        boost::mutex::scoped_lock lock(mMutex);

        // use already created service, or create new one
        ServiceMap::iterator it = mServices.find(serviceName);
        if (it==mServices.end())
            it = mServices.insert(std::make_pair(serviceName,
                                                 Service(serviceName))).first;

        // reflect the service object and add its methods to the service
        Service& s = it->second;
        RPCReflector r(s);
        serviceObject.reflect(r);

        MIRA_LOG(NOTICE) << "Registered service object for service: '" <<
                             serviceName << "'. With the following new methods:";
        foreach(const RPCSignature& m, r.getAddedMethods())
            MIRA_LOG(NOTICE) << "    " << m;

        if(addedMethods!=NULL)
            *addedMethods = r.getAddedMethods();

        if(addedInterfaces!=NULL)
            *addedInterfaces = r.getAddedInterfaces();

        return s;
    }

    void unregisterService(const std::string& serviceName)
    {
        boost::mutex::scoped_lock lock(mMutex);
        MIRA_LOG(NOTICE) << "Unregistering service: '" << serviceName << "'.";
        mServices.erase(serviceName);
    }

public:

    const ServiceMap& getServices() const
    {
        return mServices;
    }

    const Service& getService(const std::string& name) const
    {
        boost::mutex::scoped_lock lock(mMutex);
        auto i = mServices.find(name);
        if (i == mServices.end())
            MIRA_THROW(XInvalidParameter, "No such service '" << name << "'");

        return i->second;
    }

    bool existsService(const std::string& name) const
    {
        boost::mutex::scoped_lock lock(mMutex);
        auto i = mServices.find(name);
        return i!= mServices.end();
    }

    std::list<std::string> queryServicesForInterface(const std::string& interface) const
    {
        boost::mutex::scoped_lock lock(mMutex);
        std::list<std::string> res;
        // for all services
        for(auto it=mServices.begin(); it!=mServices.end(); ++it)
            // does service implements the desired interface?
            if(it->second.interfaces.count(interface)!=0)
                res.push_back(it->second.name); // yes? so add its name to the list
        return res;
    }

    bool implementsInterface(const std::string& name, const std::string& interface) const
    {
        boost::mutex::scoped_lock lock(mMutex);
        auto it = mServices.find(name);
        if (it == mServices.end())
            return false;
        return it->second.interfaces.count(interface) > 0;
    }

public:

    template <typename Backend>
    void processCall(typename Backend::ServerRequest& request,
                     typename Backend::ServerResponse& response)
    {
        boost::shared_ptr<TRPCInvoker<Backend>> invoker;
        Service* service;
        std::string callId;

        Method* method = NULL;
        if(!processCallCommon<Backend>(request, response, callId, invoker, service, method))
            return; // an error has occured, abort, error already is reported in response

        // and finally invoke the method
        response.setHeader(callId);
        try
        {
            invoker->invoke(request, response);
        }
        catch(std::exception& ex) {
            // handle possible exceptions in getParameter() and return them to caller
            MIRA_LOG(WARNING) << "Invalid RPC Request: " << ex.what();
            response.returnException(RPC_INVALID_PARAMS, ex.what());
        }
    }

    template <typename Backend>
    class DeferredInvoker : public AbstractDeferredInvoker
    {
    public:
        DeferredInvoker(typename Backend::ServerRequest& request,
                        typename Backend::ServerResponse& response,
                        const std::string& callId,
                        const std::string& service,
                        const std::string& method,
                        boost::shared_ptr<TRPCInvoker<Backend>> invoker) :
            AbstractDeferredInvoker(callId,service,method),
            mRequest(request), mResponse(response),
            mInvoker(invoker) {}
    public:
        virtual void invoke() {
            mResponse.setHeader(mCallId);
            try
            {
                mInvoker->invoke(mRequest, mResponse);
            }
            catch(std::exception& ex) {
                // handle possible exceptions in getParameter() and return them to caller
                MIRA_LOG(WARNING) << "Invalid RPC Request: " << ex.what();
                mResponse.returnException(RPC_INVALID_PARAMS, ex.what());
            }
            try {
                if(mFinishHandler!=NULL)
                    mFinishHandler->onRPCfinished(this);
            } catch(std::exception& ex) {
                // catch broken promise
                MIRA_LOG_EXCEPTION(ERROR, ex);
            }
        }
    private:
        typename Backend::ServerRequest& mRequest;
        typename Backend::ServerResponse& mResponse;
        boost::shared_ptr<TRPCInvoker<Backend>> mInvoker;
    };

    template <typename Backend>
    AbstractDeferredInvokerPtr processCallDeferred(typename Backend::ServerRequest& request,
                                                   typename Backend::ServerResponse& response)
    {
        boost::shared_ptr<TRPCInvoker<Backend>> invoker;
        Service* service = NULL;
        std::string callId;

        Method* method;

        if(processCallCommon<Backend>(request, response, callId, invoker, service, method)) {
            assert(service!=NULL);
            AbstractDeferredInvokerPtr deferredInvoker(
                new DeferredInvoker<Backend>(request, response, callId,
                                             service->name,
                                             method->signature.name, invoker));
            return deferredInvoker;
        } else {
            // error already is reported in response
            return AbstractDeferredInvokerPtr();
        }
    }

protected:

    template <typename Backend>
    bool processCallCommon(typename Backend::ServerRequest& request,
                           typename Backend::ServerResponse& response,
                           std::string &oCallId,
                           boost::shared_ptr<TRPCInvoker<Backend>>& oInvoker,
                           Service* &oService, Method* &oMethod)
    {
        std::string name;
        std::string callId;
        std::string serviceStr;

        try {
            request.getHeader(callId, serviceStr);
        } catch(std::exception& ex) {
            // handle possible exceptions in getHeader() and return them to caller
            MIRA_LOG(WARNING) << "Invalid RPC Request: " << ex.what();
            generateErrorResponse<Backend>(callId, RPC_INVALID_REQUEST, ex.what(), response);
            return false;
        }

        boost::mutex::scoped_lock lock(mMutex);

        // find the service
        auto serviceIt = mServices.find(serviceStr);

        if(serviceIt==mServices.end()) {
            MIRA_LOG(WARNING) << "RPC service not found: " << serviceStr;
            generateErrorResponse<Backend>(callId, RPC_METHOD_NOT_FOUND,
                                           "Cannot process RPC call, a service named '"
                                           + serviceStr + "' does not exist", response);
            return false;
        }

        Service& service = serviceIt->second;

        // find the method
        oMethod=NULL;
        Method* method = NULL;

        foreach(const Method& m, service.methods)
        {
            if(request.checkSignature(m.signature)) {
                // found method with the correct signature
                method = const_cast<Method*>(&m);
                break; // break for each
            }
        }

        // abort, if we did not find the method, but produce a (hopefully)
        // helpful exception message
        if(method==NULL) {
            std::string methodName = request.getSignature().name;
            std::string candidates;
            foreach(const Method& m, service.methods)
            {
                if(methodName == m.signature.name) {
                    if(!candidates.empty())
                        candidates += ", ";
                    candidates += toString(m.signature);
                }
            }

            if(!candidates.empty())
                candidates = " Candidates are: " + candidates;
            else
                candidates = " No candidates found.";

            MIRA_LOG(WARNING) << "RPC method does not exist: "
                              << toString(request.getSignature())
                              << ", candidates: " << candidates;
            generateErrorResponse<Backend>(callId,
                                           RPC_INVALID_PARAMS, "Cannot process RPC call, the service '" +
                                           serviceStr + "' does not have a method '" +
                                           toString(request.getSignature()) + "'." + candidates,
                                           response);
            return false;
        }

        // now get the invoker for our backend
        auto invokerIt = method->invokers.find(typeId<Backend>());
        if(invokerIt == method->invokers.end())
            MIRA_THROW(XLogical, "Cannot process RPC call, no invoker for service method '" <<
                                 serviceStr << "." << request.getSignature() <<
                                 "' was registered for backend type '" <<
                                 typeName<Backend>() << "'.");
        // the above exception is not send to the caller, since it is caused by
        // a problem here in our server

        // cast invoker into typed invoker for our Backend
        oInvoker = boost::static_pointer_cast<TRPCInvoker<Backend>>(invokerIt->second);
        oCallId = callId;
        oService = &service;
        oMethod = method;
        return true;
    }

    template <typename Backend>
    void generateErrorResponse(const std::string& callId,
                               RPCError reason, const std::string& message,
                               typename Backend::ServerResponse& oResponse)
    {
        oResponse.setHeader(callId);
        oResponse.returnException(reason, message);
    }

private:

    ServiceMap mServices;
    mutable boost::mutex mMutex;
};


} // namespace

#endif /* _MIRA_RPCSERVER_H_ */