diff --git a/.gitignore b/.gitignore index f13c4961..5a5a0675 100644 --- a/.gitignore +++ b/.gitignore @@ -179,3 +179,4 @@ AZURE_BLOB_STORAGE_IMPROVEMENTS.md SFTP_STORAGE_ANALYSIS.md NFS_LOCKING.md RELEASE-REVIEW.md +ASYNC_REVIEW.md diff --git a/src/main/java/me/desair/tus/server/TusFileUploadService.java b/src/main/java/me/desair/tus/server/TusFileUploadService.java index 5afa99d6..5b5c20fa 100644 --- a/src/main/java/me/desair/tus/server/TusFileUploadService.java +++ b/src/main/java/me/desair/tus/server/TusFileUploadService.java @@ -533,7 +533,6 @@ public UploadInfo process( boolean wasInProgress = true; try (UploadLock lock = acquireUploadLock(method, request.getRequestURI())) { - request.setUploadLock(lock); // Evaluate initial progress state safely under the acquired upload lock // to avoid uncoordinated disk I/O or race conditions with concurrent requests. wasInProgress = checkWasInProgress(request, ownerKey); diff --git a/src/main/java/me/desair/tus/server/core/CorePatchRequestHandler.java b/src/main/java/me/desair/tus/server/core/CorePatchRequestHandler.java index cba46faa..8ed264c8 100644 --- a/src/main/java/me/desair/tus/server/core/CorePatchRequestHandler.java +++ b/src/main/java/me/desair/tus/server/core/CorePatchRequestHandler.java @@ -72,8 +72,7 @@ public HttpProblemDetails process( // If a locking service is provided, wrap the input stream in an InterruptibleInputStream // and register it with the locking service to allow for interruption of the upload. if (lockingService != null) { - InterruptibleInputStream interruptibleStream = - new InterruptibleInputStream(stream, servletRequest.getUploadLock()); + InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(stream); lockingService.registerInputStream(servletRequest.getRequestURI(), interruptibleStream); stream = interruptibleStream; } diff --git a/src/main/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandler.java b/src/main/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandler.java index 111fbc1e..6594e191 100644 --- a/src/main/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandler.java +++ b/src/main/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandler.java @@ -47,8 +47,7 @@ public HttpProblemDetails process( if (uploadInfo != null && uploadInfo.isUploadInProgress()) { InputStream stream = servletRequest.getContentInputStream(); if (uploadLockingService != null) { - InterruptibleInputStream interruptibleStream = - new InterruptibleInputStream(stream, servletRequest.getUploadLock()); + InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(stream); uploadLockingService.registerInputStream(location, interruptibleStream); stream = interruptibleStream; } diff --git a/src/main/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandler.java b/src/main/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandler.java index 52ba709e..5ec06fa4 100644 --- a/src/main/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandler.java +++ b/src/main/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandler.java @@ -58,8 +58,7 @@ public HttpProblemDetails process( InputStream is = servletRequest.getContentInputStream(); if (is != null) { if (uploadLockingService != null) { - InterruptibleInputStream interruptibleStream = - new InterruptibleInputStream(is, servletRequest.getUploadLock()); + InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(is); uploadLockingService.registerInputStream(requestUri, interruptibleStream); is = interruptibleStream; } diff --git a/src/main/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandler.java b/src/main/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandler.java index 8b018dba..1c47d68c 100644 --- a/src/main/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandler.java +++ b/src/main/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandler.java @@ -87,8 +87,7 @@ public HttpProblemDetails process( InputStream is = servletRequest.getContentInputStream(); if (is != null && servletRequest.getContentLengthLong() != 0) { if (uploadLockingService != null) { - InterruptibleInputStream interruptibleStream = - new InterruptibleInputStream(is, servletRequest.getUploadLock()); + InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(is); uploadLockingService.registerInputStream(uploadUri, interruptibleStream); is = interruptibleStream; } diff --git a/src/main/java/me/desair/tus/server/util/InterruptibleInputStream.java b/src/main/java/me/desair/tus/server/util/InterruptibleInputStream.java index 84a58488..7fe2257a 100644 --- a/src/main/java/me/desair/tus/server/util/InterruptibleInputStream.java +++ b/src/main/java/me/desair/tus/server/util/InterruptibleInputStream.java @@ -2,7 +2,6 @@ import java.io.IOException; import java.io.InputStream; -import me.desair.tus.server.upload.UploadLock; /** * An InputStream wrapper that can be interrupted by another thread. When interrupted, it throws an @@ -11,51 +10,19 @@ public class InterruptibleInputStream extends InputStream { private final InputStream delegate; - private UploadLock uploadLock; private volatile boolean interrupted = false; /** - * Constructs an interruptible input stream wrapping the given delegate stream and associating it - * with the provided {@link UploadLock}. + * Constructs an interruptible input stream wrapping the given delegate stream. * * @param delegate The delegate input stream to wrap - * @param uploadLock The upload lock associated with this stream, or null * @throws IllegalArgumentException if the delegate is null */ - public InterruptibleInputStream(InputStream delegate, UploadLock uploadLock) { + public InterruptibleInputStream(InputStream delegate) { if (delegate == null) { throw new IllegalArgumentException("Delegate InputStream cannot be null"); } this.delegate = delegate; - this.uploadLock = uploadLock; - } - - /** - * Constructs an interruptible input stream wrapping the given delegate stream. - * - * @param delegate The delegate input stream to wrap - * @throws IllegalArgumentException if the delegate is null - */ - public InterruptibleInputStream(InputStream delegate) { - this(delegate, null); - } - - /** - * Sets the corresponding {@link UploadLock} associated with this stream. - * - * @param uploadLock The upload lock - */ - public void setCorrespondingUploadLock(UploadLock uploadLock) { - this.uploadLock = uploadLock; - } - - /** - * Retrieves the corresponding {@link UploadLock} associated with this stream. - * - * @return The upload lock, or null if none is set - */ - public UploadLock getCorrespondingUploadLock() { - return uploadLock; } private void checkInterrupted() throws IOException { @@ -64,6 +31,17 @@ private void checkInterrupted() throws IOException { } } + /** + * Interrupts the active stream by marking it interrupted and closing the underlying delegate + * stream to abort any blocking socket read operations. + * + *

Note: Stream interruption intentionally does NOT release the upload lock. The thread + * processing the upload must continue holding the lock while it persists any buffered or received + * payload bytes to storage (e.g. S3, Azure Blob, or Disk) and updates the metadata offset. The + * upload lock is released only when the upload handler finishes processing and exits its + * try-with-resources lock block. This guarantees that concurrent or resuming requests (such as + * HEAD requests) wait for all in-flight bytes to be committed before reading the upload offset. + */ public void interrupt() { interrupted = true; try { @@ -71,9 +49,6 @@ public void interrupt() { } catch (IOException e) { // Ignore close exception during interrupt } - if (uploadLock != null) { - uploadLock.release(); - } } public boolean isInterrupted() { diff --git a/src/main/java/me/desair/tus/server/util/TusServletRequest.java b/src/main/java/me/desair/tus/server/util/TusServletRequest.java index d05ebdf8..37b0bd49 100644 --- a/src/main/java/me/desair/tus/server/util/TusServletRequest.java +++ b/src/main/java/me/desair/tus/server/util/TusServletRequest.java @@ -16,7 +16,6 @@ import me.desair.tus.server.HttpHeader; import me.desair.tus.server.TusExtension; import me.desair.tus.server.checksum.ChecksumAlgorithm; -import me.desair.tus.server.upload.UploadLock; import org.apache.commons.codec.binary.Base64; import org.apache.commons.io.input.BoundedInputStream; import org.apache.commons.lang3.StringUtils; @@ -24,7 +23,6 @@ public class TusServletRequest extends HttpServletRequestWrapper { - private UploadLock uploadLock; private BoundedInputStream countingInputStream; private Map digestInputStreamMap = new EnumMap<>(ChecksumAlgorithm.class); @@ -152,24 +150,6 @@ private boolean hasChunkedTransferEncoding() { return Strings.CI.equals("chunked", getHeader(HttpHeader.TRANSFER_ENCODING)); } - /** - * Retrieves the active {@link UploadLock} associated with this request. - * - * @return The active upload lock, or null if no lock is held - */ - public UploadLock getUploadLock() { - return uploadLock; - } - - /** - * Sets the active {@link UploadLock} associated with this request. - * - * @param uploadLock The active upload lock - */ - public void setUploadLock(UploadLock uploadLock) { - this.uploadLock = uploadLock; - } - private MessageDigest getMessageDigest(ChecksumAlgorithm algorithm) { if (digestInputStreamMap.containsKey(algorithm)) { return digestInputStreamMap.get(algorithm).getMessageDigest(); diff --git a/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java b/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java index df5f5c87..13812ae1 100644 --- a/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java +++ b/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java @@ -16,6 +16,7 @@ import static org.junit.Assert.fail; import jakarta.servlet.http.HttpServletResponse; +import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.nio.charset.StandardCharsets; @@ -1694,6 +1695,180 @@ public void run() { bgThread.join(2000); } + /** + * Tests that when an upload stream is interrupted/paused midway after receiving partial bytes, + * the thread performing the upload continues to hold the lock while persisting all received bytes + * to storage (S3, Azure Blob, Disk) and updating the metadata offset. A concurrent HEAD request + * (resuming the upload) waits for the upload thread to complete persistence and release the lock, + * receiving the up-to-date offset without rollback or data loss. + */ + @Test + public void testInterruptedUploadPersistsBytesBeforeHeadResponds() throws Exception { + // 1. Create upload resource for 1000 bytes + servletRequest.setMethod("POST"); + servletRequest.setRequestURI(UPLOAD_URI); + servletRequest.addHeader(HttpHeader.CONTENT_LENGTH, 0); + servletRequest.addHeader(HttpHeader.UPLOAD_LENGTH, 1000L); + servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0"); + tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY); + String location = + UPLOAD_URI + + StringUtils.substringAfter( + servletResponse.getHeader(HttpHeader.LOCATION), UPLOAD_URI); + + // 2. Start a PATCH in a background thread that streams 100 bytes and then blocks + final byte[] payload = new byte[100]; + java.util.Arrays.fill(payload, (byte) 'A'); + final java.util.concurrent.CountDownLatch bytesDelivered = + new java.util.concurrent.CountDownLatch(1); + final java.util.concurrent.atomic.AtomicReference bgException = + new java.util.concurrent.atomic.AtomicReference<>(); + + InputStream partialBlockingStream = + new InputStream() { + private final InputStream byteStream = new ByteArrayInputStream(payload); + private volatile boolean closed = false; + + @Override + public int read() throws IOException { + int b = byteStream.read(); + if (b != -1) { + return b; + } + // All 100 payload bytes delivered; notify that upload thread has buffered them + bytesDelivered.countDown(); + synchronized (this) { + while (!closed) { + try { + this.wait(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Stream read interrupted", e); + } + } + } + throw new IOException("Stream closed"); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int read = byteStream.read(b, off, len); + if (read != -1) { + return read; + } + bytesDelivered.countDown(); + synchronized (this) { + while (!closed) { + try { + this.wait(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Stream read interrupted", e); + } + } + } + throw new IOException("Stream closed"); + } + + @Override + public void close() throws IOException { + synchronized (this) { + closed = true; + this.notifyAll(); + } + } + }; + + final MockHttpServletRequest bgRequest = + new MockHttpServletRequest() { + @Override + public jakarta.servlet.ServletInputStream getInputStream() { + return new jakarta.servlet.ServletInputStream() { + @Override + public int read() throws IOException { + return partialBlockingStream.read(); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + return partialBlockingStream.read(b, off, len); + } + + @Override + public void close() throws IOException { + partialBlockingStream.close(); + } + + @Override + public boolean isFinished() { + return false; + } + + @Override + public boolean isReady() { + return true; + } + + @Override + public void setReadListener(jakarta.servlet.ReadListener readListener) {} + }; + } + }; + bgRequest.setMethod("PATCH"); + bgRequest.setRequestURI(location); + bgRequest.addHeader(HttpHeader.CONTENT_TYPE, "application/offset+octet-stream"); + bgRequest.addHeader(HttpHeader.CONTENT_LENGTH, 100); + bgRequest.addHeader(HttpHeader.UPLOAD_OFFSET, 0); + bgRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0"); + + Thread bgThread = + new Thread( + () -> { + try { + MockHttpServletResponse bgResponse = new MockHttpServletResponse(); + tusFileUploadService.process(bgRequest, bgResponse, OWNER_KEY); + } catch (Exception e) { + bgException.set(e); + } + }); + bgThread.start(); + + // Wait for the background thread to read all 100 bytes and enter blocking wait + assertTrue(bytesDelivered.await(5, java.util.concurrent.TimeUnit.SECONDS)); + + // 3. Concurrent HEAD request (resume request) arrives within the same second + MockHttpServletRequest headRequest = new MockHttpServletRequest(); + MockHttpServletResponse headResponse = new MockHttpServletResponse(); + headRequest.setMethod("HEAD"); + headRequest.setRequestURI(location); + headRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0"); + + tusFileUploadService.process(headRequest, headResponse, OWNER_KEY); + + // The HEAD request must succeed (204) and see the newly persisted offset of 100 bytes! + assertThat(headResponse.getStatus(), is(204)); + assertThat(headResponse.getHeader(HttpHeader.UPLOAD_OFFSET), is("100")); + + bgThread.join(3000); + + // 4. Resume upload: send remaining 900 bytes starting at offset 100 + MockHttpServletRequest resumeRequest = new MockHttpServletRequest(); + MockHttpServletResponse resumeResponse = new MockHttpServletResponse(); + resumeRequest.setMethod("PATCH"); + resumeRequest.setRequestURI(location); + resumeRequest.addHeader(HttpHeader.CONTENT_TYPE, "application/offset+octet-stream"); + resumeRequest.addHeader(HttpHeader.CONTENT_LENGTH, 900); + resumeRequest.addHeader(HttpHeader.UPLOAD_OFFSET, 100); + resumeRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0"); + byte[] remainingPayload = new byte[900]; + java.util.Arrays.fill(remainingPayload, (byte) 'B'); + resumeRequest.setContent(remainingPayload); + + tusFileUploadService.process(resumeRequest, resumeResponse, OWNER_KEY); + assertThat(resumeResponse.getStatus(), is(204)); + assertThat(resumeResponse.getHeader(HttpHeader.UPLOAD_OFFSET), is("1000")); + } + @Test public void testCreationWithUploadOptions() throws Exception { reset(); diff --git a/src/test/java/me/desair/tus/server/core/CorePatchRequestHandlerTest.java b/src/test/java/me/desair/tus/server/core/CorePatchRequestHandlerTest.java index 4f4372bf..05576a70 100644 --- a/src/test/java/me/desair/tus/server/core/CorePatchRequestHandlerTest.java +++ b/src/test/java/me/desair/tus/server/core/CorePatchRequestHandlerTest.java @@ -2,7 +2,7 @@ import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; -import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.eq; @@ -20,7 +20,6 @@ import me.desair.tus.server.exception.UploadNotFoundException; import me.desair.tus.server.upload.UploadId; import me.desair.tus.server.upload.UploadInfo; -import me.desair.tus.server.upload.UploadLock; import me.desair.tus.server.upload.UploadLockingService; import me.desair.tus.server.upload.UploadStorageService; import me.desair.tus.server.util.InterruptibleInputStream; @@ -171,9 +170,7 @@ public void processWithLockingService() throws Exception { .thenReturn(updatedInfo); UploadLockingService mockLocking = mock(UploadLockingService.class); - UploadLock mockLock = mock(UploadLock.class); TusServletRequest tusRequest = new TusServletRequest(servletRequest); - tusRequest.setUploadLock(mockLock); handler.process( HttpMethod.PATCH, @@ -188,7 +185,7 @@ public void processWithLockingService() throws Exception { ArgumentCaptor.forClass(InterruptibleInputStream.class); verify(mockLocking, times(1)) .registerInputStream(eq(servletRequest.getRequestURI()), captor.capture()); - assertEquals(mockLock, captor.getValue().getCorrespondingUploadLock()); + assertNotNull(captor.getValue()); verify(uploadStorageService, times(1)).append(eq(info), any(InputStream.class)); } diff --git a/src/test/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandlerTest.java b/src/test/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandlerTest.java index 0cca689a..9001c2c0 100644 --- a/src/test/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandlerTest.java +++ b/src/test/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandlerTest.java @@ -2,7 +2,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.is; -import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -14,7 +14,6 @@ import me.desair.tus.server.HttpHeader; import me.desair.tus.server.HttpMethod; import me.desair.tus.server.upload.UploadInfo; -import me.desair.tus.server.upload.UploadLock; import me.desair.tus.server.upload.UploadLockingService; import me.desair.tus.server.upload.UploadStorageService; import me.desair.tus.server.util.InterruptibleInputStream; @@ -45,12 +44,10 @@ public void testProcessWithLockingService() throws Exception { TusServletResponse response = mock(TusServletResponse.class); UploadStorageService storageService = mock(UploadStorageService.class); UploadLockingService lockingService = mock(UploadLockingService.class); - UploadLock mockLock = mock(UploadLock.class); when(request.getHeader(HttpHeader.CONTENT_LENGTH)).thenReturn("5"); when(response.getHeader(HttpHeader.LOCATION)).thenReturn("/files/123"); when(request.getContentInputStream()).thenReturn(new ByteArrayInputStream("hello".getBytes())); - when(request.getUploadLock()).thenReturn(mockLock); UploadInfo uploadInfo = new UploadInfo(); uploadInfo.setLength(10L); @@ -69,7 +66,7 @@ public void testProcessWithLockingService() throws Exception { ArgumentCaptor captor = ArgumentCaptor.forClass(InterruptibleInputStream.class); verify(lockingService).registerInputStream(eq("/files/123"), captor.capture()); - assertEquals(mockLock, captor.getValue().getCorrespondingUploadLock()); + assertNotNull(captor.getValue()); verify(response).setHeader(HttpHeader.UPLOAD_OFFSET, "5"); } diff --git a/src/test/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandlerTest.java b/src/test/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandlerTest.java index 6573ddb9..46f3a3d1 100644 --- a/src/test/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandlerTest.java +++ b/src/test/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandlerTest.java @@ -2,12 +2,11 @@ import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -15,7 +14,6 @@ import me.desair.tus.server.HttpMethod; import me.desair.tus.server.upload.UploadId; import me.desair.tus.server.upload.UploadInfo; -import me.desair.tus.server.upload.UploadLock; import me.desair.tus.server.upload.UploadLockingService; import me.desair.tus.server.upload.UploadStorageService; import me.desair.tus.server.util.InterruptibleInputStream; @@ -80,9 +78,7 @@ public void testProcessPartialAppend() throws Exception { when(storageService.getUploadInfo("/files/append-id", "owner")).thenReturn(info); when(storageService.append(any(UploadInfo.class), any())).thenReturn(updated); - UploadLock mockLock = mock(UploadLock.class); TusServletRequest tusRequest = new TusServletRequest(request); - tusRequest.setUploadLock(mockLock); handler.process( HttpMethod.PATCH, @@ -100,7 +96,7 @@ public void testProcessPartialAppend() throws Exception { ArgumentCaptor captor = ArgumentCaptor.forClass(InterruptibleInputStream.class); verify(lockingService).registerInputStream(eq("/files/append-id"), captor.capture()); - assertEquals(mockLock, captor.getValue().getCorrespondingUploadLock()); + assertNotNull(captor.getValue()); } @Test diff --git a/src/test/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandlerTest.java b/src/test/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandlerTest.java index 05efa2be..f6dce0c7 100644 --- a/src/test/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandlerTest.java +++ b/src/test/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandlerTest.java @@ -2,13 +2,12 @@ import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.nullable; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -16,7 +15,6 @@ import me.desair.tus.server.HttpMethod; import me.desair.tus.server.upload.UploadId; import me.desair.tus.server.upload.UploadInfo; -import me.desair.tus.server.upload.UploadLock; import me.desair.tus.server.upload.UploadLockingService; import me.desair.tus.server.upload.UploadStorageService; import me.desair.tus.server.util.InterruptibleInputStream; @@ -111,9 +109,7 @@ public void testProcessCreationRegistersInputStream() throws Exception { when(storageService.create(any(UploadInfo.class), nullable(String.class))).thenReturn(info); when(storageService.append(any(UploadInfo.class), any())).thenReturn(info); - UploadLock mockLock = mock(UploadLock.class); TusServletRequest tusRequest = new TusServletRequest(request); - tusRequest.setUploadLock(mockLock); handler.process( HttpMethod.POST, @@ -127,7 +123,7 @@ public void testProcessCreationRegistersInputStream() throws Exception { ArgumentCaptor captor = ArgumentCaptor.forClass(InterruptibleInputStream.class); verify(lockingService).registerInputStream(eq("/files/creation-id"), captor.capture()); - assertEquals(mockLock, captor.getValue().getCorrespondingUploadLock()); + assertNotNull(captor.getValue()); } @Test diff --git a/src/test/java/me/desair/tus/server/util/InterruptibleInputStreamTest.java b/src/test/java/me/desair/tus/server/util/InterruptibleInputStreamTest.java index 3b975772..f250caed 100644 --- a/src/test/java/me/desair/tus/server/util/InterruptibleInputStreamTest.java +++ b/src/test/java/me/desair/tus/server/util/InterruptibleInputStreamTest.java @@ -1,13 +1,10 @@ package me.desair.tus.server.util; import static org.junit.Assert.*; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; -import me.desair.tus.server.upload.UploadLock; import org.junit.Test; public class InterruptibleInputStreamTest { @@ -44,6 +41,20 @@ public void testInterruptBeforeRead() throws IOException { } catch (IOException e) { assertTrue(e.getMessage().contains("interrupted")); } + + try { + iis.read(new byte[2]); + fail("Expected IOException on read(byte[]) after interrupt"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("interrupted")); + } + + try { + iis.read(new byte[2], 0, 1); + fail("Expected IOException on read(byte[], int, int) after interrupt"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("interrupted")); + } } @Test @@ -97,6 +108,7 @@ public void testDelegateMethods() throws IOException { assertEquals(3, iis.read()); iis.reset(); assertEquals(1, iis.read()); + iis.close(); // Test reset after interrupt iis.interrupt(); @@ -167,50 +179,47 @@ public int read(byte[] b, int off, int len) throws IOException { } @Test - public void testInterruptWithUploadLock() throws IOException { - InputStream bis = new ByteArrayInputStream(new byte[] {1, 2, 3}); - UploadLock uploadLock = mock(UploadLock.class); - - InterruptibleInputStream iis = new InterruptibleInputStream(bis, uploadLock); - assertEquals(uploadLock, iis.getCorrespondingUploadLock()); - - iis.interrupt(); - - assertTrue(iis.isInterrupted()); - verify(uploadLock).release(); - } + public void testReadAfterUnderlyingStreamException() throws IOException { + InputStream faultyStream = + new InputStream() { + @Override + public int read() throws IOException { + throw new IOException("faulty read"); + } - @Test - public void testSetAndGetCorrespondingUploadLock() throws IOException { - InputStream bis = new ByteArrayInputStream(new byte[] {1, 2, 3}); - InterruptibleInputStream iis = new InterruptibleInputStream(bis); - assertNull(iis.getCorrespondingUploadLock()); + @Override + public int read(byte[] b) throws IOException { + throw new IOException("faulty read array"); + } - UploadLock uploadLock = mock(UploadLock.class); - iis.setCorrespondingUploadLock(uploadLock); - assertEquals(uploadLock, iis.getCorrespondingUploadLock()); + @Override + public int read(byte[] b, int off, int len) throws IOException { + throw new IOException("faulty read range"); + } + }; + InterruptibleInputStream iis = new InterruptibleInputStream(faultyStream); iis.interrupt(); - verify(uploadLock).release(); - } - @Test - public void testInterruptWithoutUploadLock() { - InputStream bis = new ByteArrayInputStream(new byte[] {1, 2, 3}); - InterruptibleInputStream iis = new InterruptibleInputStream(bis, null); + try { + iis.read(); + fail("Expected IOException"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("interrupted")); + } - iis.interrupt(); - assertTrue(iis.isInterrupted()); - } + try { + iis.read(new byte[1]); + fail("Expected IOException"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("interrupted")); + } - @Test - public void testNullConstructorArgWithUploadLock() { - UploadLock uploadLock = mock(UploadLock.class); try { - new InterruptibleInputStream(null, uploadLock); - fail("Expected IllegalArgumentException"); - } catch (IllegalArgumentException e) { - assertEquals("Delegate InputStream cannot be null", e.getMessage()); + iis.read(new byte[1], 0, 1); + fail("Expected IOException"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("interrupted")); } } } diff --git a/src/test/java/me/desair/tus/server/util/TusServletRequestTest.java b/src/test/java/me/desair/tus/server/util/TusServletRequestTest.java index dd4d4d56..4ac1f621 100644 --- a/src/test/java/me/desair/tus/server/util/TusServletRequestTest.java +++ b/src/test/java/me/desair/tus/server/util/TusServletRequestTest.java @@ -21,7 +21,6 @@ import me.desair.tus.server.HttpHeader; import me.desair.tus.server.TusExtension; import me.desair.tus.server.checksum.ChecksumAlgorithm; -import me.desair.tus.server.upload.UploadLock; import org.apache.commons.io.IOUtils; import org.junit.Before; import org.junit.Test; @@ -245,14 +244,4 @@ public void getHeaderNotFound() { when(servletRequest.getHeader("X-My-Header")).thenReturn(null); assertNull(request.getHeader("X-My-Header")); } - - @Test - public void testGetAndSetUploadLock() { - assertNull(request.getUploadLock()); - - UploadLock uploadLock = mock(UploadLock.class); - request.setUploadLock(uploadLock); - - assertEquals(uploadLock, request.getUploadLock()); - } }