add grpc server implementation (#15)

* implement GatewayClient * update client implement * add grpc server implementation
WeBankBlockchain · Sep 4, 2024 · 58a61ac · 58a61ac
1 parent 55da761
commit 58a61ac
Show file tree

Hide file tree

Showing 25 changed files with 700 additions and 150 deletions.
diff --git a/cpp/cmake/TargetSettings.cmake b/cpp/cmake/TargetSettings.cmake
@@ -23,9 +23,12 @@ set(TARS_PROTOCOL_TARGET "wedpr-tars-protocol")
 # wedpr-protocol/protobuf
 set(PB_PROTOCOL_TARGET "wedpr-pb-protocol")
 
-# wedpr-protocol/grpc-client
-set(SERVICE_CLIENT_TARGET "service-client")
-set(SERVICE_CLIENT_PB_TARGET "service-client-pb")
+# wedpr-protocol/grpc/client
+set(SERVICE_CLIENT_TARGET "wedpr-client")
+# wedpr-protocol/grpc/server
+set(SERVICE_SERVER_TARGET "wedpr-server")
+# wedpr-protocol/proto generated file
+set(SERVICE_PB_TARGET "service-pb")
 
 # ppc-front
 SET(FRONT_TARGET "ppc-front")

diff --git a/cpp/ppc-framework/gateway/IGateway.h b/cpp/ppc-framework/gateway/IGateway.h
@@ -63,10 +63,12 @@ class IGateway
 
     virtual void asyncSendbroadcastMessage(ppc::protocol::RouteType routeType,
         ppc::protocol::MessageOptionalHeader::Ptr const& routeInfo, bcos::bytes&& payload) = 0;
-    virtual void registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo) = 0;
-    virtual void unRegisterNodeInfo(bcos::bytesConstRef nodeID) = 0;
-    virtual void registerTopic(bcos::bytesConstRef nodeID, std::string const& topic) = 0;
-    virtual void unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic) = 0;
+    virtual bcos::Error::Ptr registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo) = 0;
+    virtual bcos::Error::Ptr unRegisterNodeInfo(bcos::bytesConstRef nodeID) = 0;
+    virtual bcos::Error::Ptr registerTopic(
+        bcos::bytesConstRef nodeID, std::string const& topic) = 0;
+    virtual bcos::Error::Ptr unRegisterTopic(
+        bcos::bytesConstRef nodeID, std::string const& topic) = 0;
 };
 
 }  // namespace ppc::gateway
diff --git a/cpp/ppc-framework/protocol/INodeInfo.h b/cpp/ppc-framework/protocol/INodeInfo.h
@@ -40,6 +40,8 @@ class INodeInfo
 
     virtual std::string const& endPoint() const = 0;
     virtual bcos::bytesConstRef nodeID() const = 0;
+    virtual void setNodeID(bcos::bytesConstRef nodeID) = 0;
+    virtual void setEndPoint(std::string const& endPoint) = 0;
 
     // components
     virtual void setComponents(std::set<std::string> const& components) = 0;

diff --git a/cpp/ppc-gateway/ppc-gateway/gateway/GatewayImpl.cpp b/cpp/ppc-gateway/ppc-gateway/gateway/GatewayImpl.cpp
@@ -183,20 +183,26 @@ void GatewayImpl::onReceiveBroadcastMessage(MessageFace::Ptr msg, WsSession::Ptr
     m_localRouter->dispatcherMessage(p2pMessage, nullptr);
 }
 
-void GatewayImpl::registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo)
+bcos::Error::Ptr GatewayImpl::registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo)
 {
     m_localRouter->registerNodeInfo(nodeInfo);
+    return nullptr;
 }
 
-void GatewayImpl::unRegisterNodeInfo(bcos::bytesConstRef nodeID)
+bcos::Error::Ptr GatewayImpl::unRegisterNodeInfo(bcos::bytesConstRef nodeID)
 {
     m_localRouter->unRegisterNode(nodeID.toBytes());
+    return nullptr;
 }
