feat: implement PushService class

src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java

@@ -61,18 +61,16 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti
         if (this.size + sizeOfDoc >= this.maxQueueSize) {
             this.flush();
         }
-        if (document != null) {
-            documentToAddList.add(document);
-            this.size += sizeOfDoc;
-        }
+        documentToAddList.add(document);
+        this.size += sizeOfDoc;
     }
 
     /**
-     * Adds a {@link DeleteDocument} to the upload queue and flushes the queue if
+     * Adds the {@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.
+     * @param document The document to be deleted from the index.
      * @throws IOException          If an I/O error occurs during the upload.
      * @throws InterruptedException If the upload process is interrupted.
      */
@@ -85,10 +83,8 @@ public void add(DeleteDocument document) throws IOException, InterruptedExceptio
         if (this.size + sizeOfDoc >= this.maxQueueSize) {
             this.flush();
         }
-        if (document != null) {
-            documentToDeleteList.add(document);
-            this.size += sizeOfDoc;
-        }
+        documentToDeleteList.add(document);
+        this.size += sizeOfDoc;
     }
 
     public BatchUpdate getBatch() {

src/main/java/com/coveo/pushapiclient/PushService.java

@@ -0,0 +1,51 @@
+package com.coveo.pushapiclient;
+
+import java.io.IOException;
+import java.net.http.HttpResponse;
+
+import com.google.gson.Gson;
+
+public class PushService {
+    private final PushEnabledSource source;
+    private final PlatformClient platformClient;
+    private PushServiceInternal service;
+
+    public PushService(PushEnabledSource source) {
+        String apiKey = source.getApiKey();
+        String organizationId = source.getOrganizationId();
+        PlatformUrl platformUrl = source.getPlatformUrl();
+        UploadStrategy uploader = this.getUploadStrategy();
+        DocumentUploadQueue queue = new DocumentUploadQueue(uploader);
+
+        this.platformClient = new PlatformClient(apiKey, organizationId, platformUrl);
+        this.service = new PushServiceInternal(queue);
+        this.source = source;
+    }
+
+    public void addOrUpdate(DocumentBuilder document) throws IOException, InterruptedException {
+        // TODO: LENS-843: include partial document updates
+        this.service.addOrUpdate(document);
+    }
+
+    public void delete(DeleteDocument document) throws IOException, InterruptedException {
+        this.service.delete(document);
+    }
+
+    public void close() throws IOException, InterruptedException {
+        this.service.close();
+    }
+
+    private UploadStrategy getUploadStrategy() {
+        return (batchUpdate) -> {
+            String sourceId = this.getSourceId();
+            HttpResponse<String> resFileContainer = this.platformClient.createFileContainer();
+            FileContainer fileContainer = new Gson().fromJson(resFileContainer.body(), FileContainer.class);
+            this.platformClient.uploadContentToFileContainer(fileContainer, new Gson().toJson(batchUpdate.marshal()));
+            return this.platformClient.pushFileContainerContent(sourceId, fileContainer);
+        };
+    }
+
+    private String getSourceId() {
+        return this.source.getId();
+    }
+}

src/main/java/com/coveo/pushapiclient/PushServiceInternal.java

@@ -0,0 +1,24 @@
+package com.coveo.pushapiclient;
+
+import java.io.IOException;
+
+public class PushServiceInternal {
+    private DocumentUploadQueue queue;
+
+    public PushServiceInternal(DocumentUploadQueue queue) {
+        this.queue = queue;
+    }
+
+    public void addOrUpdate(DocumentBuilder document) throws IOException, InterruptedException {
+        this.queue.add(document);
+    }
+
+    public void delete(DeleteDocument document) throws IOException, InterruptedException {
+        this.queue.add(document);
+    }
+
+    public void close() throws IOException, InterruptedException {
+        queue.flush();
+    }
+
+}

src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java

@@ -104,10 +104,17 @@ public void testShouldReturnBatch() throws IOException, InterruptedException {
     }
 
     @Test
-    public void testFlushShouldNotUploadDocumentaWhenRequiredSizeIsNotMet() throws IOException, InterruptedException {
+    public void testFlushShouldNotUploadDocumentsWhenRequiredSizeIsNotMet() throws IOException, InterruptedException {
+        // Adding 2MB document to the queue => queue has now 3MB of free space
+        // (5MB - 2MB = 3MB)
         queue.add(documentToAdd);
+        // Adding 2MB document to the queue => queue has now 1MB of free space
+        // (3MB - 2MB = 1MB)
         queue.add(documentToDelete);
 
+        // The maximum queue size has not been reached yet (1MB left of free space).
+        // Therefore, the accumulated documents will not be automatically flushed.
+        // Unless the user runs `.flush()` the queue will keep the 4MB of documents
         verify(uploadStrategy, times(0)).apply(any(BatchUpdate.class));
     }
 
@@ -127,10 +134,11 @@ public void testShouldAutomaticallyFlushAccumulatedDocuments() throws IOExceptio
 
         // 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.
+        // limit by 1MB. Therefore, the 2 first added documents will automatically be
+        // uploaded to the source.
         queue.add(firstBulkyDocument);
         queue.add(secondBulkyDocument);
+        verify(uploadStrategy, times(0)).apply(any(BatchUpdate.class));
 
         // 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

src/test/java/com/coveo/pushapiclient/PushServiceInternalTest.java

@@ -0,0 +1,75 @@
+package com.coveo.pushapiclient;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import com.coveo.pushapiclient.exceptions.NoOpenStreamException;
+
+import java.io.IOException;
+import java.net.http.HttpResponse;
+
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class PushServiceInternalTest {
+    @Mock
+    private DocumentUploadQueue queue;
+
+    @InjectMocks
+    private PushServiceInternal service;
+
+    @Mock
+    private HttpResponse<String> httpResponse;
+
+    private AutoCloseable closeable;
+    private DocumentBuilder documentA;
+    private DocumentBuilder documentB;
+    private DeleteDocument documentC;
+
+    @Before
+    public void setUp() throws Exception {
+        documentA = new DocumentBuilder("https://my.document.uri?ref=1", "My first document title");
+        documentB = new DocumentBuilder("https://my.document.uri?ref=2", "My second document title");
+        documentC = new DeleteDocument("https://my.document.uri?ref=3");
+
+        closeable = MockitoAnnotations.openMocks(this);
+
+    }
+
+    @After
+    public void closeService() throws Exception {
+        closeable.close();
+    }
+
+    @Test
+    public void testShouldAddNewDocumentToQueue() throws IOException, InterruptedException {
+        service.addOrUpdate(documentA);
+        service.addOrUpdate(documentB);
+
+        verify(this.queue, times(1)).add(documentA);
+        verify(this.queue, times(1)).add(documentB);
+    }
+
+    @Test
+    public void testAddShouldAddDocumentToDeleteToQueue() throws IOException, InterruptedException {
+        service.delete(documentC);
+
+        verify(queue, times(1)).add(documentC);
+    }
+
+    @Test
+    public void testCloseShouldFlushBufferedDocuments()
+            throws IOException, InterruptedException, NoOpenStreamException {
+        service.addOrUpdate(documentA);
+        service.addOrUpdate(documentB);
+        service.delete(documentC);
+        service.close();
+
+        verify(queue, times(1)).flush();
+    }
+
+}