diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index e3f75203..1768f2ff 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -15,10 +15,10 @@ jobs: steps: - uses: actions/checkout@v2 - - name: Set up JDK 16 + - name: Set up JDK 11 uses: actions/setup-java@v2 with: - java-version: '16' + java-version: '11' distribution: 'adopt' - name: Build with Maven run: mvn -B package --file pom.xml diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index f69a22c6..4b2d418e 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -14,4 +14,4 @@ jobs: analyze-java: uses: coveo/public-actions/.github/workflows/java-maven-openjdk11-codeql.yml@main with: - runs-on: "['linux', 'x64', 'ec2.instance-type = t3.large']" + runs-on: ubuntu-latest diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java new file mode 100644 index 00000000..b21e0dd5 --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -0,0 +1,21 @@ +package com.coveo.pushapiclient; + +import java.io.IOException; + +// TODO: LENS-851 - Make public +class DocumentUploadQueue { + private final UploadStrategy uploader; + + public DocumentUploadQueue(UploadStrategy uploader) { + this.uploader = uploader; + } + + public void flush() throws IOException, InterruptedException { + throw new UnsupportedOperationException("Unimplemented method (TODO: LENS-856)"); + } + + public void add(DocumentBuilder document) throws IOException, InterruptedException { + throw new UnsupportedOperationException("Unimplemented method (TODO: LENS-856)"); + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/PlatformClient.java b/src/main/java/com/coveo/pushapiclient/PlatformClient.java index f54aca7b..7f0a1705 100644 --- a/src/main/java/com/coveo/pushapiclient/PlatformClient.java +++ b/src/main/java/com/coveo/pushapiclient/PlatformClient.java @@ -282,6 +282,47 @@ public HttpResponse deleteDocument(String sourceId, String documentId, B return this.httpClient.send(request, HttpResponse.BodyHandlers.ofString()); } + public HttpResponse openStream(String sourceId) throws IOException, InterruptedException { + String[] headers = this.getHeaders(this.getAuthorizationHeader(), this.getContentTypeApplicationJSONHeader()); + // TODO: LENS-875: standardize string manipulation + URI uri = URI.create(this.getBasePushURL() + String.format("/sources/%s/stream/open", sourceId)); + + // TODO: LENS-876: reduce code duplication + HttpRequest request = HttpRequest.newBuilder() + .headers(headers) + .uri(uri) + .POST(HttpRequest.BodyPublishers.ofString("")) + .build(); + + return this.httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + } + + public HttpResponse closeStream(String sourceId, String streamId) throws IOException, InterruptedException { + String[] headers = this.getHeaders(this.getAuthorizationHeader(), this.getContentTypeApplicationJSONHeader()); + URI uri = URI.create(this.getBasePushURL() + String.format("/sources/%s/stream/%s/close", sourceId, streamId)); + + HttpRequest request = HttpRequest.newBuilder() + .headers(headers) + .uri(uri) + .POST(HttpRequest.BodyPublishers.ofString("")) + .build(); + + return this.httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + } + + public HttpResponse requireStreamChunk(String sourceId, String streamId) throws IOException, InterruptedException { + String[] headers = this.getHeaders(this.getAuthorizationHeader(), this.getContentTypeApplicationJSONHeader()); + URI uri = URI.create(this.getBasePushURL() + String.format("/sources/%s/stream/%s/chunk", sourceId, streamId)); + + HttpRequest request = HttpRequest.newBuilder() + .headers(headers) + .uri(uri) + .POST(HttpRequest.BodyPublishers.ofString("")) + .build(); + + return this.httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + } + /** * Create a file container. See [Creating a File Container](https://docs.coveo.com/en/43). * diff --git a/src/main/java/com/coveo/pushapiclient/StreamResponse.java b/src/main/java/com/coveo/pushapiclient/StreamResponse.java new file mode 100644 index 00000000..61f2ceb7 --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/StreamResponse.java @@ -0,0 +1,11 @@ +package com.coveo.pushapiclient; + +import java.util.Map; + +public class StreamResponse { + public String uploadUri; + public String fileId; + public String streamId; + public Map requiredHeaders; + +} diff --git a/src/main/java/com/coveo/pushapiclient/StreamService.java b/src/main/java/com/coveo/pushapiclient/StreamService.java new file mode 100644 index 00000000..b9212e20 --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -0,0 +1,116 @@ +package com.coveo.pushapiclient; + +import java.io.IOException; +import java.net.http.HttpResponse; + +import com.coveo.pushapiclient.exceptions.NoOpenStreamException; +import com.google.gson.Gson; + +// TODO: LENS-851 - Make public +class StreamService { + private final StreamEnabledSource source; + private final PlatformClient platformClient; + private StreamServiceInternal service; + private String streamId; + private DocumentUploadQueue queue; + + /** + * Creates a service to stream your documents to the provided source by + * interacting with the Stream API. + * + *