-void GatewayImpl::registerTopic(bcos::bytesConstRef nodeID, std::string const& topic)
+
+bcos::Error::Ptr GatewayImpl::registerTopic(bcos::bytesConstRef nodeID, std::string const& topic)
 {
     m_localRouter->registerTopic(nodeID, topic);
+    return nullptr;
 }
-void GatewayImpl::unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic)
+
+bcos::Error::Ptr GatewayImpl::unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic)
 {
     m_localRouter->unRegisterTopic(nodeID, topic);
+    return nullptr;
 }
diff --git a/cpp/ppc-gateway/ppc-gateway/gateway/GatewayImpl.h b/cpp/ppc-gateway/ppc-gateway/gateway/GatewayImpl.h
@@ -61,10 +61,10 @@ class GatewayImpl : public IGateway, public std::enable_shared_from_this<Gateway
         ppc::protocol::MessageOptionalHeader::Ptr const& routeInfo, bcos::bytes&& payload) override;
 
 
-    void registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo) override;
-    void unRegisterNodeInfo(bcos::bytesConstRef nodeID) override;
-    void registerTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
-    void unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
+    bcos::Error::Ptr registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo) override;
+    bcos::Error::Ptr unRegisterNodeInfo(bcos::bytesConstRef nodeID) override;
+    bcos::Error::Ptr registerTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
+    bcos::Error::Ptr unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
 
 protected:
     virtual void onReceiveP2PMessage(

diff --git a/cpp/wedpr-protocol/grpc/CMakeLists.txt b/cpp/wedpr-protocol/grpc/CMakeLists.txt
@@ -26,7 +26,8 @@ foreach(proto_file ${MESSAGES_PROTOS})
     )
 endforeach()
 
-add_library(${SERVICE_CLIENT_PB_TARGET} ${GRPC_MESSAGES_SRCS})
-target_link_libraries(${SERVICE_CLIENT_PB_TARGET} PUBLIC ${PB_PROTOCOL_TARGET} gRPC::grpc++_unsecure)
+add_library(${SERVICE_PB_TARGET} ${GRPC_MESSAGES_SRCS})
+target_link_libraries(${SERVICE_PB_TARGET} PUBLIC ${PB_PROTOCOL_TARGET} gRPC::grpc++_unsecure)
 
-add_subdirectory(client)
+add_subdirectory(client)
+add_subdirectory(server)
diff --git a/cpp/wedpr-protocol/grpc/client/CMakeLists.txt b/cpp/wedpr-protocol/grpc/client/CMakeLists.txt
@@ -1,3 +1,3 @@
 file(GLOB_RECURSE SRCS *.cpp)
 add_library(${SERVICE_CLIENT_TARGET} ${SRCS})
-target_link_libraries(${SERVICE_CLIENT_TARGET} PUBLIC ${SERVICE_CLIENT_PB_TARGET} ${PB_PROTOCOL_TARGET})
+target_link_libraries(${SERVICE_CLIENT_TARGET} PUBLIC ${SERVICE_PB_TARGET} ${PB_PROTOCOL_TARGET})
diff --git a/cpp/wedpr-protocol/grpc/client/Common.h b/cpp/wedpr-protocol/grpc/client/Common.h
@@ -20,4 +20,5 @@
 #pragma once
 #include "ppc-framework/Common.h"
 
-#define GRPC_CLIENT_LOG(LEVEL) BCOS_LOG(LEVEL) << "[GRPC][CLIENT]"
+#define GRPC_CLIENT_LOG(LEVEL) BCOS_LOG(LEVEL) << "[GRPC][CLIENT]"
+#define GATEWAY_CLIENT_LOG(LEVEL) BCOS_LOG(LEVEL) << "[GATEWAY][CLIENT]"
diff --git a/cpp/wedpr-protocol/grpc/client/FrontClient.cpp b/cpp/wedpr-protocol/grpc/client/FrontClient.cpp
@@ -18,16 +18,13 @@
  * @date 2024-09-02
  */
 #include "FrontClient.h"
+#include "protobuf/RequestConverter.h"
 #include "wedpr-protocol/protobuf/Common.h"
 
 
 using namespace ppc::protocol;
 using namespace ppc::proto;
