BugFix: Fix some problems at stream invoking (#60)

- fix stream crash at ServerContext because it has been moved
- fix stream transinfo not pass into ServerContext
- delete deprecated interface and useless code in trpc/client/make_client_context.h/.cc

Author: weimch

trpc/client/make_client_context.cc

@@ -21,10 +21,6 @@
 #include "trpc/util/hash_util.h"
 #include "trpc/util/log/logging.h"
 #include "trpc/util/time.h"
-// #ifdef TRPC_BUILD_INCLUDE_RPCZ
-// #include "trpc/rpcz/filter/rpcz_filter_index.h"
-// #include "trpc/rpcz/span.h"
-// #endif
 
 namespace trpc {
 
@@ -75,39 +71,6 @@ void RegisterMakeClientContextCallback(MakeClientContextCallback&& callback) {
   callbacks.emplace_back(std::move(callback));
 }
 
-ClientContextPtr MakeClientContext(const ServerContextPtr& ctx) {
-  ClientContextPtr client_ctx = MakeRefCounted<ClientContext>();
-
-  const auto& trans_info = ctx->GetPbReqTransInfo();
-  if (trans_info.size() > 0) {
-    client_ctx->SetReqTransInfo(trans_info.begin(), trans_info.end());
-  }
-
-  client_ctx->SetMessageType(ctx->GetMessageType());
-  client_ctx->SetCallerName(ctx->GetCalleeName());
-  client_ctx->SetCallerFuncName(ctx->GetFuncName());
-
-  RunMakeClientContextCallbacks(ctx, client_ctx);
-
-  // Calculate the remaining timeout and set it to the client context.
-  int64_t nowms = static_cast<int64_t>(trpc::time::GetMilliSeconds());
-  int64_t cost_time = nowms - ctx->GetRecvTimestamp();
-  int64_t left_time = static_cast<int64_t>(ctx->GetTimeout()) - cost_time;
-  if (left_time < 0) {
-    left_time = 0;
-  }
-
-  client_ctx->SetTimeout(left_time);
-
-  if (ctx->IsUseFullLinkTimeout()) {
-    client_ctx->SetFullLinkTimeout(left_time);
-  } else {
-    client_ctx->SetTimeout(left_time);
-  }
-
-  return client_ctx;
-}
-
 ClientContextPtr MakeClientContext(const ServiceProxyPtr& proxy) {
   TRPC_ASSERT(proxy && "proxy must be created before create a ClientContext obj");
   return MakeRefCounted<ClientContext>(proxy->GetClientCodec());

trpc/client/make_client_context.h

@@ -68,12 +68,4 @@ void BackFillServerTransInfo(const ClientContextPtr& client_context, const Serve
 using MakeClientContextCallback = std::function<void(const ServerContextPtr&, ClientContextPtr&)>;
 void RegisterMakeClientContextCallback(MakeClientContextCallback&& callback);
 
-/// @brief Create client context based on server context
-/// @param ctx server context
-/// @return client context
-/// @deprecated Use MakeClientContext(const ServerContextPtr& ctx, const ServiceProxyPtr& proxy) instead.
-/// @private
-[[deprecated("use the interface with params as 'ctx' and 'proxy' ")]]
-ClientContextPtr MakeClientContext(const ServerContextPtr& ctx);
-
 }  // namespace trpc

trpc/client/service_proxy.cc

@@ -692,6 +692,10 @@ stream::StreamReaderWriterProviderPtr ServiceProxy::SelectStreamProvider(const C
   TRPC_ASSERT(codec_->Name() == "trpc" || codec_->Name() == "http");
   TRPC_ASSERT(thread_model_ != nullptr);
 
+  if (context->GetResponse() == nullptr) {
+    context->SetResponse(codec_->CreateResponsePtr());
+  }
+
   FillClientContext(context);
 
   // Get one address of peer server

trpc/stream/grpc/grpc_server_stream_handler.cc

@@ -269,7 +269,7 @@ StreamReaderWriterProviderPtr FiberGrpcServerStreamHandler::CreateStream(StreamO
       TRPC_LOG_ERROR("stream " << stream_id << " already exists.");
       return nullptr;
     }
-    ServerContextPtr& context = std::any_cast<ServerContextPtr&>(options.context.context);
+    ServerContextPtr context = std::any_cast<ServerContextPtr>(options.context.context);
     auto stream = MakeRefCounted<GrpcServerStream>(std::move(options), session_.get(), &session_mutex_, &session_cv_);
     // Set the stream for the context so that it can be obtained in stream_rpc_method_handler.
     context->SetStreamReaderWriterProvider(stream);

trpc/stream/http/async/server/stream_handler.cc

@@ -29,7 +29,7 @@ StreamReaderWriterProviderPtr HttpServerAsyncStreamHandler::CreateStream(StreamO
     return nullptr;
   }
 
-  ServerContextPtr& context = std::any_cast<ServerContextPtr&>(options.context.context);
+  ServerContextPtr context = std::any_cast<ServerContextPtr>(options.context.context);
   options.connection_id = GetMutableStreamOptions()->connection_id;
   stream_ = MakeRefCounted<HttpServerAsyncStream>(std::move(options));
   context->SetStreamReaderWriterProvider(stream_);

trpc/stream/trpc/trpc_server_stream.cc

@@ -118,6 +118,8 @@ RetCode TrpcServerStream::HandleInit(StreamRecvMessage&& msg) {
   context->SetCallerName(request_metadata.caller());
   context->SetFuncName(request_metadata.func());
   context->SetReqEncodeType(init_frame.stream_init_metadata.content_type());
+  context->GetMutablePbReqTransInfo()->insert(request_metadata.trans_info().begin(),
+                                              request_metadata.trans_info().end());
 
   std::string var_path = "trpc/stream_rpc/server" + request_metadata.func();
   auto stream_var = StreamVarHelper::GetInstance()->GetOrCreateStreamVar(var_path);
@@ -272,6 +274,11 @@ RetCode TrpcServerStream::SendClose(StreamSendMessage&& msg) {
     return RetCode::kError;
   }
 
+  const auto& context = std::any_cast<const ServerContextPtr&>(GetMutableStreamOptions()->context.context);
+
+  auto& close_meta = std::any_cast<TrpcStreamCloseMeta&>(msg.metadata);
+  close_meta.mutable_trans_info()->insert(context->GetPbRspTransInfo().begin(), context->GetPbRspTransInfo().end());
+
   RetCode ret = TrpcStream::SendClose(std::move(msg));
   if (TRPC_UNLIKELY(ret != RetCode::kSuccess)) {
     return ret;
@@ -281,7 +288,6 @@ RetCode TrpcServerStream::SendClose(StreamSendMessage&& msg) {
 
   Stop();
 
-  const auto& context = std::any_cast<const ServerContextPtr&>(GetMutableStreamOptions()->context.context);
   RunMessageFilter(FilterPoint::SERVER_PRE_SEND_MSG, context);
 
   TRPC_FMT_DEBUG("server stream, send close end, stream id: {}", GetId());

trpc/stream/trpc/trpc_server_stream_handler.cc

@@ -35,7 +35,7 @@ StreamReaderWriterProviderPtr TrpcServerStreamHandler::CreateStream(StreamOption
   // Since IsNewStream has previously determined that the stream does not exist, it is likely that it still does not
   // exist, so create the stream first.
   options.connection_id = options_.connection_id;
-  ServerContextPtr& context = std::any_cast<ServerContextPtr&>(options.context.context);
+  ServerContextPtr context = std::any_cast<ServerContextPtr>(options.context.context);
   auto stream = MakeRefCounted<TrpcServerStream>(std::move(options));
   context->SetStreamReaderWriterProvider(stream);