feat: implement DocumentUploadQueue class

.gitignore

@@ -1,3 +1,4 @@
 /target/
 .env
 /.idea/
+.vscode

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<DocumentBuilder> documentToAddList;
+    private ArrayList<DeleteDocument> 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<DocumentBuilder>(this.documentToAddList),
+                new ArrayList<DeleteDocument>(this.documentToDeleteList));
+    }
+
+    public boolean isEmpty() {
+        // TODO: LENS-843: include partial document updates
+        return documentToAddList.isEmpty() && documentToDeleteList.isEmpty();
     }
 
 }

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<DeleteDocument> 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<DeleteDocument> 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));
+    }
+}