Change Value propagation to Zen or Jupiter

This change ensures we retain the right content type of kCompressedBinary when propagating values from Zen->UpstreamZen.  This is done via an RPC that posts a CbPackage.  Furthermore when propagating from Zen->Jupiter, it composes its own referencing CbObject for them instead of sending a octet content type and Jupiter defining the referencing CbObject.  When fetching Values from Jupiter, this new composed CbObject is still interpreted correctly by Zen.

1 files changed, 30 insertions, 0 deletions

diff --git a/zenserver/upstream/zen.cpp b/zenserver/upstream/zen.cpp
index 0570dd316..098f53bb0 100644
--- a/zenserver/upstream/zen.cpp
+++ b/zenserver/upstream/zen.cpp
@@ -3,10 +3,12 @@
 #include "zen.h"
 
 #include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinarypackage.h>
 #include <zencore/compactbinaryvalidation.h>
 #include <zencore/fmtutils.h>
 #include <zencore/session.h>
 #include <zencore/stream.h>
+#include <zenhttp/httpshared.h>
 
 #include "cache/structuredcachestore.h"
 #include "diag/formatters.h"
@@ -539,4 +541,32 @@ ZenStructuredCacheSession::InvokeRpc(const CbObjectView& Request)
 	return {.Response = std::move(Buffer), .Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
 }
 
+ZenCacheResult
+ZenStructuredCacheSession::InvokeRpc(const CbPackage& Request)
+{
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_Client.ServiceUrl() << "/z$/$rpc";
+
+	SharedBuffer Message = FormatPackageMessageBuffer(Request).Flatten();
+
+	cpr::Session& Session = m_SessionState->GetSession();
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}});
+	Session.SetBody(cpr::Body{reinterpret_cast<const char*>(Message.GetData()), Message.GetSize()});
+
+	cpr::Response Response = Session.Post();
+	ZEN_DEBUG("POST {}", Response);
+
+	if (Response.error)
+	{
+		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = std::move(Response.error.message)};
+	}
+
+	const bool	   Success = Response.status_code == 200;
+	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
+
+	return {.Response = std::move(Buffer), .Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
+}
+
 }  // namespace zen