+ * To perform full document + * updates, use the {@PushService}, since pushing documents with the + * {@StreamService} is equivalent to triggering a full source rebuild. The + * {@StreamService} can also be used for an initial catalog upload. + * + * @param source The source to which you want to send your documents. + */ + public StreamService(StreamEnabledSource source) { + String apiKey = source.getApiKey(); + String organizationId = source.getOrganizationId(); + PlatformUrl platformUrl = source.getPlatformUrl(); + UploadStrategy uploader = this.getUploadStrategy(); + + this.source = source; + this.queue = new DocumentUploadQueue(uploader); + this.platformClient = new PlatformClient(apiKey, organizationId, platformUrl); + this.service = new StreamServiceInternal(this.source, this.queue, this.platformClient); + } + + /** + * Adds documents to the previously specified source. + * This function will open a stream before uploading documents into it. + * + *

+ * If called several times, the service will automatically batch documents and + * create new stream chunks whenever the data payload exceeds the + * batch size limit + * set for the Stream API. + * + *

+ * Once there are no more documents to add, it is important to call the {@link StreamService#close} function + * in order to send any buffered documents and close the open stream. + * Otherwise, changes will not be reflected in the index. + * + *

+ * 

+     * {@code
+     * //...
+     * StreamService service = new StreamService(source));
+     * for (DocumentBuilder document : fictionalDocumentList) {
+     *     service.add(document);
+     * }
+     * service.close(document);
+     *
+ * + *

+ * For more code samples, visit Stream data to your catalog source + * + * @param document The documentBuilder to add to your source + * @throws InterruptedException + * @throws IOException + */ + public void add(DocumentBuilder document) throws IOException, InterruptedException { + this.service.add(document); + } + + /** + * Sends any buffered documents and closes the stream. + * + *

+ * Upon invoking this method, any indexed items not added through this {@link StreamService} instance will be removed. + * All documents added from the initialization of the service until the invocation of the {@link StreamService#close} function + * will completely replace the previous content of the source. + * + *

+ * When you upload a catalog into a source, it will replace the previous content + * of the source completely. Expect a 15-minute delay for the removal of the old + * items from the index. + * + * @return + * @throws IOException + * @throws InterruptedException + * @throws NoOpenStreamException + */ + public HttpResponse close() throws IOException, InterruptedException, NoOpenStreamException { + return this.service.close(); + } + + private UploadStrategy getUploadStrategy() { + return (batchUpdate) -> { + String sourceId = this.getSourceId(); + HttpResponse resFileContainer = this.platformClient.requireStreamChunk(sourceId, this.streamId); + FileContainer fileContainer = new Gson().fromJson(resFileContainer.body(), FileContainer.class); + String batchUpdateJson = new Gson().toJson(batchUpdate.marshal()); + return this.platformClient.uploadContentToFileContainer(fileContainer, + batchUpdateJson); + + }; + } + + private String getSourceId() { + return this.source.getId(); + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/StreamServiceInternal.java b/src/main/java/com/coveo/pushapiclient/StreamServiceInternal.java new file mode 100644 index 00000000..81f47c85 --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/StreamServiceInternal.java @@ -0,0 +1,52 @@ +package com.coveo.pushapiclient; + +import java.io.IOException; +import java.net.http.HttpResponse; + +import com.coveo.pushapiclient.exceptions.NoOpenStreamException; +import com.google.gson.Gson; + +/** + * For internal use only. Made to easily test the service without having to use PowerMock + */ +class StreamServiceInternal { + private final StreamEnabledSource source; + private final PlatformClient platformClient; + private String streamId; + private DocumentUploadQueue queue; + + public StreamServiceInternal(StreamEnabledSource source, DocumentUploadQueue queue, PlatformClient platformClient) { + this.source = source; + this.queue = queue; + this.platformClient = platformClient; + } + + public void add(DocumentBuilder document) throws IOException, InterruptedException { + if (this.streamId == null) { + this.streamId = this.getStreamId(); + } + queue.add(document); + } + + public HttpResponse close() throws IOException, InterruptedException, NoOpenStreamException { + if (this.streamId == null) { + throw new NoOpenStreamException( + "No open stream detected. A stream will automatically be opened once you start adding documents."); + } + queue.flush(); + String sourceId = this.getSourceId(); + return this.platformClient.closeStream(sourceId, this.streamId); + } + + private String getStreamId() throws IOException, InterruptedException { + String sourceId = this.getSourceId(); + HttpResponse response = this.platformClient.openStream(sourceId); + StreamResponse streamResponse = new Gson().fromJson(response.body(), StreamResponse.class); + return streamResponse.streamId; + } + + private String getSourceId() { + return this.source.getId(); + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/UploadStrategy.java b/src/main/java/com/coveo/pushapiclient/UploadStrategy.java new file mode 100644 index 00000000..e77dd01c --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/UploadStrategy.java @@ -0,0 +1,9 @@ +package com.coveo.pushapiclient; + +import java.io.IOException; +import java.net.http.HttpResponse; + +@FunctionalInterface +public interface UploadStrategy { + HttpResponse apply(BatchUpdate batchUpdate) throws IOException, InterruptedException; +} diff --git a/src/main/java/com/coveo/pushapiclient/exceptions/NoOpenStreamException.java b/src/main/java/com/coveo/pushapiclient/exceptions/NoOpenStreamException.java new file mode 100644 index 00000000..91b2cf3a --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/exceptions/NoOpenStreamException.java @@ -0,0 +1,7 @@ +package com.coveo.pushapiclient.exceptions; + +public class NoOpenStreamException extends Exception { + public NoOpenStreamException(String errorMessage) { + super(errorMessage); + } +} diff --git a/src/test/java/com/coveo/pushapiclient/PlatformClientTest.java b/src/test/java/com/coveo/pushapiclient/PlatformClientTest.java index 8acf472b..b8b3d779 100644 --- a/src/test/java/com/coveo/pushapiclient/PlatformClientTest.java +++ b/src/test/java/com/coveo/pushapiclient/PlatformClientTest.java @@ -266,6 +266,39 @@ public void testPushFileContainerContent() throws IOException, InterruptedExcept assertAuthorizationHeader(); } + @Test + public void testOpenStream() throws IOException, InterruptedException { + client.openStream("my_source"); + verify(httpClient).send(argument.capture(), any(HttpResponse.BodyHandlers.ofString().getClass())); + + assertEquals("POST", argument.getValue().method()); + assertTrue(argument.getValue().uri().getPath().contains("the_org_id/sources/my_source/stream/open")); + assertApplicationJsonHeader(); + assertAuthorizationHeader(); + } + + @Test + public void testRequireStreamChunk() throws IOException, InterruptedException { + client.requireStreamChunk("my_source", "stream_id"); + verify(httpClient).send(argument.capture(), any(HttpResponse.BodyHandlers.ofString().getClass())); + + assertEquals("POST", argument.getValue().method()); + assertTrue(argument.getValue().uri().getPath().contains("the_org_id/sources/my_source/stream/stream_id/chunk")); + assertApplicationJsonHeader(); + assertAuthorizationHeader(); + } + + @Test + public void testCloseStream() throws IOException, InterruptedException { + client.closeStream("my_source", "stream_id"); + verify(httpClient).send(argument.capture(), any(HttpResponse.BodyHandlers.ofString().getClass())); + + assertEquals("POST", argument.getValue().method()); + assertTrue(argument.getValue().uri().getPath().contains("the_org_id/sources/my_source/stream/stream_id/close")); + assertApplicationJsonHeader(); + assertAuthorizationHeader(); + } + @Test public void testDeleteDocument() throws IOException, InterruptedException { client.deleteDocument("my_source", document().uri, true); diff --git a/src/test/java/com/coveo/pushapiclient/StreamServiceInternalTest.java b/src/test/java/com/coveo/pushapiclient/StreamServiceInternalTest.java new file mode 100644 index 00000000..5779b240 --- /dev/null +++ b/src/test/java/com/coveo/pushapiclient/StreamServiceInternalTest.java @@ -0,0 +1,96 @@ +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; +import static org.mockito.Mockito.when; + +public class StreamServiceInternalTest { + @Mock + private StreamEnabledSource source; + + @Mock + private DocumentUploadQueue queue; + + @Mock + private PlatformClient platformClient; + + @InjectMocks + private StreamServiceInternal service; + + @Mock + private HttpResponse httpResponse; + + private AutoCloseable closeable; + private DocumentBuilder documentA; + private DocumentBuilder documentB; + + @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"); + + closeable = MockitoAnnotations.openMocks(this); + + when(httpResponse.body()).thenReturn("{\"streamId\": \"stream-id\"}"); + when(platformClient.openStream("my-source-id")).thenReturn(httpResponse); + when(source.getId()).thenReturn("my-source-id"); + } + + @After + public void closeService() throws Exception { + closeable.close(); + } + + @Test + public void testAddShouldOpenANewStream() throws IOException, InterruptedException { + service.add(documentA); + service.add(documentB); + + verify(this.platformClient, times(1)).openStream("my-source-id"); + } + + @Test + public void testAddShouldAddDocumentToQueue() throws IOException, InterruptedException { + service.add(documentA); + service.add(documentB); + + verify(queue, times(1)).add(documentA); + verify(queue, times(1)).add(documentB); + } + + @Test + public void testCloseShouldCloseOpenStream() throws IOException, InterruptedException, NoOpenStreamException { + service.add(documentA); + service.close(); + + verify(platformClient, times(1)).closeStream("my-source-id", "stream-id"); + } + + @Test + public void testCloseShouldFlushBufferedDocuments() + throws IOException, InterruptedException, NoOpenStreamException { + service.add(documentA); + service.close(); + + verify(queue, times(1)).flush(); + } + + @Test(expected = NoOpenStreamException.class) + public void givenNoOpenStream_whenClose_thenShouldThrow() + throws IOException, InterruptedException, NoOpenStreamException { + service.close(); + } + +} \ No newline at end of file