From a3c44e1416022b5d3ad917a434035ef84e2db26c Mon Sep 17 00:00:00 2001 From: shunkica Date: Sun, 9 Aug 2026 12:12:47 +0000 Subject: [PATCH] Bound heap usage when digesting signed attachments; #380 WSS4J's AttachmentContentSignatureTransform#processAttachment digests each signed attachment (cid: reference) by calling mark(Integer.MAX_VALUE) on the attachment source stream, reading it to the end and calling reset(), so the attachment stays readable afterwards. The file backed streams handed out so far (NonBlockingBufferedInputStream) honour that mark by buffering everything read after it on the heap, so the peak heap scales with the attachment size (a 250 MB attachment peaks at ~390 MB) although the content is already spilled to a temporary file. This affects the signing and the verification side alike. Add com.helger.phase4.util.MarkableFileInputStream, which implements mark/reset by re-positioning the underlying FileChannel, and hand it out at all the file backed provider sites: - WSS4JAttachment.createOutgoingFileAttachment (File and byte[] overloads) - WSS4JAttachment.createIncomingFileAttachment (both overloads) - SoapHeaderElementProcessorWSS4J (decrypted attachment temporary file) - AS4IncomingHandler._createReadMultipleISP Digesting then runs in constant heap - a 250 MB attachment completes at -Xmx64m, where the previous stream throws an OutOfMemoryError - and the 2 GB limit of heap buffering mark/reset (see the WSS4J comment) no longer applies to file backed attachments; a 2.5 GB attachment was verified at -Xmx64m. The digest input is unchanged, only the carrier of the same bytes is swapped. In-memory attachments (<= 64 KB) keep their byte array streams, which already support constant heap mark/reset. Not covered: incoming attachments that are encrypted as well as signed. There WSS4J digests its own CipherInputStream, which does not support mark/reset, so AttachmentContentSignatureTransform wraps it in a BufferedInputStream itself and buffers the plaintext on the heap. That needs a separate change on the decryption path. --- .../phase4/attachment/WSS4JAttachment.java | 11 +- .../phase4/incoming/AS4IncomingHandler.java | 3 +- .../soap/SoapHeaderElementProcessorWSS4J.java | 3 +- .../phase4/util/MarkableFileInputStream.java | 242 ++++++++++++++++++ .../attachment/WSS4JAttachmentTest.java | 56 ++++ .../util/MarkableFileInputStreamTest.java | 231 +++++++++++++++++ 6 files changed, 539 insertions(+), 7 deletions(-) create mode 100644 phase4-lib/src/main/java/com/helger/phase4/util/MarkableFileInputStream.java create mode 100644 phase4-lib/src/test/java/com/helger/phase4/util/MarkableFileInputStreamTest.java diff --git a/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java b/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java index f0f5bda1e..313613872 100644 --- a/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java +++ b/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java @@ -54,6 +54,7 @@ import com.helger.phase4.logging.Phase4LoggerFactory; import com.helger.phase4.model.message.MessageHelperMethods; import com.helger.phase4.util.AS4ResourceHelper; +import com.helger.phase4.util.MarkableFileInputStream; import jakarta.activation.DataHandler; import jakarta.activation.DataSource; @@ -462,8 +463,8 @@ public static WSS4JAttachment createOutgoingFileAttachment (@NonNull final File } // Set a stream provider that can be read multiple times (opens a new - // FileInputStream internally) - final IHasInputStream aISP = HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aRealFile)); + // stream internally) + final IHasInputStream aISP = HasInputStream.multiple (() -> MarkableFileInputStream.create (aRealFile)); ret.setSourceStreamProvider (aISP); if (eCompressionMode != null) { @@ -531,7 +532,7 @@ public static WSS4JAttachment createOutgoingFileAttachment (final byte @NonNull aOS.write (aSrcData); } } - final IHasInputStream aISP = HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aRealFile)); + final IHasInputStream aISP = HasInputStream.multiple (() -> MarkableFileInputStream.create (aRealFile)); ret.setSourceStreamProvider (aISP); // Preserve the compressed data for non-repudiation purposes - the // signature digests are calculated over the compressed data @@ -629,7 +630,7 @@ public static WSS4JAttachment createIncomingFileAttachment (@NonNull final AS4In aOS.write (nProbedByte); StreamHelper.copyInputStreamToOutputStream (aDecodedIS, aOS); } - ret.setSourceStreamProvider (HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aTempFile))); + ret.setSourceStreamProvider (HasInputStream.multiple (() -> MarkableFileInputStream.create (aTempFile))); } // Read all MIME part headers @@ -739,7 +740,7 @@ public static WSS4JAttachment createIncomingFileAttachment (@NonNull final MimeB { aBodyPart.getDataHandler ().writeTo (aOS); } - ret.setSourceStreamProvider (HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aTempFile))); + ret.setSourceStreamProvider (HasInputStream.multiple (() -> MarkableFileInputStream.create (aTempFile))); } // Read all MIME part headers diff --git a/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java b/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java index 22e9cc591..28298ef9b 100644 --- a/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java +++ b/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java @@ -98,6 +98,7 @@ import com.helger.phase4.profile.IAS4ProfileValidator.EAS4ProfileValidationMode; import com.helger.phase4.util.AS4ResourceHelper; import com.helger.phase4.util.AS4XMLHelper; +import com.helger.phase4.util.MarkableFileInputStream; import com.helger.phase4.util.Phase4Exception; import com.helger.phase4.util.Phase4IncomingException; import com.helger.web.multipart.MultipartProgressNotifier; @@ -651,7 +652,7 @@ private static IHasInputStream _createReadMultipleISP (@NonNull final AS4Resourc } aTempFileWrapper.set (aTempFile); } - return FileHelper.getBufferedInputStream (aTempFile); + return MarkableFileInputStream.create (aTempFile); } catch (final IOException ex) { diff --git a/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java b/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java index a6a8bc9b8..70753a123 100644 --- a/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java +++ b/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java @@ -70,6 +70,7 @@ import com.helger.phase4.model.error.EEbmsError; import com.helger.phase4.model.pmode.IPMode; import com.helger.phase4.model.pmode.leg.PModeLeg; +import com.helger.phase4.util.MarkableFileInputStream; import com.helger.phase4.wss.WSSConfigManager; import com.helger.phase4.wss.WSSSynchronizer; import com.helger.xml.XMLHelper; @@ -356,7 +357,7 @@ private ESuccess _verifyAndDecrypt (@NonNull final Document aSOAPDoc, { LOGGER.error ("Failed to write response attachment to temporary file '" + aTempFile.getAbsolutePath () + "'"); } - aResponseAttachment.setSourceStreamProvider (HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aTempFile))); + aResponseAttachment.setSourceStreamProvider (HasInputStream.multiple (() -> MarkableFileInputStream.create (aTempFile))); } // Remember in State diff --git a/phase4-lib/src/main/java/com/helger/phase4/util/MarkableFileInputStream.java b/phase4-lib/src/main/java/com/helger/phase4/util/MarkableFileInputStream.java new file mode 100644 index 000000000..94c1538c9 --- /dev/null +++ b/phase4-lib/src/main/java/com/helger/phase4/util/MarkableFileInputStream.java @@ -0,0 +1,242 @@ +/* + * Copyright (C) 2015-2026 Philip Helger (www.helger.com) + * philip[at]helger[dot]com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.helger.phase4.util; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.StandardOpenOption; +import java.util.Objects; + +import org.jspecify.annotations.NonNull; +import org.jspecify.annotations.Nullable; +import org.slf4j.Logger; + +import com.helger.annotation.concurrent.NotThreadSafe; +import com.helger.base.enforce.ValueEnforcer; +import com.helger.phase4.logging.Phase4LoggerFactory; + +/** + * A buffered {@link InputStream} on a {@link File} that supports {@link #mark(int)} and + * {@link #reset()} with constant heap usage, by re-positioning the underlying {@link FileChannel} + * instead of buffering all the bytes read after the mark.
+ * This is relevant for the streams handed over to WSS4J: for every signed attachment, + * AttachmentContentSignatureTransform#processAttachment calls + * mark (Integer.MAX_VALUE) on the source stream, reads it to the end to calculate the + * digest and calls reset () afterwards, so that the attachment stays readable. On a + * heap buffering stream (like BufferedInputStream or + * NonBlockingBufferedInputStream) that mark/read/reset sequence keeps the complete + * attachment on the heap, so the heap usage scales with the attachment size. With this class it + * stays constant, and the 2 GB limit of heap buffering streams does not apply. See issue #380.
+ * The mark position is initially 0, so a {@link #reset()} without a preceding {@link #mark(int)} + * re-reads from the beginning (like {@link java.io.ByteArrayInputStream}). + * + * @since 4.6.0 + */ +@NotThreadSafe +public class MarkableFileInputStream extends InputStream +{ + /** The default size of the internal read buffer in bytes */ + public static final int DEFAULT_BUFFER_SIZE = 16 * 1024; + + private static final Logger LOGGER = Phase4LoggerFactory.getLogger (MarkableFileInputStream.class); + + private final FileChannel m_aChannel; + // Read buffer; between reads always in "drain" mode: position..limit are the + // unread bytes + private final ByteBuffer m_aBuffer; + // File offset of the next byte to be read from the channel (mirrors + // FileChannel.position) + private long m_nChannelPos = 0; + // File offset to fall back to on reset + private long m_nMarkPos = 0; + + /** + * Constructor using {@link #DEFAULT_BUFFER_SIZE}. + * + * @param aFile + * The file to read. May not be null. + * @throws IOException + * If the file cannot be opened for reading. + */ + public MarkableFileInputStream (@NonNull final File aFile) throws IOException + { + this (aFile, DEFAULT_BUFFER_SIZE); + } + + /** + * Constructor. + * + * @param aFile + * The file to read. May not be null. + * @param nBufferSize + * The size of the internal read buffer in bytes. Must be > 0. + * @throws IOException + * If the file cannot be opened for reading. + */ + public MarkableFileInputStream (@NonNull final File aFile, final int nBufferSize) throws IOException + { + ValueEnforcer.notNull (aFile, "File"); + ValueEnforcer.isGT0 (nBufferSize, "BufferSize"); + m_aChannel = FileChannel.open (aFile.toPath (), StandardOpenOption.READ); + m_aBuffer = ByteBuffer.allocate (nBufferSize); + // The buffer starts out empty + m_aBuffer.limit (0); + } + + /** + * @return The offset of the next byte to be delivered by this stream. + */ + private long _getLogicalPos () + { + return m_nChannelPos - m_aBuffer.remaining (); + } + + /** + * Fill the internal buffer from the channel. + * + * @return false if EOF was reached. + */ + private boolean _fill () throws IOException + { + m_aBuffer.clear (); + final int nRead = m_aChannel.read (m_aBuffer); + m_aBuffer.flip (); + if (nRead <= 0) + return false; + m_nChannelPos += nRead; + return true; + } + + /** + * Re-position the channel and discard all buffered bytes. + */ + private void _seek (final long nPos) throws IOException + { + m_aChannel.position (nPos); + m_nChannelPos = nPos; + m_aBuffer.position (0).limit (0); + } + + @Override + public int read () throws IOException + { + if (!m_aBuffer.hasRemaining () && !_fill ()) + return -1; + return m_aBuffer.get () & 0xff; + } + + @Override + public int read (final byte @NonNull [] aBuf, final int nOfs, final int nLen) throws IOException + { + Objects.checkFromIndexSize (nOfs, nLen, aBuf.length); + if (nLen == 0) + return 0; + + if (!m_aBuffer.hasRemaining ()) + { + // Read large requests directly from the channel, bypassing the buffer + if (nLen >= m_aBuffer.capacity ()) + { + final int nRead = m_aChannel.read (ByteBuffer.wrap (aBuf, nOfs, nLen)); + if (nRead > 0) + m_nChannelPos += nRead; + return nRead; + } + if (!_fill ()) + return -1; + } + final int nRead = Math.min (m_aBuffer.remaining (), nLen); + m_aBuffer.get (aBuf, nOfs, nRead); + return nRead; + } + + @Override + public long skip (final long nBytes) throws IOException + { + if (nBytes <= 0) + return 0; + final long nCurPos = _getLogicalPos (); + final long nSize = m_aChannel.size (); + if (nCurPos >= nSize) + return 0; + // Never skip beyond EOF + final long nSkipped = Math.min (nBytes, nSize - nCurPos); + _seek (nCurPos + nSkipped); + return nSkipped; + } + + @Override + public int available () throws IOException + { + final long nRemaining = m_aBuffer.remaining () + Math.max (0, m_aChannel.size () - m_nChannelPos); + return nRemaining > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) nRemaining; + } + + @Override + public boolean markSupported () + { + return true; + } + + /** + * {@inheritDoc}
+ * The read limit parameter is ignored: as {@link #reset()} only re-positions the underlying + * {@link FileChannel}, the mark can never be invalidated, no matter how many bytes are read. + */ + @Override + public void mark (final int nReadLimit) + { + m_nMarkPos = _getLogicalPos (); + } + + @Override + public void reset () throws IOException + { + _seek (m_nMarkPos); + } + + @Override + public void close () throws IOException + { + m_aChannel.close (); + } + + /** + * Factory method that logs an error and returns null if the file cannot be opened - + * same semantics as FileHelper.getBufferedInputStream. + * + * @param aFile + * The file to read. May not be null. + * @return null if the file cannot be opened for reading. + */ + @Nullable + public static MarkableFileInputStream create (@NonNull final File aFile) + { + try + { + return new MarkableFileInputStream (aFile); + } + catch (final IOException ex) + { + LOGGER.warn ("Failed to open file '" + aFile.getAbsolutePath () + "' for reading", ex); + return null; + } + } +} diff --git a/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java b/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java index e40874f6c..2cfa8b095 100644 --- a/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java +++ b/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java @@ -26,6 +26,7 @@ import java.io.File; import java.io.IOException; +import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.util.concurrent.ThreadLocalRandom; @@ -40,6 +41,7 @@ import com.helger.io.file.SimpleFileIO; import com.helger.mime.CMimeType; import com.helger.phase4.util.AS4ResourceHelper; +import com.helger.phase4.util.MarkableFileInputStream; import jakarta.mail.MessagingException; @@ -257,6 +259,60 @@ public void testIncomingEmpty () throws IOException, MessagingException } } + private static void _assertMarkableAndReReadable (@NonNull final WSS4JAttachment a) throws IOException + { + final InputStream aIS = a.getSourceStream (); + // Anything else buffers the whole attachment on the heap on mark/reset + assertTrue ("Not a " + MarkableFileInputStream.class.getSimpleName () + ": " + aIS, + aIS instanceof MarkableFileInputStream); + + // Consume the stream the way WSS4J digests a signed attachment + aIS.mark (Integer.MAX_VALUE); + final byte [] aDigested = aIS.readAllBytes (); + aIS.reset (); + + // Afterwards WSS4J re-uses the same stream object for the attachment + // content + assertArrayEquals (aDigested, aIS.readAllBytes ()); + } + + /** + * All file backed attachment source streams must support mark/reset with constant heap usage, + * because WSS4J digests each signed attachment via mark/read-to-end/reset. See issue #380. + */ + @Test + public void testFileBackedSourceStreamsAreMarkable () throws IOException, MessagingException + { + // Larger than the in-memory threshold, so that a temporary file is used + final byte [] aContent = new byte [WSS4JAttachment.MAX_IN_MEMORY_BYTES + 1024]; + ThreadLocalRandom.current ().nextBytes (aContent); + + final File fSrc = m_aRule.newFile ("payload.bin"); + SimpleFileIO.writeFile (fSrc, aContent); + + try (final AS4ResourceHelper aResHelper = new AS4ResourceHelper ()) + { + // Outgoing, uncompressed - the source file is used as-is + final AS4OutgoingAttachment aOA1 = AS4OutgoingAttachment.builder () + .data (fSrc) + .mimeType (CMimeType.APPLICATION_OCTET_STREAM) + .build (); + _assertMarkableAndReReadable (WSS4JAttachment.createOutgoingFileAttachment (aOA1, aResHelper)); + + // Outgoing, compressed - a temporary file with the compressed content + final AS4OutgoingAttachment aOA2 = AS4OutgoingAttachment.builder () + .data (aContent) + .mimeType (CMimeType.APPLICATION_OCTET_STREAM) + .compressionGZIP () + .build (); + _assertMarkableAndReReadable (WSS4JAttachment.createOutgoingFileAttachment (aOA2, aResHelper)); + + // Incoming, above the in-memory threshold - a temporary file + final AS4IncomingMimePart aMimePart = _createIncomingMimePart (aContent); + _assertMarkableAndReReadable (WSS4JAttachment.createIncomingFileAttachment (aMimePart, aResHelper)); + } + } + @Test public void testIncomingExactlyThresholdSize () throws IOException, MessagingException { diff --git a/phase4-lib/src/test/java/com/helger/phase4/util/MarkableFileInputStreamTest.java b/phase4-lib/src/test/java/com/helger/phase4/util/MarkableFileInputStreamTest.java new file mode 100644 index 000000000..c2cf3c873 --- /dev/null +++ b/phase4-lib/src/test/java/com/helger/phase4/util/MarkableFileInputStreamTest.java @@ -0,0 +1,231 @@ +/* + * Copyright (C) 2015-2026 Philip Helger (www.helger.com) + * philip[at]helger[dot]com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.helger.phase4.util; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; +import java.util.Random; + +import org.jspecify.annotations.NonNull; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import com.helger.base.io.nonblocking.NonBlockingByteArrayOutputStream; +import com.helger.io.file.SimpleFileIO; + +/** + * Test class for class {@link MarkableFileInputStream}. + */ +public final class MarkableFileInputStreamTest +{ + @Rule + public final TemporaryFolder m_aRule = new TemporaryFolder (); + + @NonNull + private static byte [] _createRandomBytes (final int nLen) + { + final byte [] ret = new byte [nLen]; + // Deterministic seed + new Random (20260803L).nextBytes (ret); + return ret; + } + + @NonNull + private File _createFile (final byte [] aContent) throws IOException + { + final File ret = m_aRule.newFile (); + SimpleFileIO.writeFile (ret, aContent); + return ret; + } + + /** + * Consume the stream exactly the way WSS4J's + * AttachmentContentSignatureTransform#processAttachment does when calculating the + * digest of a signed attachment: mark (Integer.MAX_VALUE), read to the end, + * reset (). Afterwards WSS4J re-uses the same stream object for the attachment + * content, so a subsequent full read must deliver the identical bytes from the start. + */ + @NonNull + private static byte [] _digestLikeWSS4J (@NonNull final InputStream aIS) throws IOException + { + assertTrue ("WSS4J relies on mark/reset support", aIS.markSupported ()); + aIS.mark (Integer.MAX_VALUE); + final byte [] ret = _readAll (aIS); + aIS.reset (); + return ret; + } + + @NonNull + private static byte [] _readAll (@NonNull final InputStream aIS) throws IOException + { + try (final NonBlockingByteArrayOutputStream aAll = new NonBlockingByteArrayOutputStream ()) + { + int nBytes; + // Same buffer size as WSS4J uses + final byte [] aBuf = new byte [8192]; + while ((nBytes = aIS.read (aBuf)) != -1) + aAll.write (aBuf, 0, nBytes); + return aAll.toByteArray (); + } + } + + @Test + public void testDigestThenReRead () throws IOException + { + // Clearly larger than the internal buffer + final byte [] aPayload = _createRandomBytes (1024 * 1024 + 17); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertArrayEquals ("The digested bytes must be the file content", aPayload, _digestLikeWSS4J (aIS)); + + // Re-read the same stream object from the beginning + assertArrayEquals ("The content after reset must be identical", aPayload, _readAll (aIS)); + } + } + + @Test + public void testMarkAtNonZeroOffset () throws IOException + { + final byte [] aPayload = _createRandomBytes (100_000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + // Read some bytes, crossing the internal buffer boundary + final byte [] aHead = new byte [MarkableFileInputStream.DEFAULT_BUFFER_SIZE + 100]; + int nRead = 0; + while (nRead < aHead.length) + nRead += aIS.read (aHead, nRead, aHead.length - nRead); + assertArrayEquals (Arrays.copyOf (aPayload, aHead.length), aHead); + + aIS.mark (Integer.MAX_VALUE); + final byte [] aTail1 = _readAll (aIS); + aIS.reset (); + final byte [] aTail2 = _readAll (aIS); + assertArrayEquals (Arrays.copyOfRange (aPayload, aHead.length, aPayload.length), aTail1); + assertArrayEquals (aTail1, aTail2); + } + } + + @Test + public void testResetWithoutMarkReReadsFromStart () throws IOException + { + final byte [] aPayload = _createRandomBytes (1000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertArrayEquals (aPayload, _readAll (aIS)); + aIS.reset (); + assertArrayEquals (aPayload, _readAll (aIS)); + } + } + + @Test + public void testSingleByteRead () throws IOException + { + // Small file, read byte by byte + final byte [] aPayload = _createRandomBytes (100); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f, 8)) + { + for (final byte b : aPayload) + assertEquals (b & 0xff, aIS.read ()); + assertEquals (-1, aIS.read ()); + } + } + + @Test + public void testEmptyFile () throws IOException + { + final File f = _createFile (new byte [0]); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertEquals (0, aIS.available ()); + assertEquals (-1, aIS.read ()); + aIS.mark (Integer.MAX_VALUE); + aIS.reset (); + assertEquals (-1, aIS.read ()); + } + } + + @Test + public void testSkipAndAvailable () throws IOException + { + final byte [] aPayload = _createRandomBytes (50_000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertEquals (aPayload.length, aIS.available ()); + assertEquals (0, aIS.skip (0)); + assertEquals (0, aIS.skip (-1)); + assertEquals (10_000, aIS.skip (10_000)); + assertEquals (aPayload.length - 10_000, aIS.available ()); + assertEquals (aPayload[10_000] & 0xff, aIS.read ()); + // Skipping beyond EOF is truncated + assertEquals (aPayload.length - 10_001, aIS.skip (Long.MAX_VALUE)); + assertEquals (0, aIS.skip (1)); + assertEquals (-1, aIS.read ()); + assertEquals (0, aIS.available ()); + } + } + + @Test + public void testReadLargerThanBuffer () throws IOException + { + // Requests larger than the internal buffer are served from the channel + // directly + final byte [] aPayload = _createRandomBytes (70_000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f, 1024)) + { + aIS.mark (Integer.MAX_VALUE); + final byte [] aBuf = new byte [aPayload.length]; + int nTotal = 0; + int nRead; + while (nTotal < aBuf.length && (nRead = aIS.read (aBuf, nTotal, aBuf.length - nTotal)) != -1) + nTotal += nRead; + assertEquals (aPayload.length, nTotal); + assertArrayEquals (aPayload, aBuf); + assertEquals (-1, aIS.read ()); + + // The mark must still be valid, no matter how many bytes were read + aIS.reset (); + assertArrayEquals (aPayload, _readAll (aIS)); + } + } + + @Test + public void testCreateNonExistingFile () + { + assertNull (MarkableFileInputStream.create (new File (m_aRule.getRoot (), "does-not-exist.bin"))); + } +}