diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index 79014137..66355664 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/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() { diff --git a/src/main/java/com/coveo/pushapiclient/PushService.java b/src/main/java/com/coveo/pushapiclient/PushService.java new file mode 100644 index 00000000..185203f0 --- /dev/null +++ b/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 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(); + } +} diff --git a/src/main/java/com/coveo/pushapiclient/PushServiceInternal.java b/src/main/java/com/coveo/pushapiclient/PushServiceInternal.java new file mode 100644 index 00000000..7f3f0c4c --- /dev/null +++ b/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(); + } + +} diff --git a/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java b/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java index 5e346a41..97c11087 100644 --- a/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java +++ b/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 diff --git a/src/test/java/com/coveo/pushapiclient/PushServiceInternalTest.java b/src/test/java/com/coveo/pushapiclient/PushServiceInternalTest.java new file mode 100644 index 00000000..2a214f7e --- /dev/null +++ b/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 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(); + } + +} \ No newline at end of file