gRPC interface and segment reporter refactor (#60)

Author: Shikugawa

cpp2sky/internal/BUILD

@@ -22,3 +22,9 @@ cc_library(
   ],
   visibility = ["//visibility:public"],
 )
+
+cc_library(
+  name = "stream_builder_interface",
+  hdrs = ["stream_builder.h"],
+  visibility = ["//visibility:public"],
+)

cpp2sky/internal/async_client.h

@@ -15,31 +15,15 @@
 #pragma once
 
 #include <google/protobuf/message.h>
+#include <grpcpp/generic/generic_stub.h>
 #include <grpcpp/grpcpp.h>
 
-#include <condition_variable>
 #include <memory>
 
 using google::protobuf::Message;
 
 namespace cpp2sky {
 
-template <class RequestType, class ResponseType>
-class TracerStub {
- public:
-  virtual ~TracerStub() = default;
-
-  /**
-   * Initialize request writer.
-   */
-  virtual std::unique_ptr<grpc::ClientAsyncWriter<RequestType>> createWriter(
-      grpc::ClientContext* ctx, ResponseType* response,
-      grpc::CompletionQueue* cq, void* tag) = 0;
-};
-
-template <class RequestType, class ResponseType>
-using TracerStubPtr = std::unique_ptr<TracerStub<RequestType, ResponseType>>;
-
 template <class RequestType, class ResponseType>
 class AsyncClient {
  public:
@@ -51,91 +35,105 @@ class AsyncClient {
   virtual void sendMessage(RequestType message) = 0;
 
   /**
-   * Get writer.
+   * Drain pending message.
    */
-  virtual std::unique_ptr<grpc::ClientAsyncWriter<RequestType>> createWriter(
-      grpc::ClientContext* ctx, ResponseType* response, void* tag) = 0;
+  virtual void drainPendingMessage(RequestType message) = 0;
 
   /**
-   * Peer address of current gRPC client..
+   * Start stream if there is no living stream.
    */
-  virtual std::string peerAddress() = 0;
+  virtual void startStream() = 0;
 
   /**
-   * Drain pending message.
+   * Completion queue.
    */
-  virtual void drainPendingMessage(RequestType message) = 0;
+  virtual grpc::CompletionQueue& completionQueue() = 0;
 
   /**
-   * Reset stream if it is living.
+   * gRPC Stub
    */
-  virtual void resetStream() = 0;
+  virtual grpc::TemplatedGenericStub<RequestType, ResponseType>& stub() = 0;
+};
 
-  /**
-   * Start stream if there is no living stream.
-   */
-  virtual void startStream() = 0;
+template <class RequestType, class ResponseType>
+using AsyncClientPtr = std::unique_ptr<AsyncClient<RequestType, ResponseType>>;
+
+template <class RequestType, class ResponseType>
+class AsyncStream {
+ public:
+  virtual ~AsyncStream() = default;
 
   /**
-   * The number of drained pending messages.
+   * Send message. It will move the state from Init to Write.
    */
-  virtual size_t numOfMessages() = 0;
+  virtual void sendMessage(RequestType message) = 0;
 };
 
-enum class Operation : uint8_t {
+enum class StreamState : uint8_t {
   Initialized = 0,
-  Connected = 1,
+  Ready = 1,
   Idle = 2,
   WriteDone = 3,
+  ReadDone = 4,
 };
 
-template <class RequestType, class ResponseType>
-using AsyncClientPtr = std::unique_ptr<AsyncClient<RequestType, ResponseType>>;
-
-template <class RequestType>
-class AsyncStream {
+class AsyncStreamCallback {
  public:
-  virtual ~AsyncStream() = default;
+  /**
+   * Callback when stream ready event occured.
+   */
+  virtual void onReady() = 0;
 
   /**
-   * Start stream. It will move the state of stream to Init.
+   * Callback when idle event occured.
    */
-  virtual bool startStream() = 0;
+  virtual void onIdle() = 0;
 
   /**
-   * Send message. It will move the state from Init to Write.
+   * Callback when write done event occured.
    */
-  virtual void sendMessage(RequestType message) = 0;
+  virtual void onWriteDone() = 0;
 
   /**
-   * Restore drained message.
+   * Callback when read done event occured.
    */
-  virtual void undrainMessage(RequestType message) = 0;
+  virtual void onReadDone() = 0;
 
   /**
-   * Handle incoming event related to this stream.
+   * Callback when stream had finished with arbitrary error.
    */
-  virtual void handleOperation(Operation incoming_op) = 0;
+  virtual void onStreamFinish() = 0;
 };
 
-template <class RequestType>
-using AsyncStreamPtr = std::shared_ptr<AsyncStream<RequestType>>;
-
-template <class RequestType, class ResponseType>
-class AsyncStreamFactory {
+struct StreamCallbackTag {
  public:
-  virtual ~AsyncStreamFactory() = default;
-
-  /**
-   * Create async stream entity
-   */
-  virtual AsyncStreamPtr<RequestType> create(
-      AsyncClient<RequestType, ResponseType>* client,
-      std::condition_variable& cv) = 0;
+  void callback(bool stream_finished) {
+    if (stream_finished) {
+      callback_->onStreamFinish();
+      return;
+    }
+
+    switch (state_) {
+      case StreamState::Ready:
+        callback_->onReady();
+        break;
+      case StreamState::WriteDone:
+        callback_->onWriteDone();
+        break;
+      case StreamState::Idle:
+        callback_->onIdle();
+        break;
+      case StreamState::ReadDone:
+        callback_->onReadDone();
+        break;
+    }
+  }
+
+  StreamState state_;
+  AsyncStreamCallback* callback_;
 };
 
 template <class RequestType, class ResponseType>
-using AsyncStreamFactoryPtr =
-    std::unique_ptr<AsyncStreamFactory<RequestType, ResponseType>>;
+using AsyncStreamPtr = std::shared_ptr<AsyncStream<RequestType, ResponseType>>;
 
 }  // namespace cpp2sky

cpp2sky/internal/stream_builder.h

@@ -0,0 +1,56 @@
+// Copyright 2021 SkyAPM
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+
+//     http://www.apache.org/licenses/LICENSE-2.0
+
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <grpcpp/grpcpp.h>
+
+#include <condition_variable>
+
+namespace cpp2sky {
+
+template <class RequestType, class ResponseType>
+class ClientStreamingStreamBuilder {
+ public:
+  virtual ~ClientStreamingStreamBuilder() = default;
+
+  /**
+   * Create async stream entity
+   */
+  virtual AsyncStreamPtr<RequestType, ResponseType> create(
+      AsyncClient<RequestType, ResponseType>& client,
+      std::condition_variable& cv) = 0;
+};
+
+template <class RequestType, class ResponseType>
+using ClientStreamingStreamBuilderPtr =
+    std::unique_ptr<ClientStreamingStreamBuilder<RequestType, ResponseType>>;
+
+template <class RequestType, class ResponseType>
+class UnaryStreamBuilder {
+ public:
+  virtual ~UnaryStreamBuilder() = default;
+
+  /**
+   * Create async stream entity
+   */
+  virtual AsyncStreamPtr<RequestType, ResponseType> create(
+      AsyncClient<RequestType, ResponseType>& client, RequestType request) = 0;
+};
+
+template <class RequestType, class ResponseType>
+using UnaryStreamBuilderPtr =
+    std::unique_ptr<UnaryStreamBuilder<RequestType, ResponseType>>;
+
+}  // namespace cpp2sky

example/sample.cc

@@ -27,7 +27,7 @@ TracerConfig config;
 
 void init() {
   config.set_instance_name("node_0");
-  config.set_service_name("");
+  config.set_service_name("mesh");
   config.set_address("0.0.0.0:11800");
 }
 
@@ -42,6 +42,7 @@ int main() {
     std::string context = req.get_header_value(kPropagationHeader.data());
 
     TracingContextPtr tracing_context;
+
     if (!context.empty()) {
       // 2. Create tracing context with propagated information.
       tracing_context = tracer->newContext(createSpanContext(context));

source/BUILD

@@ -19,12 +19,36 @@ cc_library(
     "@skywalking_data_collect_protocol//language-agent:tracing_protocol_cc_grpc",
     "@com_github_grpc_grpc//:grpc++",
     "//cpp2sky/internal:async_client_interface",
+    "//cpp2sky/internal:stream_builder_interface",
     "//cpp2sky:cpp2sky_interface",
     "//source/utils:util_lib",
   ],
   visibility = ["//visibility:public"],
 )
 
+# TODO(shikugawa): This is a terrible hack to disable unexpected behavior for 
+# test that has introduced GRPC_TEST.
+cc_library(
+  name = "async_client_impl_test",
+  hdrs = [
+    "grpc_async_client_impl.h",
+  ],
+  srcs = [
+    "grpc_async_client_impl.cc",
+  ],
+  deps = [
+    "@skywalking_data_collect_protocol//language-agent:tracing_protocol_cc_proto",
+    "@skywalking_data_collect_protocol//language-agent:tracing_protocol_cc_grpc",
+    "@com_github_grpc_grpc//:grpc++",
+    "//cpp2sky/internal:async_client_interface",
+    "//cpp2sky/internal:stream_builder_interface",
+    "//cpp2sky:cpp2sky_interface",
+    "//source/utils:util_lib",
+  ],
+  copts = ["-DGRPC_TEST"],
+  visibility = ["//visibility:public"],
+)
+
 cc_library(
   name = "cpp2sky_data_lib",
   hdrs  =[
@@ -42,4 +66,4 @@ cc_library(
     "//source/utils:util_lib",
   ],
   visibility = ["//visibility:public"],
-)
+)