-using grpc::Channel;
-using grpc::ClientAsyncResponseReader;
-using grpc::ClientContext;
-using grpc::CompletionQueue;
-using grpc::Status;
+using namespace grpc;
 
 void FrontClient::onReceiveMessage(ppc::protocol::Message::Ptr const& msg, ReceiveMsgFunc callback)
 {
@@ -37,15 +34,8 @@ void FrontClient::onReceiveMessage(ppc::protocol::Message::Ptr const& msg, Recei
     msg->encode(encodedData);
     receivedMsg.set_data(encodedData.data(), encodedData.size());
 
-    auto grpcCallback = [callback](ClientContext const&, Status const& status, Error&& response) {
-        auto error = std::make_shared<bcos::Error>(response.errorcode(), response.errormessage());
-        callback(error);
-    };
-
-    auto call = std::make_shared<AsyncClientCall>(grpcCallback);
-    call->responseReader =
-        m_stub->PrepareAsynconReceiveMessage(&call->context, receivedMsg, &m_client->queue());
-    call->responseReader->StartCall();
-    // send request, upon completion of the RPC, "reply" be updated with the server's response
-    call->responseReader->Finish(&call->reply, &call->status, (void*)call.get());
+    ClientContext context;
+    auto response = std::make_shared<Error>();
+    m_stub->async()->onReceiveMessage(&context, &receivedMsg, response.get(),
+        [response, callback](Status status) { callback(toError(status, std::move(*response))); });
 }
diff --git a/cpp/wedpr-protocol/grpc/client/FrontClient.h b/cpp/wedpr-protocol/grpc/client/FrontClient.h
@@ -23,20 +23,17 @@
 
 namespace ppc::protocol
 {
-class FrontClient : public ppc::front::IFrontClient
+class FrontClient : public ppc::front::IFrontClient, public GrpcClient
 {
 public:
     using Ptr = std::shared_ptr<FrontClient>;
-    FrontClient(GrpcClient::Ptr client)
-      : m_client(std::move(client)), m_stub(ppc::proto::Front::NewStub(m_client->channel()))
-    {}
+    FrontClient(std::shared_ptr<grpc::Channel> channel) : GrpcClient(std::move(channel)) {}
 
     ~FrontClient() override = default;
     void onReceiveMessage(
         ppc::protocol::Message::Ptr const& _msg, ppc::protocol::ReceiveMsgFunc _callback) override;
 
 private:
     std::unique_ptr<ppc::proto::Front::Stub> m_stub;
-    GrpcClient::Ptr m_client;
 };
 }  // namespace ppc::protocol
diff --git a/cpp/wedpr-protocol/grpc/client/GatewayClient.cpp b/cpp/wedpr-protocol/grpc/client/GatewayClient.cpp
@@ -18,24 +18,59 @@
  * @date 2024-09-02
  */
 #include "GatewayClient.h"
+#include "Common.h"
+#include "protobuf/RequestConverter.h"
 
 using namespace ppc;
+using namespace ppc::proto;
+using namespace grpc;
 using namespace ppc::gateway;
 using namespace ppc::protocol;
 
-
-void GatewayClient::start() {}
-void GatewayClient::stop() {}
-
 void GatewayClient::asyncSendMessage(RouteType routeType,
     MessageOptionalHeader::Ptr const& routeInfo, bcos::bytes&& payload, long timeout,
     ReceiveMsgFunc callback)
-{}
+{
+    auto request = generateRequest(routeType, routeInfo, std::move(payload), timeout);
+    ClientContext context;
+    auto response = std::make_shared<Error>();
+    m_stub->async()->asyncSendMessage(&context, request.get(), response.get(),
+        [callback, response](Status status) { callback(toError(status, std::move(*response))); });
+}
+
+
+bcos::Error::Ptr GatewayClient::registerNodeInfo(INodeInfo::Ptr const& nodeInfo)
+{
+    auto request = toNodeInfoRequest(nodeInfo);
+    ClientContext context;
+    std::shared_ptr<ppc::proto::Error> response = std::make_shared<ppc::proto::Error>();
+    auto status = m_stub->registerNodeInfo(&context, *request, response.get());
+    return toError(status, std::move(*response));
+}
+
+bcos::Error::Ptr GatewayClient::unRegisterNodeInfo(bcos::bytesConstRef nodeID)
+{
+    auto request = toNodeInfoRequest(nodeID, "");
+    ClientContext context;
+    std::shared_ptr<ppc::proto::Error> response = std::make_shared<ppc::proto::Error>();
+    auto status = m_stub->unRegisterNodeInfo(&context, *request, response.get());
+    return toError(status, std::move(*response));
+}
+bcos::Error::Ptr GatewayClient::registerTopic(bcos::bytesConstRef nodeID, std::string const& topic)
+{
+    auto request = toNodeInfoRequest(nodeID, topic);
+    ClientContext context;
+    std::shared_ptr<ppc::proto::Error> response = std::make_shared<ppc::proto::Error>();
+    auto status = m_stub->registerTopic(&context, *request, response.get());
+    return toError(status, std::move(*response));
+}
 
-void GatewayClient::asyncSendbroadcastMessage(
-    RouteType routeType, MessageOptionalHeader::Ptr const& routeInfo, bcos::bytes&& payload)
-{}
-void GatewayClient::registerNodeInfo(INodeInfo::Ptr const& nodeInfo) {}
-void GatewayClient::unRegisterNodeInfo(bcos::bytesConstRef nodeID) {}
-void GatewayClient::registerTopic(bcos::bytesConstRef nodeID, std::string const& topic) {}
-void GatewayClient::unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic) {}
+bcos::Error::Ptr GatewayClient::unRegisterTopic(
+    bcos::bytesConstRef nodeID, std::string const& topic)
+{
+    auto request = toNodeInfoRequest(nodeID, topic);
+    ClientContext context;
+    std::shared_ptr<ppc::proto::Error> response = std::make_shared<ppc::proto::Error>();
+    auto status = m_stub->unRegisterTopic(&context, *request, response.get());
+    return toError(status, std::move(*response));
+}
diff --git a/cpp/wedpr-protocol/grpc/client/GatewayClient.h b/cpp/wedpr-protocol/grpc/client/GatewayClient.h
@@ -18,20 +18,24 @@
  * @date 2024-09-02
  */
 #pragma once
