GRPCServer.h

#pragma once
 
#include <xrpld/app/main/Application.h>
#include <xrpld/rpc/Context.h>
#include <xrpld/rpc/GRPCHandlers.h>
#include <xrpld/rpc/Role.h>
#include <xrpld/rpc/detail/Handler.h>
 
#include <xrpl/core/JobQueue.h>
#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
#include <xrpl/resource/Charge.h>
#include <xrpl/server/InfoSub.h>
 
#include <grpcpp/grpcpp.h>
 
namespace xrpl {
 
// Interface that CallData implements
class Processor
{
public:
    virtual ~Processor() = default;
 
    Processor() = default;
 
    Processor(Processor const&) = delete;
 
    Processor&
    operator=(Processor const&) = delete;
 
    // process a request that has arrived. Can only be called once per instance
    virtual void
    process() = 0;
 
    // create a new instance of this CallData object, with the same type
    //(same template parameters) as original. This is called when a CallData
    // object starts processing a request. Creating a new instance allows the
    // server to handle additional requests while the first is being processed
    virtual std::shared_ptr<Processor>
    clone() = 0;
 
    // true if this object has finished processing the request. Object will be
    // deleted once this function returns true
    virtual bool
    isFinished() = 0;
};
 
class GRPCServerImpl final
{
private:
    // CompletionQueue returns events that have occurred, or events that have
    // been cancelled
    std::unique_ptr<grpc::ServerCompletionQueue> cq_;
 
    std::vector<std::shared_ptr<Processor>> requests_;
 
    // The gRPC service defined by the .proto files
    org::xrpl::rpc::v1::XRPLedgerAPIService::AsyncService service_;
 
    std::unique_ptr<grpc::Server> server_;
 
    Application& app_;
 
    std::string serverAddress_;
    std::uint16_t serverPort_ = 0;
 
    std::vector<boost::asio::ip::address> secureGatewayIPs_;
 
    beast::Journal journal_;
 
    // typedef for function to bind a listener
    // This is always of the form:
    // org::xrpl::rpc::v1::XRPLedgerAPIService::AsyncService::Request[RPC NAME]
    template <class Request, class Response>
    using BindListener = std::function<void(
        org::xrpl::rpc::v1::XRPLedgerAPIService::AsyncService&,
        grpc::ServerContext*,
        Request*,
        grpc::ServerAsyncResponseWriter<Response>*,
        grpc::CompletionQueue*,
        grpc::ServerCompletionQueue*,
        void*)>;
 
    // typedef for actual handler (that populates a response)
    // handlers are defined in rpc/GRPCHandlers.h
    template <class Request, class Response>
    using Handler = std::function<std::pair<Response, grpc::Status>(RPC::GRPCContext<Request>&)>;
    // This implementation is currently limited to v1 of the API
    static unsigned constexpr apiVersion = 1;
 
    template <class Request, class Response>
    using Forward = std::function<
        grpc::Status(org::xrpl::rpc::v1::XRPLedgerAPIService::Stub*, grpc::ClientContext*, Request, Response*)>;
 
public:
    explicit GRPCServerImpl(Application& app);
 
    GRPCServerImpl(GRPCServerImpl const&) = delete;
 
    GRPCServerImpl&
    operator=(GRPCServerImpl const&) = delete;
 
    void
    shutdown();
 
    // setup the server and listeners
    // returns true if server started successfully
    bool
    start();
 
    // the main event loop
    void
    handleRpcs();
 
    // Create a CallData object for each RPC. Return created objects in vector
    std::vector<std::shared_ptr<Processor>>
    setupListeners();
 
    // Obtaining actually binded endpoint (if port 0 was used for server setup).
    boost::asio::ip::tcp::endpoint
    getEndpoint() const;
 
private:
    // Class encompassing the state and logic needed to serve a request.
    template <class Request, class Response>
    class CallData : public Processor, public std::enable_shared_from_this<CallData<Request, Response>>
    {
    private:
        // The means of communication with the gRPC runtime for an asynchronous
        // server.
        org::xrpl::rpc::v1::XRPLedgerAPIService::AsyncService& service_;
 
        // The producer-consumer queue for asynchronous server notifications.
        grpc::ServerCompletionQueue& cq_;
 
        // Context for the rpc, allowing to tweak aspects of it such as the use
        // of compression, authentication, as well as to send metadata back to
        // the client.
        grpc::ServerContext ctx_;
 
