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.

1 files changed, 23 insertions, 9 deletions

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)