+#include "GrpcClient.h"
 #include "ppc-framework/gateway/IGateway.h"
 
 namespace ppc::protocol
 {
-class GatewayClient : public ppc::gateway::IGateway
+class GatewayClient : public ppc::gateway::IGateway, public GrpcClient
 {
 public:
     using Ptr = std::shared_ptr<GatewayClient>;
-    GatewayClient() = default;
+    GatewayClient(std::shared_ptr<grpc::Channel> channel)
+      : GrpcClient(std::move(channel)), m_stub(ppc::proto::Gateway::NewStub(m_channel))
+    {}
+
     ~GatewayClient() override = default;
 
 
-    void start() override;
-    void stop() override;
+    void start() override {}
+    void stop() override {}
 
     /**
      * @brief send message to gateway
@@ -52,10 +56,14 @@ class GatewayClient : public ppc::gateway::IGateway
         long timeout, ppc::protocol::ReceiveMsgFunc callback) override;
 
     void asyncSendbroadcastMessage(ppc::protocol::RouteType routeType,
-        ppc::protocol::MessageOptionalHeader::Ptr const& routeInfo, bcos::bytes&& payload) override;
-    void registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo) override;
-    void unRegisterNodeInfo(bcos::bytesConstRef nodeID) override;
-    void registerTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
-    void unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
+        ppc::protocol::MessageOptionalHeader::Ptr const& routeInfo, bcos::bytes&& payload) override
+    {}
+    bcos::Error::Ptr registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo) override;
+    bcos::Error::Ptr unRegisterNodeInfo(bcos::bytesConstRef nodeID) override;
+    bcos::Error::Ptr registerTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
+    bcos::Error::Ptr unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic) override;
+
+private:
+    std::unique_ptr<ppc::proto::Gateway::Stub> m_stub;
 };
 }  // namespace ppc::protocol
diff --git a/cpp/wedpr-protocol/grpc/client/GrpcClient.cpp b/cpp/wedpr-protocol/grpc/client/GrpcClient.cpp