rpc.hpp

Go to the documentation of this file.

#ifndef KAGOME_RPC_HPP
#define KAGOME_RPC_HPP

#include <functional>
#include <memory>

#include "common/buffer.hpp"
#include "libp2p/basic/readwriter.hpp"
#include "libp2p/host/host.hpp"
#include "libp2p/peer/peer_info.hpp"
#include "libp2p/peer/protocol.hpp"
#include "log/logger.hpp"
#include "outcome/outcome.hpp"

namespace kagome::network {
  template <typename MessageReadWriterT>
  struct RPC {
    template <typename Request, typename Response>
    static void read(std::shared_ptr<libp2p::basic::ReadWriter> read_writer,
                     std::function<outcome::result<Response>(Request)> cb,
                     std::function<void(outcome::result<void>)> error_cb) {
      auto msg_read_writer =
          std::make_shared<MessageReadWriterT>(std::move(read_writer));
      msg_read_writer->template read<Request>(
          [msg_read_writer, cb = std::move(cb), error_cb = std::move(error_cb)](
              auto &&request_res) mutable {
            if (!request_res) {
              return error_cb(request_res.error());
            }

            auto response_res = cb(std::move(request_res.value()));
            if (!response_res) {
              return error_cb(response_res.error());
            }

            msg_read_writer->template write<Response>(
                response_res.value(),
                [error_cb = std::move(error_cb)](auto &&write_res) {
                  if (!write_res) {
                    return error_cb(write_res.error());
                  }
                });
          });
    }

    template <typename Request>
    static void read(std::shared_ptr<libp2p::basic::ReadWriter> read_writer,
                     std::function<void(outcome::result<Request>)> cb) {
      auto msg_read_writer =
          std::make_shared<MessageReadWriterT>(std::move(read_writer));
      msg_read_writer->template read<Request>(
          [msg_read_writer, cb = std::move(cb)](auto &&msg_res) mutable {
            if (!msg_res) {
              return cb(msg_res.error());
            }

            cb(std::move(msg_res.value()));
          });
    }

    template <typename Request, typename Response>
    static void write(libp2p::Host &host,
                      const libp2p::peer::PeerInfo &peer_info,
                      const libp2p::peer::Protocol &protocol,
                      Request request,
                      std::function<void(outcome::result<Response>)> cb) {
      host.newStream(
          peer_info.id,
          {protocol},
          [request = std::move(request),
           cb = std::move(cb)](auto &&stream_res) mutable {
            if (!stream_res) {
              return cb(stream_res.error());
            }

            auto stream_and_proto = std::move(stream_res.value());
            auto &stream = stream_and_proto.stream;

            auto log = log::createLogger("rpc_writter", "network");
            SL_DEBUG(log,
                     "Sending blocks request to {}",
                     stream->remotePeerId().value().toBase58());

            auto read_writer = std::make_shared<MessageReadWriterT>(stream);
            read_writer->template write<Request>(
                request,
                [read_writer, stream = std::move(stream), cb = std::move(cb)](
                    auto &&write_res) mutable {
                  if (!write_res) {
                    stream->reset();
                    return cb(write_res.error());
                  }

                  auto log = log::createLogger("rpc_writter", "network");
                  SL_DEBUG(log,
                           "Request to {} sent successfully",
                           stream->remotePeerId().value().toBase58());

                  read_writer->template read<Response>(
                      [stream = std::move(stream),
                       cb = std::move(cb)](auto &&msg_res) {
                        if (!msg_res) {
                          stream->reset();
                          return cb(msg_res.error());
                        }

                        stream->close([](auto &&) {});
                        return cb(std::move(msg_res.value()));
                      });
                });
          });
    }

    template <typename Request>
    static void write(libp2p::Host &host,
                      const libp2p::peer::PeerInfo &peer_info,
                      const libp2p::peer::Protocol &protocol,
                      Request request,
                      std::function<void(outcome::result<void>)> cb) {
      host.newStream(peer_info.id,
                     {protocol},
                     [request = std::move(request),
                      cb = std::move(cb)](auto &&stream_res) mutable {
                       if (!stream_res) {
                         return cb(stream_res.error());
                       }

                       auto stream_and_proto = std::move(stream_res.value());
                       auto read_writer = std::make_shared<MessageReadWriterT>(
                           stream_and_proto.stream);
                       read_writer->template write<Request>(
                           request,
                           [stream = std::move(stream_and_proto.stream),
                            cb = std::move(cb)](auto &&write_res) {
                             if (!write_res) {
                               stream->reset();
                               return cb(write_res.error());
                             }

                             stream->close([](auto &&) {});
                             return cb(outcome::success());
                           });
                     });
    }
  };
}  // namespace kagome::network

#endif  // KAGOME_RPC_HPP