Format all rpc package responses using `FormatPackageMessageBuffer` to avoid memory copy (#174)

When reading upstream, fall back to old rpc response to handle older instances.

2 files changed, 33 insertions, 35 deletions

diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index cd679d186..75fb8ac41 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -1211,12 +1211,9 @@ HttpStructuredCacheService::HandleRpcPutCacheRecords(zen::HttpServerRequest& Req
 	CbPackage RpcResponse;
 	RpcResponse.SetObject(ResponseObject.Save());
 
-	BinaryWriter MemStream;
-	RpcResponse.Save(MemStream);
+	CompositeBuffer RpcResponseBuffer = FormatPackageMessageBuffer(RpcResponse);
 
-	Request.WriteResponse(HttpResponseCode::OK,
-						  HttpContentType::kCbPackage,
-						  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
+	Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, RpcResponseBuffer);
 }
 
 HttpStructuredCacheService::PutResult
@@ -1635,12 +1632,9 @@ HttpStructuredCacheService::HandleRpcGetCacheRecords(zen::HttpServerRequest& Htt
 	ResponseObject.EndArray();
 	ResponsePackage.SetObject(ResponseObject.Save());
 
-	BinaryWriter MemStream;
-	ResponsePackage.Save(MemStream);
+	CompositeBuffer RpcResponseBuffer = FormatPackageMessageBuffer(ResponsePackage);
 
-	HttpRequest.WriteResponse(HttpResponseCode::OK,
-							  HttpContentType::kCbPackage,
-							  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
+	HttpRequest.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, RpcResponseBuffer);
 }
 
 void
@@ -1748,12 +1742,9 @@ HttpStructuredCacheService::HandleRpcPutCacheValues(zen::HttpServerRequest& Requ
 	CbPackage RpcResponse;
 	RpcResponse.SetObject(ResponseObject.Save());
 
-	BinaryWriter MemStream;
-	RpcResponse.Save(MemStream);
+	CompositeBuffer RpcResponseBuffer = FormatPackageMessageBuffer(RpcResponse);
 
-	Request.WriteResponse(HttpResponseCode::OK,
-						  HttpContentType::kCbPackage,
-						  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
+	Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, RpcResponseBuffer);
 }
 
 void
@@ -1943,13 +1934,9 @@ HttpStructuredCacheService::HandleRpcGetCacheValues(zen::HttpServerRequest& Http
 	ResponseObject.EndArray();
 
 	RpcResponse.SetObject(ResponseObject.Save());
+	CompositeBuffer RpcResponseBuffer = FormatPackageMessageBuffer(RpcResponse);
 
-	BinaryWriter MemStream;
-	RpcResponse.Save(MemStream);
-
-	HttpRequest.WriteResponse(HttpResponseCode::OK,
-							  HttpContentType::kCbPackage,
-							  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
+	HttpRequest.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, RpcResponseBuffer);
 }
 
 namespace cache::detail {
@@ -2451,12 +2438,9 @@ HttpStructuredCacheService::WriteGetCacheChunksResponse(std::string_view
 
 	RpcResponse.SetObject(Writer.Save());
 
-	BinaryWriter MemStream;
-	RpcResponse.Save(MemStream);
+	CompositeBuffer RpcResponseBuffer = FormatPackageMessageBuffer(RpcResponse);
 
-	HttpRequest.WriteResponse(HttpResponseCode::OK,
-							  HttpContentType::kCbPackage,
-							  IoBuffer(IoBuffer::Wrap, MemStream.GetData(), MemStream.GetSize()));
+	HttpRequest.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, RpcResponseBuffer);
 }
 
 void
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index c1c0395e7..610946385 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -15,6 +15,8 @@
 #include <zencore/timer.h>
 #include <zencore/trace.h>
 
+#include <zenhttp/httpshared.h>
+
 #include <zenstore/cidstore.h>
 
 #include <auth/authmgr.h>
@@ -887,7 +889,6 @@ namespace detail {
 			}
 			BatchRequest.EndObject();
 
-			CbPackage	   BatchResponse;
 			ZenCacheResult Result;
 
 			{
@@ -899,12 +900,13 @@ namespace detail {
 
 			if (Result.Success)
 			{
-				if (BatchResponse.TryLoad(Result.Response))
+				CbPackage BatchResponse;
+				if (ParsePackageMessageWithLegacyFallback(Result.Response, BatchResponse))
 				{
 					CbArrayView Results = BatchResponse.GetObject()["Result"sv].AsArrayView();
 					if (Results.Num() != Requests.size())
 					{
-						ZEN_WARN("Upstream::Zen::GetCacheRecords invalid number of Requests from Upstream.");
+						ZEN_WARN("Upstream::Zen::GetCacheRecords invalid number of Response results from Upstream.");
 					}
 					else
 					{
@@ -921,6 +923,10 @@ namespace detail {
 						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
 					}
 				}
+				else
+				{
+					ZEN_WARN("Upstream::Zen::GetCacheRecords invalid Response from Upstream.");
+				}
 			}
 
 			for (CacheKeyRequest* Request : Requests)
@@ -1004,7 +1010,6 @@ namespace detail {
 			}
 			BatchRequest.EndObject();
 
-			CbPackage	   BatchResponse;
 			ZenCacheResult Result;
 
 			{
@@ -1016,12 +1021,13 @@ namespace detail {
 
 			if (Result.Success)
 			{
-				if (BatchResponse.TryLoad(Result.Response))
+				CbPackage BatchResponse;
+				if (ParsePackageMessageWithLegacyFallback(Result.Response, BatchResponse))
 				{
 					CbArrayView Results = BatchResponse.GetObject()["Result"sv].AsArrayView();
 					if (CacheValueRequests.size() != Results.Num())
 					{
-						ZEN_WARN("Upstream::Zen::GetCacheValues invalid number of Requests from Upstream.");
+						ZEN_WARN("Upstream::Zen::GetCacheValues invalid number of Response results from Upstream.");
 					}
 					else
 					{
@@ -1068,6 +1074,10 @@ namespace detail {
 						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
 					}
 				}
+				else
+				{
+					ZEN_WARN("Upstream::Zen::GetCacheValues invalid Response from Upstream.");
+				}
 			}
 
 			for (CacheValueRequest* RequestPtr : CacheValueRequests)
@@ -1135,7 +1145,6 @@ namespace detail {
 			}
 			BatchRequest.EndObject();
 
-			CbPackage	   BatchResponse;
 			ZenCacheResult Result;
 
 			{
@@ -1147,12 +1156,13 @@ namespace detail {
 
 			if (Result.Success)
 			{
-				if (BatchResponse.TryLoad(Result.Response))
+				CbPackage BatchResponse;
+				if (ParsePackageMessageWithLegacyFallback(Result.Response, BatchResponse))
 				{
 					CbArrayView Results = BatchResponse.GetObject()["Result"sv].AsArrayView();
 					if (CacheChunkRequests.size() != Results.Num())
 					{
-						ZEN_WARN("Upstream::Zen::GetCacheChunks invalid number of Requests from Upstream.");
+						ZEN_WARN("Upstream::Zen::GetCacheChunks invalid number of Response results from Upstream.");
 					}
 					else
 					{
@@ -1199,6 +1209,10 @@ namespace detail {
 						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
 					}
 				}
+				else
+				{
+					ZEN_WARN("Upstream::Zen::GetCacheChunks invalid Response from Upstream.");
+				}
 			}
 
 			for (CacheChunkRequest* RequestPtr : CacheChunkRequests)