diff --git a/.gitignore b/.gitignore index 0f802245..d5a7ca72 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ /target/ .env /.idea/ +.vscode \ No newline at end of file diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index b21e0dd5..79014137 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -1,21 +1,105 @@ package com.coveo.pushapiclient; import java.io.IOException; +import java.util.ArrayList; -// TODO: LENS-851 - Make public +/** + * Represents a queue for uploading documents using a specified upload strategy + */ class DocumentUploadQueue { private final UploadStrategy uploader; + private final int maxQueueSize = 5 * 1024 * 1024; + private ArrayList documentToAddList; + private ArrayList documentToDeleteList; + private int size; + /** + * Constructs a new DocumentUploadQueue object with a default maximum queue size + * limit of 5MB. + * + * @param uploader The upload strategy to be used for document uploads. + */ public DocumentUploadQueue(UploadStrategy uploader) { + this.documentToAddList = new ArrayList<>(); + this.documentToDeleteList = new ArrayList<>(); this.uploader = uploader; } + /** + * Flushes the accumulated documents by applying the upload strategy. + * + * @throws IOException If an I/O error occurs during the upload. + * @throws InterruptedException If the upload process is interrupted. + */ public void flush() throws IOException, InterruptedException { - throw new UnsupportedOperationException("Unimplemented method (TODO: LENS-856)"); + if (this.isEmpty()) { + return; + } + BatchUpdate batch = this.getBatch(); + // TODO: LENS-871: support concurrent requests + this.uploader.apply(batch); + this.size = 0; + this.documentToAddList.clear(); + this.documentToDeleteList.clear(); } + /** + * Adds a {@link DocumentBuilder} to the upload queue and flushes the queue if + * it exceeds the maximum content length. + * See {@link DocumentUploadQueue#flush}. + * + * @param document The document to be added to the index. + * @throws IOException If an I/O error occurs during the upload. + * @throws InterruptedException If the upload process is interrupted. + */ public void add(DocumentBuilder document) throws IOException, InterruptedException { - throw new UnsupportedOperationException("Unimplemented method (TODO: LENS-856)"); + if (document == null) { + return; + } + + final int sizeOfDoc = document.marshal().getBytes().length; + if (this.size + sizeOfDoc >= this.maxQueueSize) { + this.flush(); + } + if (document != null) { + documentToAddList.add(document); + this.size += sizeOfDoc; + } + } + + /** + * Adds a {@link DeleteDocument} to the upload queue and flushes the queue if + * it exceeds the maximum content length. + * See {@link DocumentUploadQueue#flush}. + * + * @param document The document to be delete from the index. + * @throws IOException If an I/O error occurs during the upload. + * @throws InterruptedException If the upload process is interrupted. + */ + public void add(DeleteDocument document) throws IOException, InterruptedException { + if (document == null) { + return; + } + + final int sizeOfDoc = document.marshalJsonObject().toString().getBytes().length; + if (this.size + sizeOfDoc >= this.maxQueueSize) { + this.flush(); + } + if (document != null) { + documentToDeleteList.add(document); + this.size += sizeOfDoc; + } + } + + public BatchUpdate getBatch() { + return new BatchUpdate( + new ArrayList(this.documentToAddList), + new ArrayList(this.documentToDeleteList)); + } + + public boolean isEmpty() { + // TODO: LENS-843: include partial document updates + return documentToAddList.isEmpty() && documentToDeleteList.isEmpty(); } } diff --git a/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java b/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java new file mode 100644 index 00000000..5e346a41 --- /dev/null +++ b/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java @@ -0,0 +1,192 @@ +package com.coveo.pushapiclient; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.io.IOException; +import java.util.ArrayList; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +public class DocumentUploadQueueTest { + + @Mock + private UploadStrategy uploadStrategy; + + @InjectMocks + private DocumentUploadQueue queue; + + private AutoCloseable closeable; + private DocumentBuilder documentToAdd; + private DeleteDocument documentToDelete; + + private int oneMegaByte = 1 * 1024 * 1024; + + private String generateStringFromBytes(int numBytes) { + // Check if the number of bytes is valid + if (numBytes <= 0) { + return ""; + } + + // Create a byte array with the specified length + byte[] bytes = new byte[numBytes]; + + // Fill the byte array with a pattern of ASCII characters + byte pattern = 65; // ASCII value for 'A' + for (int i = 0; i < numBytes; i++) { + bytes[i] = pattern; + } + + return new String(bytes); + } + + private DocumentBuilder generateDocumentFromSize(int numBytes) { + return new DocumentBuilder("https://my.document.uri?ref=1", + "My bulky document") + .withData(generateStringFromBytes(numBytes)); + } + + @Before + public void setup() { + String twoMegaByteData = generateStringFromBytes(2 * oneMegaByte); + + documentToAdd = new DocumentBuilder( + "https://my.document.uri?ref=1", + "My new document") + .withData(twoMegaByteData); + + documentToDelete = new DeleteDocument("https://my.document.uri?ref=3"); + + closeable = MockitoAnnotations.openMocks(this); + } + + @After + public void closeService() throws Exception { + closeable.close(); + } + + @Test + public void testIsEmpty() throws IOException, InterruptedException { + assertTrue(queue.isEmpty()); + } + + @Test + public void testIsNotEmpty() throws IOException, InterruptedException { + queue.add(documentToAdd); + assertFalse(queue.isEmpty()); + } + + @Test + public void testShouldReturnBatch() throws IOException, InterruptedException { + BatchUpdate batchUpdate = new BatchUpdate( + new ArrayList<>() { + { + add(documentToAdd); + } + }, new ArrayList<>() { + { + add(documentToDelete); + } + }); + queue.add(documentToAdd); + queue.add(documentToDelete); + + assertEquals(batchUpdate, queue.getBatch()); + } + + @Test + public void testFlushShouldNotUploadDocumentaWhenRequiredSizeIsNotMet() throws IOException, InterruptedException { + queue.add(documentToAdd); + queue.add(documentToDelete); + + verify(uploadStrategy, times(0)).apply(any(BatchUpdate.class)); + } + + @Test + public void testShouldAutomaticallyFlushAccumulatedDocuments() throws IOException, InterruptedException { + DocumentBuilder firstBulkyDocument = generateDocumentFromSize(2 * oneMegaByte); + DocumentBuilder secondBulkyDocument = generateDocumentFromSize(2 * oneMegaByte); + DocumentBuilder thirdBulkyDocument = generateDocumentFromSize(2 * oneMegaByte); + ArrayList emptyList = new ArrayList<>(); + BatchUpdate firstBatch = new BatchUpdate( + new ArrayList<>() { + { + add(firstBulkyDocument); + add(secondBulkyDocument); + } + }, emptyList); + + // Adding 3 documents of 2MB to the queue. After adding the first 2 documents, + // the queue size will reach 6MB, which exceeds the maximum queue size + // limit. Therefore, the 2 first added documents will automatically be uploaded + // to the source. + queue.add(firstBulkyDocument); + queue.add(secondBulkyDocument); + + // The 3rd document added to the queue will be included in a separate batch, + // which will not be uploaded unless the `flush()` method is called or until the + // queue size limit has been reached + queue.add(thirdBulkyDocument); + + verify(uploadStrategy, times(1)).apply(any(BatchUpdate.class)); + verify(uploadStrategy, times(1)).apply(firstBatch); + } + + @Test + public void testShouldManuallyFlushAccumulatedDocuments() throws IOException, InterruptedException { + DocumentBuilder firstBulkyDocument = generateDocumentFromSize(2 * oneMegaByte); + DocumentBuilder secondBulkyDocument = generateDocumentFromSize(2 * oneMegaByte); + DocumentBuilder thirdBulkyDocument = generateDocumentFromSize(2 * oneMegaByte); + ArrayList emptyList = new ArrayList<>(); + BatchUpdate firstBatch = new BatchUpdate( + new ArrayList<>() { + { + add(firstBulkyDocument); + add(secondBulkyDocument); + } + }, emptyList); + + BatchUpdate secondBatch = new BatchUpdate( + new ArrayList<>() { + { + add(thirdBulkyDocument); + } + }, emptyList); + + // Adding 3 documents of 2MB to the queue. After adding the first 2 documents, + // the queue size will reach 6MB, which exceeds the maximum queue size + // limit. Therefore, the 2 first added documents will automatically be uploaded + // to the source. + queue.add(firstBulkyDocument); + queue.add(secondBulkyDocument); + queue.add(thirdBulkyDocument); + + queue.flush(); + + // Additional flush will have no effect if documents where already flushed + queue.flush(); + + verify(uploadStrategy, times(2)).apply(any(BatchUpdate.class)); + verify(uploadStrategy, times(1)).apply(firstBatch); + verify(uploadStrategy, times(1)).apply(secondBatch); + } + + @Test + public void testAddingEmptyDocument() throws IOException, InterruptedException { + DocumentBuilder nullDocument = null; + + queue.add(nullDocument); + queue.flush(); + + verify(uploadStrategy, times(0)).apply(any(BatchUpdate.class)); + } +}