From 5946df191d96bb10d57f56d4ddd7932274bc13ab Mon Sep 17 00:00:00 2001
From: Gavin Halliday <gavin.halliday@lexisnexis.com>
Date: Tue, 25 Jun 2024 13:15:15 +0100
Subject: [PATCH 1/2] HPCC-32136 Allow the input and output to be replaced in a
 buffered class

Signed-off-by: Gavin Halliday <gavin.halliday@lexisnexis.com>
---
 system/jlib/jstream.cpp            | 16 ++++++++++++++++
 system/jlib/jstream.hpp            |  6 +++++-
 testing/unittests/jstreamtests.cpp |  1 +
 3 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/system/jlib/jstream.cpp b/system/jlib/jstream.cpp
index fa76a9632f8..64641763a3a 100644
--- a/system/jlib/jstream.cpp
+++ b/system/jlib/jstream.cpp
@@ -405,6 +405,11 @@ class CBlockedSerialInputStream : public CInterfaceOf<IBufferedSerialInputStream
         input->reset(_offset, _flen);
     }
 
+    virtual void replaceInput(ISerialInputStream * newInput) override
+    {
+        input.set(newInput);
+    }
+
 protected:
     inline byte * data(size32_t offset) { return (byte *)buffer.get() + offset; }
     inline size32_t available() const { return dataLength - bufferOffset; }
@@ -792,6 +797,11 @@ class CBlockedSerialOutputStream final : public CInterfaceOf<IBufferedSerialOutp
 
     virtual offset_t tell() const override { return blockOffset+bufferOffset; }
 
+    virtual void replaceOutput(ISerialOutputStream * newOutput) override
+    {
+        output.set(newOutput);
+    }
+
 //-------------------------------------------------------
 //Helper functions for CThreadedBlockedSerialOutputStream
 //doCommit() and doResume are also used by this class
@@ -1007,6 +1017,12 @@ class CThreadedBlockedSerialOutputStream final : public CInterfaceOf<IBufferedSe
         checkForPendingWrite();
     }
 
+    virtual void replaceOutput(ISerialOutputStream * newOutput) override
+    {
+        stream[0].replaceOutput(newOutput);
+        stream[1].replaceOutput(newOutput);
+    }
+
 protected:
     virtual offset_t tell() const override
     {
diff --git a/system/jlib/jstream.hpp b/system/jlib/jstream.hpp
index 03751d8e29a..2245859f36d 100644
--- a/system/jlib/jstream.hpp
+++ b/system/jlib/jstream.hpp
@@ -63,7 +63,10 @@ interface ISerialStream : extends ISerialInputStream
                                                                     // if got<wanted then approaching eof
                                                                     // if got>wanted then got is size available in buffer
 };
-using IBufferedSerialInputStream = ISerialStream;
+interface IBufferedSerialInputStream : extends ISerialStream
+{
+    virtual void replaceInput(ISerialInputStream * newInput) = 0;
+};
 
 /* example of reading a nul terminated string using ISerialStream peek and skip
 {
@@ -100,6 +103,7 @@ interface IBufferedSerialOutputStream : extends ISerialOutputStream
     virtual void commit(size32_t written) = 0 ;      // commit the data written to the block returned by reserve
     virtual void suspend(size32_t wanted) = 0;   // Reserve some bytes and prevent data being flushed to the next stage until endNested is called.  May nest.
     virtual void resume(size32_t len, const void * ptr) = 0;  // update the data allocated by suspend and allow flushing.
+    virtual void replaceOutput(ISerialOutputStream * newOutput) = 0;
 };
 
 interface ICompressor;
diff --git a/testing/unittests/jstreamtests.cpp b/testing/unittests/jstreamtests.cpp
index 98303578ccc..7e13a495e04 100644
--- a/testing/unittests/jstreamtests.cpp
+++ b/testing/unittests/jstreamtests.cpp
@@ -292,6 +292,7 @@ class NullOuputStream : public CInterfaceOf<IBufferedSerialOutputStream>
     virtual void suspend(size32_t wanted) {}
     virtual void resume(size32_t len, const void * ptr) {}
     virtual offset_t tell() const override { return 0; }
+    virtual void replaceOutput(ISerialOutputStream * newOutput) override {}
 };
 
 class JlibStreamStressTest : public CppUnit::TestFixture

From 6835477778a17332da0284334271c3b793feead7 Mon Sep 17 00:00:00 2001
From: Gavin Halliday <gavin.halliday@lexisnexis.com>
Date: Tue, 25 Jun 2024 16:17:22 +0100
Subject: [PATCH 2/2] HPCC-32136 Minor modifications to streaming classes

Signed-off-by: Gavin Halliday <gavin.halliday@lexisnexis.com>
---
 system/jlib/jstream.cpp | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/system/jlib/jstream.cpp b/system/jlib/jstream.cpp
index 64641763a3a..8f31718a8e1 100644
--- a/system/jlib/jstream.cpp
+++ b/system/jlib/jstream.cpp
@@ -290,7 +290,7 @@ IByteInputStream *createInputStream(int handle)
 // This means the buffer size is likely to be bigger than the block size - the class is passed
 // an initial estimate for the potential overlap.
 
-class CBlockedSerialInputStream : public CInterfaceOf<IBufferedSerialInputStream>
+class CBlockedSerialInputStream final : public CInterfaceOf<IBufferedSerialInputStream>
 {
 public:
     CBlockedSerialInputStream(ISerialInputStream * _input, size32_t _blockReadSize)
@@ -318,7 +318,7 @@ class CBlockedSerialInputStream : public CInterfaceOf<IBufferedSerialInputStream
         }
 
         //While there are blocks larger than the buffer size read directly into the target buffer
-        while (sizeRead + blockReadSize <= len)
+        while (unlikely(sizeRead + blockReadSize <= len))
         {
             size32_t got = readNextBlock(blockReadSize, target+sizeRead);
             if ((got == 0) || (got == BufferTooSmall))
@@ -327,7 +327,7 @@ class CBlockedSerialInputStream : public CInterfaceOf<IBufferedSerialInputStream
             nextBlockOffset += got;
         }
 
-        while ((sizeRead < len) && !endOfStream)
+        while (likely((sizeRead < len) && !endOfStream))
         {
             assertex(bufferOffset == dataLength);
             // NOTE: This could read less than a block, even if a whole block was requested.
@@ -368,7 +368,7 @@ class CBlockedSerialInputStream : public CInterfaceOf<IBufferedSerialInputStream
     virtual void get(size32_t len, void * ptr) override
     {
         size32_t numRead = read(len, ptr);
-        if (numRead != len)
+        if (unlikely(numRead != len))
             throw makeStringExceptionV(-1, "End of input stream for read of %u bytes at offset %llu", len, tell()-numRead);
     }