Updating grpc plugin with OrphanablePtr. (#104)

* Updating grpc byte stream based on changes to grpc repo.

Author: Vizerai

opencensus/plugins/grpc/internal/client_filter.cc

@@ -73,8 +73,8 @@ void CensusClientCallData::OnDoneRecvMessageCb(void *user_data,
   // Stream messages are no longer valid after receiving trailing metadata.
   if ((*calld->recv_message_) != nullptr) {
     calld->context_.Span().AddReceivedMessageEvent(
-        calld->recv_message_count_++, (*calld->recv_message_)->length,
-        (*calld->recv_message_)->length);
+        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));
 }
@@ -117,8 +117,8 @@ void CensusClientCallData::StartTransportStreamOpBatch(
   if (op->send_message() != nullptr) {
     context_.Span().AddSentMessageEvent(
         sent_message_count_++,
-        op->op()->payload->send_message.send_message->length,
-        op->op()->payload->send_message.send_message->length);
+        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

@@ -21,6 +21,7 @@
 #include "absl/time/time.h"
 #include "opencensus/plugins/grpc/internal/channel_filter.h"
 #include "opencensus/plugins/grpc/internal/filter.h"
+#include "src/core/lib/gprpp/orphanable.h"
 
 namespace opencensus {
 
@@ -80,7 +81,7 @@ class CensusClientCallData : public grpc::CallData {
   // Server elapsed time in nanoseconds.
   uint64_t elapsed_time_;
   // The received message--may be null.
-  grpc_byte_stream **recv_message_;  // Not owned.
+  grpc_core::OrphanablePtr<grpc_core::ByteStream> *recv_message_;  // Not owned.
   // Number of messages in this RPC.
   uint32_t recv_message_count_;
   uint32_t sent_message_count_;

opencensus/plugins/grpc/internal/server_filter.cc

@@ -73,8 +73,8 @@ void CensusServerCallData::OnDoneRecvMessageCb(void *user_data,
   // Stream messages are no longer valid after receiving trailing metadata.
   if ((*calld->recv_message_) != nullptr) {
     calld->context_.Span().AddReceivedMessageEvent(
-        calld->recv_message_count_++, (*calld->recv_message_)->length,
-        (*calld->recv_message_)->length);
+        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));
 }
@@ -139,8 +139,8 @@ void CensusServerCallData::StartTransportStreamOpBatch(
   if (op->send_message() != nullptr) {
     context_.Span().AddSentMessageEvent(
         sent_message_count_++,
-        op->op()->payload->send_message.send_message->length,
-        op->op()->payload->send_message.send_message->length);
+        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

@@ -23,6 +23,7 @@
 #include "include/grpc/grpc_security.h"
 #include "opencensus/plugins/grpc/internal/channel_filter.h"
 #include "opencensus/plugins/grpc/internal/filter.h"
+#include "src/core/lib/gprpp/orphanable.h"
 
 namespace opencensus {
 
@@ -78,9 +79,10 @@ class CensusServerCallData : public grpc::CallData {
   // recv message
   grpc_closure *initial_on_done_recv_message_;
   grpc_closure on_done_recv_message_;
-  absl::Time start_time_;  // For measuring elapsed time.
+  absl::Time start_time_;
   absl::Duration elapsed_time_;
-  grpc_byte_stream **recv_message_;
+  // The received message--may be null.
+  grpc_core::OrphanablePtr<grpc_core::ByteStream> *recv_message_;  // Not owned.
   uint32_t recv_message_count_;
   uint32_t sent_message_count_;
 };