        // true if finished processing request
        // Note, this variable does not need to be atomic, since it is
        // currently only accessed from one thread. However, isFinished(),
        // which returns the value of this variable, is public facing. In the
        // interest of avoiding future concurrency bugs, we make it atomic.
        std::atomic_bool finished_;
 
        Application& app_;
 
        // What we get from the client.
        Request request_;
 
        // The means to get back to the client.
        grpc::ServerAsyncResponseWriter<Response> responder_;
 
        // Function that creates a listener for specific request type
        BindListener<Request, Response> bindListener_;
 
        // Function that processes a request
        Handler<Request, Response> handler_;
 
        // Function to call to forward to another server
        Forward<Request, Response> forward_;
 
        // Condition required for this RPC
        RPC::Condition requiredCondition_;
 
        // Load type for this RPC
        Resource::Charge loadType_;
 
        std::vector<boost::asio::ip::address> const& secureGatewayIPs_;
 
    public:
        virtual ~CallData() = default;
 
        // Take in the "service" instance (in this case representing an
        // asynchronous server) and the completion queue "cq" used for
        // asynchronous communication with the gRPC runtime.
        explicit CallData(
            org::xrpl::rpc::v1::XRPLedgerAPIService::AsyncService& service,
            grpc::ServerCompletionQueue& cq,
            Application& app,
            BindListener<Request, Response> bindListener,
            Handler<Request, Response> handler,
            Forward<Request, Response> forward,
            RPC::Condition requiredCondition,
            Resource::Charge loadType,
            std::vector<boost::asio::ip::address> const& secureGatewayIPs);
 
        CallData(CallData const&) = delete;
 
        CallData&
        operator=(CallData const&) = delete;
 
        virtual void
        process() override;
 
        virtual bool
        isFinished() override;
 
        std::shared_ptr<Processor>
        clone() override;
 
    private:
        // process the request. Called inside the coroutine passed to JobQueue
        void
        process(std::shared_ptr<JobQueue::Coro> coro);
 
        // return load type of this RPC
        Resource::Charge
        getLoadType();
 
        // return the Role used for this RPC
        Role
        getRole(bool isUnlimited);
 
        // register endpoint with ResourceManager and return usage
        Resource::Consumer
        getUsage();
 
        // Returns the ip of the client
        // Empty optional if there was an error decoding the client ip
        std::optional<boost::asio::ip::address>
        getClientIpAddress();
 
        // Returns the endpoint of the client.
        // Empty optional if there was an error decoding the client
        // endpoint
        std::optional<boost::asio::ip::tcp::endpoint>
        getClientEndpoint();
 
        // If the request was proxied through
        // another rippled node, returns the ip of the originating client.
        // Empty optional if request was not proxied or there was an error
        // decoding the client ip
        std::optional<boost::asio::ip::address>
        getProxiedClientIpAddress();
 
        // If the request was proxied through
        // another rippled node, returns the endpoint of the originating client.
        // Empty optional if request was not proxied or there was an error
        // decoding the client endpoint
        std::optional<boost::asio::ip::tcp::endpoint>
        getProxiedClientEndpoint();
 
        // Returns the user specified in the request. Empty optional if no user
        // was specified
        std::optional<std::string>
        getUser();
 
        // Sets is_unlimited in response to value of clientIsUnlimited
        // Does nothing if is_unlimited is not a field of the response
        void
        setIsUnlimited(Response& response, bool isUnlimited);
 
        // True if the client is exempt from resource controls
        bool
        clientIsUnlimited();
 
        // True if the request was proxied through another rippled node prior
        // to arriving here
        bool
        wasForwarded();
 
        // forward request to a p2p node
        void
        forwardToP2p(RPC::GRPCContext<Request>& context);
 
    };  // CallData
 
};  // GRPCServerImpl
 
class GRPCServer
{
public:
    explicit GRPCServer(Application& app) : impl_(app)
    {
    }
 
    GRPCServer(GRPCServer const&) = delete;
 
    GRPCServer&
    operator=(GRPCServer const&) = delete;
 
    bool
    start();
 
    void
    stop();
 
    ~GRPCServer();
 
    boost::asio::ip::tcp::endpoint
    getEndpoint() const;
 
private:
    GRPCServerImpl impl_;
    std::thread thread_;
    bool running_ = false;
};
}  // namespace xrpl