Adding in MessageEvents for client and server. (#87)

Adding in MessageEvents for client and server.

Author: Vizerai

opencensus/plugins/grpc/internal/client_filter.cc

@@ -52,9 +52,6 @@ void CensusClientCallData::OnDoneRecvTrailingMetadataCb(void *user_data,
   CensusClientCallData *calld =
       reinterpret_cast<CensusClientCallData *>(elem->call_data);
   GPR_ASSERT(calld != nullptr);
-  if ((*calld->recv_message_) != nullptr) {
-    ++calld->recv_message_count_;
-  }
   if (error == GRPC_ERROR_NONE) {
     GPR_ASSERT(calld->recv_trailing_metadata_ != nullptr);
     FilterTrailingMetadata(calld->recv_trailing_metadata_,
@@ -75,7 +72,9 @@ void CensusClientCallData::OnDoneRecvMessageCb(void *user_data,
   GPR_ASSERT(channeld != nullptr);
   // Stream messages are no longer valid after receiving trailing metadata.
   if ((*calld->recv_message_) != nullptr) {
-    calld->recv_message_count_++;
+    calld->context_.Span().AddReceivedMessageEvent(
+        calld->recv_message_count_++, (*calld->recv_message_)->length,
+        (*calld->recv_message_)->length);
   }
   GRPC_CLOSURE_RUN(calld->initial_on_done_recv_message_, GRPC_ERROR_REF(error));
 }
@@ -116,7 +115,10 @@ void CensusClientCallData::StartTransportStreamOpBatch(
   }
 
   if (op->send_message() != nullptr) {
-    ++sent_message_count_;
+    context_.Span().AddSentMessageEvent(
+        sent_message_count_++,
+        op->op()->payload->send_message.send_message->length,
+        op->op()->payload->send_message.send_message->length);
   }
   if (op->recv_message() != nullptr) {
     recv_message_ = op->op()->payload->recv_message.recv_message;

opencensus/plugins/grpc/internal/client_filter.h

@@ -80,8 +80,8 @@ class CensusClientCallData : public grpc::CallData {
   // The received message--may be null.
   grpc_byte_stream **recv_message_;  // Not owned.
   // Number of messages in this RPC.
-  uint64_t recv_message_count_;
-  uint64_t sent_message_count_;
+  uint32_t recv_message_count_;
+  uint32_t sent_message_count_;
 };
 
 }  // namespace opencensus

opencensus/plugins/grpc/internal/server_filter.cc

@@ -72,7 +72,9 @@ void CensusServerCallData::OnDoneRecvMessageCb(void *user_data,
   GPR_ASSERT(channeld != nullptr);
   // Stream messages are no longer valid after receiving trailing metadata.
   if ((*calld->recv_message_) != nullptr) {
-    ++calld->recv_message_count_;
+    calld->context_.Span().AddReceivedMessageEvent(
+        calld->recv_message_count_++, (*calld->recv_message_)->length,
+        (*calld->recv_message_)->length);
   }
   GRPC_CLOSURE_RUN(calld->initial_on_done_recv_message_, GRPC_ERROR_REF(error));
 }
@@ -141,7 +143,10 @@ void CensusServerCallData::StartTransportStreamOpBatch(
     op->set_recv_initial_metadata_ready(&on_done_recv_initial_metadata_);
   }
   if (op->send_message() != nullptr) {
-    ++sent_message_count_;
+    context_.Span().AddSentMessageEvent(
+        sent_message_count_++,
+        op->op()->payload->send_message.send_message->length,
+        op->op()->payload->send_message.send_message->length);
   }
   if (op->recv_message() != nullptr) {
     recv_message_ = op->op()->payload->recv_message.recv_message;

opencensus/plugins/grpc/internal/server_filter.h

@@ -79,8 +79,8 @@ class CensusServerCallData : public grpc::CallData {
   absl::Time start_time_;  // For measuring elapsed time.
   absl::Duration elapsed_time_;
   grpc_byte_stream **recv_message_;
-  uint64_t recv_message_count_;
-  uint64_t sent_message_count_;
+  uint32_t recv_message_count_;
+  uint32_t sent_message_count_;
 };
 
 }  // namespace opencensus