Implemented intended package streaming API flow (but currently it "streams" from memory)

4 files changed, 37 insertions, 17 deletions

diff --git a/zenhttp/httpshared.cpp b/zenhttp/httpshared.cpp
index 85687b60b..68252a763 100644
--- a/zenhttp/httpshared.cpp
+++ b/zenhttp/httpshared.cpp
@@ -81,16 +81,16 @@ FormatPackageMessage(const CbPackage& Data)
 }
 
 CbPackage
-ParsePackageMessage(IoBuffer Payload)
+ParsePackageMessage(IoBuffer Payload, std::function<IoBuffer(const IoHash&, uint64_t)> CreateBuffer)
 {
-	MemoryInStream InStream(Payload);
-	BinaryReader   Reader(InStream);
-
 	if (!Payload)
 	{
 		return {};
 	}
 
+	MemoryInStream InStream(Payload);
+	BinaryReader   Reader(InStream);
+
 	CbPackage Package;
 
 	CbPackageHeader Hdr;
@@ -110,18 +110,24 @@ ParsePackageMessage(IoBuffer Payload)
 
 	for (uint32_t i = 0; i < ChunkCount; ++i)
 	{
-		const uint64_t AttachmentSize = AttachmentEntries[i].AttachmentSize;
-		IoBuffer	   AttachmentBuffer{AttachmentSize};
+		const CbAttachmentEntry& Entry			  = AttachmentEntries[i];
+		const uint64_t			 AttachmentSize	  = Entry.AttachmentSize;
+		IoBuffer				 AttachmentBuffer = CreateBuffer(Entry.AttachmentHash, AttachmentSize);
+
+		ZEN_ASSERT(AttachmentBuffer);
+		ZEN_ASSERT(AttachmentBuffer.Size() == AttachmentSize);
+
 		Reader.Read(AttachmentBuffer.MutableData(), AttachmentSize);
+
 		CompressedBuffer CompBuf(CompressedBuffer::FromCompressed(SharedBuffer(AttachmentBuffer)));
 
 		if (i == 0)
 		{
-			Package.SetObject(LoadCompactBinaryObject(CompBuf));
+			Package.SetObject(LoadCompactBinaryObject(std::move(CompBuf)));
 		}
 		else
 		{
-			CbAttachment Attachment(CompBuf);
+			CbAttachment Attachment(std::move(CompBuf));
 			Package.AddAttachment(Attachment);
 		}
 	}
diff --git a/zenhttp/httpshared.h b/zenhttp/httpshared.h
index e7c9e4a56..06fdb104f 100644
--- a/zenhttp/httpshared.h
+++ b/zenhttp/httpshared.h
@@ -2,8 +2,11 @@
 
 #pragma once
 
+#include <zencore/iobuffer.h>
 #include <zencore/iohash.h>
 
+#include <functional>
+
 namespace zen {
 
 class IoBuffer;
@@ -33,6 +36,10 @@ static_assert(sizeof(CbAttachmentEntry) == 32);
 
 std::vector<IoBuffer> FormatPackageMessage(const CbPackage& Data);
 CompositeBuffer		  FormatPackageMessageBuffer(const CbPackage& Data);
-CbPackage			  ParsePackageMessage(IoBuffer Payload);
+CbPackage			  ParsePackageMessage(
+				IoBuffer												  Payload,
+				std::function<IoBuffer(const IoHash& Cid, uint64_t Size)> CreateBuffer = [](const IoHash&, uint64_t Size) -> IoBuffer {
+		return IoBuffer{Size};
+				});
 
 }  // namespace zen
diff --git a/zenhttp/httpsys.cpp b/zenhttp/httpsys.cpp
index 737cee509..9ee004c5c 100644
--- a/zenhttp/httpsys.cpp
+++ b/zenhttp/httpsys.cpp
@@ -1010,9 +1010,18 @@ HttpSysTransaction::InvokeRequestHandler(HttpService& Service, IoBuffer Payload)
 
 			m_PackageHandler = Service.HandlePackageRequest(ThisRequest);
 
+			// TODO: this should really be done in a streaming fashion, currently this emulates
+			// the intended flow from an API perspective
+
 			if (m_PackageHandler)
 			{
-				CbPackage Package = ParsePackageMessage(ThisRequest.ReadPayload());
+				m_PackageHandler->OnRequestBegin();
+
+				auto CreateBuffer = [&](const IoHash& Cid, uint64_t Size) -> IoBuffer { return m_PackageHandler->CreateTarget(Cid, Size); };
+
+				CbPackage Package = ParsePackageMessage(ThisRequest.ReadPayload(), CreateBuffer);
+
+				m_PackageHandler->OnRequestComplete();
 			}
 		}
 	}
@@ -1313,7 +1322,8 @@ InitialRequestHandler::HandleCompletion(ULONG IoResult, ULONG_PTR NumberOfBytesT
 		{
 			if (m_IsInitialRequest)
 			{
-				m_ContentLength = GetContentLength(HttpReq);
+				m_ContentLength					  = GetContentLength(HttpReq);
+				const HttpContentType ContentType = GetContentType(HttpReq);
 
 				if (m_ContentLength)
 				{
@@ -1321,8 +1331,6 @@ InitialRequestHandler::HandleCompletion(ULONG IoResult, ULONG_PTR NumberOfBytesT
 					// into our embedded request buffer
 
 					m_PayloadBuffer = IoBuffer(m_ContentLength);
-
-					HttpContentType ContentType = GetContentType(HttpReq);
 					m_PayloadBuffer.SetContentType(ContentType);
 
 					uint64_t	   BytesToRead		 = m_ContentLength;
diff --git a/zenserver/testing/httptest.cpp b/zenserver/testing/httptest.cpp
index b44db2722..c4fd6003c 100644
--- a/zenserver/testing/httptest.cpp
+++ b/zenserver/testing/httptest.cpp
@@ -96,12 +96,11 @@ HttpTestingService::PackageHandler::OnRequestComplete()
 {
 }
 
-zen::IoBuffer
+IoBuffer
 HttpTestingService::PackageHandler::CreateTarget(const IoHash& Cid, uint64_t StorageSize)
 {
-	ZEN_UNUSED(Cid, StorageSize);
-
-	return {};
+	ZEN_UNUSED(Cid);
+	return IoBuffer{StorageSize};
 }
 
 }  // namespace zen