From e0296ef852acb6117263c430b17f7b91d4547e03 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Fri, 19 May 2023 11:22:04 -0400 Subject: [PATCH 01/23] create `PushSource` and `catalogSource` classes --- .../java/com/coveo/pushapiclient/ApiUrl.java | 106 +++++++ .../com/coveo/pushapiclient/BaseSource.java | 18 ++ .../coveo/pushapiclient/CatalogSource.java | 143 +++++++++ .../pushapiclient/PushEnabledSource.java | 6 + .../com/coveo/pushapiclient/PushSource.java | 300 ++++++++++++++++++ .../java/com/coveo/pushapiclient/Source.java | 1 + .../pushapiclient/StreamEnabledSource.java | 6 + .../com/coveo/pushapiclient/ApiUrlTest.java | 94 ++++++ 8 files changed, 674 insertions(+) create mode 100644 src/main/java/com/coveo/pushapiclient/ApiUrl.java create mode 100644 src/main/java/com/coveo/pushapiclient/BaseSource.java create mode 100644 src/main/java/com/coveo/pushapiclient/CatalogSource.java create mode 100644 src/main/java/com/coveo/pushapiclient/PushEnabledSource.java create mode 100644 src/main/java/com/coveo/pushapiclient/PushSource.java create mode 100644 src/main/java/com/coveo/pushapiclient/StreamEnabledSource.java create mode 100644 src/test/java/com/coveo/pushapiclient/ApiUrlTest.java diff --git a/src/main/java/com/coveo/pushapiclient/ApiUrl.java b/src/main/java/com/coveo/pushapiclient/ApiUrl.java new file mode 100644 index 00000000..508e2c7c --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/ApiUrl.java @@ -0,0 +1,106 @@ +package com.coveo.pushapiclient; + +import java.net.MalformedURLException; +import java.net.URL; +import java.util.Arrays; +import java.util.EnumSet; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Private util class to extract dynamic parts from a Push API URL + * + * @See https://docs.coveo.com/en/1546#push-api-url + */ +class ApiUrl { + private final String organizationId; + private final String sourceId; + private final PlatformUrl platformUrl; + + public ApiUrl(URL sourceUrl) throws MalformedURLException { + List identifiers = this.extractIdentifiers(sourceUrl); + this.organizationId = identifiers.get(0); + this.sourceId = identifiers.get(1); + this.platformUrl = this.extractPlatformUrl(sourceUrl); + } + + public String getOrganizationId() { + return this.organizationId; + } + + public String getSourceId() { + return this.sourceId; + } + + public PlatformUrl getPlatformUrl() { + return this.platformUrl; + } + + private List extractIdentifiers(URL sourceUrl) throws MalformedURLException { + String host = sourceUrl.getPath(); + Pattern pattern = Pattern.compile("/push/v1/organizations/([^/]+)/sources/([^/]+)", Pattern.CASE_INSENSITIVE); + Matcher matcher = pattern.matcher(host); + + if (matcher.find()) { + String organizationId = matcher.group(1); + String sourceId = matcher.group(2); + return Arrays.asList(organizationId, sourceId); + } + + String errorMessage = this + .getErrorMessage("Unable to find organization and source ids from the provided API url"); + throw new MalformedURLException(errorMessage); + } + + private PlatformUrl extractPlatformUrl(URL sourceUrl) throws MalformedURLException { + String host = sourceUrl.getHost(); + Pattern pattern = Pattern.compile("api([a-z]*)([a-z-]*)\\.cloud\\.coveo\\.com", Pattern.CASE_INSENSITIVE); + Matcher matcher = pattern.matcher(host); + + if (matcher.find()) { + String extractedEnvironment = matcher.group(1); + String extractedRegion = matcher.group(2).replace("-", ""); + + Environment urlEnvironment = extractedEnvironment.isEmpty() + ? PlatformUrl.DEFAULT_ENVIRONMENT + : EnumSet.allOf(Environment.class) + .stream() + .filter(e -> e.getValue().equalsIgnoreCase(extractedEnvironment)) + .findFirst() + .orElseThrow(() -> new MalformedURLException( + String.format("Invalid platform environment '%s'", extractedEnvironment))); + + Region urlRegion = extractedRegion.isEmpty() + ? PlatformUrl.DEFAULT_REGION + : EnumSet.allOf(Region.class) + .stream() + .filter(r -> r.getValue().equalsIgnoreCase(extractedRegion)) + .findFirst() + .orElseThrow(() -> new MalformedURLException( + String.format("Invalid platform region '%s'", extractedRegion))); + + return new PlatformUrl(urlEnvironment, urlRegion); + + } + + String invalidHostMessage = this.getErrorMessage("Invalid API URL host"); + throw new MalformedURLException(invalidHostMessage); + } + + private String getErrorMessage(String reason) { + String newLine = System.getProperty("line.separator"); + String message = "The provided API URL is invalid"; + + message.concat(newLine).concat(reason); + + message + .concat(newLine) + .concat("For a Push Source, visit: https://docs.coveo.com/en/1546") + .concat(newLine) + .concat("For a Catalog Source, visit:https://docs.coveo.com/en/3295"); + + return message; + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/BaseSource.java b/src/main/java/com/coveo/pushapiclient/BaseSource.java new file mode 100644 index 00000000..26450a9a --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/BaseSource.java @@ -0,0 +1,18 @@ +package com.coveo.pushapiclient; + +public interface BaseSource { + /** + * Return an instance of {@link PlatformClient} + * + * @return + */ + PlatformClient getPlatformClient(); + + /** + * Returns the unique identifier of the source + * + * @return + */ + String getId(); + +} diff --git a/src/main/java/com/coveo/pushapiclient/CatalogSource.java b/src/main/java/com/coveo/pushapiclient/CatalogSource.java new file mode 100644 index 00000000..2c4a2c71 --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/CatalogSource.java @@ -0,0 +1,143 @@ +package com.coveo.pushapiclient; + +import java.net.MalformedURLException; +import java.net.URL; + +// TODO: LENS-851 - Make public when ready +class CatalogSource implements StreamEnabledSource { + private final PlatformClient platformClient; + private final String sourceId; + + /** + * Create a Catalog source instance from its + * Stream API URL + * + * @param apiKey The API key used for all operations regarding your source. + *

+ * Ensure your API key has the required privileges for the + * operation you will be performing + * * + *

+ * For more information about which privileges are required, + * see + * Privilege + * Reference. + * + * @param sourceUrl The URL available when you edit your source in the Coveo + * Administration Console. The URL should contain your + * ORGANIZATION_ID and SOURCE_ID, + * which are required parameters for all operations regarding + * your source. + *

+ * Some examples of valid source URLs: + * + * 

+     * https://api.cloud.coveo.com/push/v1/organizations/my-org-if/sources/my-source-id/stream/open
+     * https://api-eu.cloud.coveo.com/push/v1/organizations/my-org-if/sources/my-source-id/stream/open
+     *
+ * + * @throws MalformedURLException + */ + public CatalogSource(String apiKey, URL sourceUrl) throws MalformedURLException { + ApiUrl parser = new ApiUrl(sourceUrl); + PlatformUrl platformUrl = parser.getPlatformUrl(); + String organizationId = parser.getOrganizationId(); + this.sourceId = parser.getSourceId(); + this.platformClient = new PlatformClient(apiKey, organizationId, + platformUrl); + } + + /** + * Create a Catalog source instance from its + * Stream API URL + * + * @param apiKey The API key used for all operations regarding your + * source. + *

+ * Ensure your API key has the required privileges for the + * operation you will be performing + * * + *

+ * For more information about which privileges are + * required, + * see + * Privilege + * Reference. + * + * @param organizationId The unique identifier of your organization. + *

+ * The Organization Id can be retrieved in the URL of your + * Coveo organization. + * + * @param sourceId The unique identifier of the target Catalog source. + *

+ * The Source Id can be retrieved when you edit your + * source in the Coveo + * Administration Console + * + */ + public CatalogSource(String apiKey, String organizationId, String sourceId) { + PlatformUrl platformUrl = new PlatformUrl(PlatformUrl.DEFAULT_ENVIRONMENT, PlatformUrl.DEFAULT_REGION); + this.sourceId = sourceId; + this.platformClient = new PlatformClient(apiKey, organizationId, + platformUrl); + } + + /** + * Create a Catalog source instance from its + * Stream API URL + * + * @param apiKey The API key used for all operations regarding your + * source. + *

+ * Ensure your API key has the required privileges for the + * operation you will be performing + * * + *

+ * For more information about which privileges are + * required, + * see + * Privilege + * Reference. + * + * @param organizationId The unique identifier of your organization. + *

+ * The Organization Id can be retrieved in the URL of your + * Coveo organization. + * + * @param sourceId The unique identifier of the target Catalog source. + *

+ * The Source Id can be retrieved when you edit your + * source in the Coveo + * Administration Console + * + * @param platformUrl The object containing additional information on the + * URL endpoint. + * You can use the {@link PlatformUrl} when your + * organization is located in a non-default Coveo + * environement and/or region. + * + */ + public CatalogSource(String apiKey, String organizationId, String sourceId, PlatformUrl platformUrl) { + this.sourceId = sourceId; + this.platformClient = new PlatformClient(apiKey, organizationId, + platformUrl); + } + + @Override + public String getId() { + return this.sourceId; + } + + @Override + public PlatformClient getPlatformClient() { + return this.platformClient; + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/PushEnabledSource.java b/src/main/java/com/coveo/pushapiclient/PushEnabledSource.java new file mode 100644 index 00000000..b90b5f1e --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/PushEnabledSource.java @@ -0,0 +1,6 @@ +package com.coveo.pushapiclient; + +// Marker Interface +public interface PushEnabledSource extends BaseSource { + +} diff --git a/src/main/java/com/coveo/pushapiclient/PushSource.java b/src/main/java/com/coveo/pushapiclient/PushSource.java new file mode 100644 index 00000000..92c777bf --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/PushSource.java @@ -0,0 +1,300 @@ +package com.coveo.pushapiclient; + +import com.google.gson.Gson; + +import java.io.IOException; +import java.net.MalformedURLException; +import java.net.URL; +import java.net.http.HttpResponse; + +// TODO: LENS-851 - Make public when ready +class PushSource implements PushEnabledSource { + private final PlatformClient platformClient; + private final String sourceId; + + @Override + public PlatformClient getPlatformClient() { + return this.platformClient; + } + + @Override + public String getId() { + return this.sourceId; + } + + /** + * Create a Push source instance from its + * Push API URL + * + * @param apiKey The API key used for all operations regarding your source. + *

+ * Ensure your API key has the required privileges for the + * operation you will be performing + * * + *

+ * For more information about which privileges are required, + * see + * Privilege + * Reference. + * + * @param sourceUrl The URL available when you edit your source in the Coveo + * Administration Console. The URL should contain your + * ORGANIZATION_ID and SOURCE_ID, + * which are required parameters for all operations regarding + * your source. + *

+ * Some examples of valid source URLs: + * + * 

+     * https://api.cloud.coveo.com/push/v1/organizations/my-org-if/sources/my-source-id/documents
+     * https://api-eu.cloud.coveo.com/push/v1/organizations/my-org-if/sources/my-source-id/documents
+     *
+ * + * @throws MalformedURLException + */ + public PushSource(String apiKey, URL sourceUrl) throws MalformedURLException { + ApiUrl parser = new ApiUrl(sourceUrl); + PlatformUrl platformUrl = parser.getPlatformUrl(); + String organizationId = parser.getOrganizationId(); + this.sourceId = parser.getSourceId(); + this.platformClient = new PlatformClient(apiKey, organizationId, + platformUrl); + } + + /** + * Create a Push source instance from its + * Stream API URL + * + * @param apiKey The API key used for all operations regarding your + * source. + *

+ * Ensure your API key has the required privileges for the + * operation you will be performing + * * + *

+ * For more information about which privileges are + * required, + * see + * Privilege + * Reference. + * + * @param organizationId The unique identifier of your organization. + *

+ * The Organization Id can be retrieved in the URL of your + * Coveo organization. + * + * @param sourceId The unique identifier of the target Push source. + *

+ * The Source Id can be retrieved when you edit your + * source in the Coveo + * Administration Console + * + */ + public PushSource(String apiKey, String organizationId, String sourceId) { + PlatformUrl platformUrl = new PlatformUrl(PlatformUrl.DEFAULT_ENVIRONMENT, PlatformUrl.DEFAULT_REGION); + this.sourceId = sourceId; + this.platformClient = new PlatformClient(apiKey, organizationId, + platformUrl); + } + + /** + * Create a Push source instance from its + * Stream API URL + * + * @param apiKey The API key used for all operations regarding your + * source. + *

+ * Ensure your API key has the required privileges for the + * operation you will be performing + * * + *

+ * For more information about which privileges are + * required, + * see + * Privilege + * Reference. + * + * @param organizationId The unique identifier of your organization. + *

+ * The Organization Id can be retrieved in the URL of your + * Coveo organization. + * + * @param sourceId The unique identifier of the target Push source. + *

+ * The Source Id can be retrieved when you edit your + * source in the Coveo + * Administration Console + * + * @param platformUrl The object containing additional information on the + * URL endpoint. + * You can use the {@link PlatformUrl} when your + * organization is located in a non-default Coveo + * environement and/or region. + * + */ + public PushSource(String apiKey, String organizationId, String sourceId, PlatformUrl platformUrl) { + this.sourceId = sourceId; + this.platformClient = new PlatformClient(apiKey, organizationId, + platformUrl); + } + + /** + * Create or update a security identity. See [Adding a Single Security + * Identity](https://docs.coveo.com/en/167) and [Security Identity + * Models](https://docs.coveo.com/en/139). + * + * @param securityProviderId + * @param securityIdentityModel + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse createOrUpdateSecurityIdentity(String securityProviderId, + SecurityIdentityModel securityIdentityModel) throws IOException, InterruptedException { + return this.platformClient.createOrUpdateSecurityIdentity(securityProviderId, securityIdentityModel); + } + + /** + * Create or update a security identity alias. See [Adding a Single + * Alias](https://docs.coveo.com/en/142) and [User Alias Definition + * Examples](https://docs.coveo.com/en/46). + * + * @param securityProviderId + * @param securityIdentityAliasModel + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse createOrUpdateSecurityIdentityAlias(String securityProviderId, + SecurityIdentityAliasModel securityIdentityAliasModel) throws IOException, InterruptedException { + return this.platformClient.createOrUpdateSecurityIdentityAlias(securityProviderId, securityIdentityAliasModel); + } + + /** + * Delete a security identity. See [Disabling a Single Security + * Identity](https://docs.coveo.com/en/84). + * + * @param securityProviderId + * @param securityIdentityDelete + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse deleteSecurityIdentity(String securityProviderId, + SecurityIdentityDelete securityIdentityDelete) throws IOException, InterruptedException { + return this.platformClient.deleteSecurityIdentity(securityProviderId, securityIdentityDelete); + } + + /** + * Update the status of a Push source. See [Updating the Status of a Push + * Source](https://docs.coveo.com/en/35). + * + * @param status + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse updateSourceStatus(PushAPIStatus status) + throws IOException, InterruptedException { + return this.platformClient.updateSourceStatus(this.sourceId, status); + } + + /** + * Delete old security identities. See [Disabling Old Security + * Identities](https://docs.coveo.com/en/33). + * + * @param securityProviderId + * @param batchDelete + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse deleteOldSecurityIdentities(String securityProviderId, + SecurityIdentityDeleteOptions batchDelete) throws IOException, InterruptedException { + return this.platformClient.deleteOldSecurityIdentities(securityProviderId, batchDelete); + } + + /** + * Manage batches of security identities. See [Manage Batches of Security + * Identities](https://docs.coveo.com/en/55). + * + * @param securityProviderId + * @param batchConfig + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse manageSecurityIdentities(String securityProviderId, + SecurityIdentityBatchConfig batchConfig) throws IOException, InterruptedException { + return this.platformClient.manageSecurityIdentities(securityProviderId, batchConfig); + } + + /** + * Manages pushing batches of Security Identities to a File Container, then into + * Coveo. See [Manage Batches of Security + * Identities](https://docs.coveo.com/en/55) + * + * @param securityProviderId + * @param batchIdentity + * @return + * @throws IOException + * @throws InterruptedException + */ + public SecurityIdentityBatchResponse batchUpdateSecurityIdentities(String securityProviderId, + BatchIdentity batchIdentity) throws IOException, InterruptedException { + SecurityIdentityBatchResponse securityIdentityBatchResponse = new SecurityIdentityBatchResponse(); + HttpResponse resFileContainer = this.platformClient.createFileContainer(); + FileContainer fileContainer = new Gson().fromJson(resFileContainer.body(), FileContainer.class); + String batchIdJson = new Gson().toJson(batchIdentity.marshal()); + securityIdentityBatchResponse.s3Response = this.platformClient.uploadContentToFileContainer(fileContainer, + batchIdJson); + if (securityIdentityBatchResponse.s3Response.statusCode() >= 200 + && securityIdentityBatchResponse.s3Response.statusCode() <= 299) { // maybe just 200 or 202 + SecurityIdentityBatchConfig batchConfig = new SecurityIdentityBatchConfig(fileContainer.fileId, 0l); + securityIdentityBatchResponse.batchResponse = this.manageSecurityIdentities(securityProviderId, + batchConfig); + } + return securityIdentityBatchResponse; + } + + /** + * Adds or updates an individual item in a push source. See [Adding a Single + * Item in a Push Source](https://docs.coveo.com/en/133). + * + * @param docBuilder + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse addOrUpdateDocument(DocumentBuilder docBuilder) + throws IOException, InterruptedException { + CompressionType compressionType = docBuilder.getDocument().compressedBinaryData != null + ? docBuilder.getDocument().compressedBinaryData.getCompressionType() + : CompressionType.UNCOMPRESSED; + return this.platformClient.pushDocument(this.sourceId, docBuilder.marshal(), docBuilder.getDocument().uri, + compressionType); + } + + /** + * Deletes a specific item from a Push source. Optionally, the child items of + * that item can also be deleted. See [Deleting an Item in a Push + * Source](https://docs.coveo.com/en/171). + * + * @param documentId + * @param deleteChildren + * @return + * @throws IOException + * @throws InterruptedException + */ + public HttpResponse deleteDocument(String documentId, Boolean deleteChildren) + throws IOException, InterruptedException { + return this.platformClient.deleteDocument(this.sourceId, documentId, deleteChildren); + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/Source.java b/src/main/java/com/coveo/pushapiclient/Source.java index 3b09a562..2288d944 100644 --- a/src/main/java/com/coveo/pushapiclient/Source.java +++ b/src/main/java/com/coveo/pushapiclient/Source.java @@ -5,6 +5,7 @@ import java.io.IOException; import java.net.http.HttpResponse; +// TODO: LENS-844 - Deprecate class public class Source { PlatformClient platformClient; diff --git a/src/main/java/com/coveo/pushapiclient/StreamEnabledSource.java b/src/main/java/com/coveo/pushapiclient/StreamEnabledSource.java new file mode 100644 index 00000000..b04912bf --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/StreamEnabledSource.java @@ -0,0 +1,6 @@ +package com.coveo.pushapiclient; + +// Marker Interface +public interface StreamEnabledSource extends BaseSource { + +} diff --git a/src/test/java/com/coveo/pushapiclient/ApiUrlTest.java b/src/test/java/com/coveo/pushapiclient/ApiUrlTest.java new file mode 100644 index 00000000..1cdda355 --- /dev/null +++ b/src/test/java/com/coveo/pushapiclient/ApiUrlTest.java @@ -0,0 +1,94 @@ +package com.coveo.pushapiclient; + +import static org.junit.Assert.assertEquals; + +import java.net.MalformedURLException; +import java.net.URL; + +import org.junit.Before; +import org.junit.Test; + +public class ApiUrlTest { + + private ApiUrl defaultUrl; + private ApiUrl regionOnlyUrl; + private ApiUrl regionOnlyUrlCaseInsensitive; + private ApiUrl environmentOnlyUrl; + private ApiUrl environmentAndRegionUrl; + private ApiUrl streamURL; + + @Before + public void setUp() throws MalformedURLException { + defaultUrl = new ApiUrl( + new URL("https://api.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + regionOnlyUrl = new ApiUrl( + new URL("https://api-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + regionOnlyUrlCaseInsensitive = new ApiUrl( + new URL("https://api-EU.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + environmentOnlyUrl = new ApiUrl( + new URL("https://apidev.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + environmentAndRegionUrl = new ApiUrl( + new URL("https://apidev-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + streamURL = new ApiUrl( + new URL("https://apidev-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/stream/open")); + + } + + @Test + public void testSourceId() { + assertEquals(defaultUrl.getSourceId(), "my-source-id"); + assertEquals(regionOnlyUrl.getSourceId(), "my-source-id"); + assertEquals(regionOnlyUrlCaseInsensitive.getSourceId(), "my-source-id"); + assertEquals(environmentOnlyUrl.getSourceId(), "my-source-id"); + assertEquals(environmentAndRegionUrl.getSourceId(), "my-source-id"); + assertEquals(streamURL.getSourceId(), "my-source-id"); + } + + @Test + public void testOrganizationId() { + assertEquals(defaultUrl.getOrganizationId(), "my-org-id"); + assertEquals(regionOnlyUrl.getOrganizationId(), "my-org-id"); + assertEquals(regionOnlyUrlCaseInsensitive.getOrganizationId(), "my-org-id"); + assertEquals(environmentOnlyUrl.getOrganizationId(), "my-org-id"); + assertEquals(environmentAndRegionUrl.getOrganizationId(), "my-org-id"); + assertEquals(streamURL.getOrganizationId(), "my-org-id"); + } + + @Test + public void testPlatformUrl() { + assertEquals(defaultUrl.getPlatformUrl().getApiUrl(), "https://api.cloud.coveo.com"); + assertEquals(regionOnlyUrl.getPlatformUrl().getApiUrl(), "https://api-au.cloud.coveo.com"); + assertEquals(regionOnlyUrlCaseInsensitive.getPlatformUrl().getApiUrl(), "https://api-eu.cloud.coveo.com"); + assertEquals(environmentOnlyUrl.getPlatformUrl().getApiUrl(), "https://apidev.cloud.coveo.com"); + assertEquals(environmentAndRegionUrl.getPlatformUrl().getApiUrl(), "https://apidev-au.cloud.coveo.com"); + assertEquals(streamURL.getPlatformUrl().getApiUrl(), "https://apidev-au.cloud.coveo.com"); + } + + @Test(expected = MalformedURLException.class) + public void testInvalidEnvironementUrl() throws MalformedURLException { + defaultUrl = new ApiUrl( + new URL("https://apifoo.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + + } + + @Test(expected = MalformedURLException.class) + public void testInvalidRegionUrl() throws MalformedURLException { + defaultUrl = new ApiUrl( + new URL("https://api-bar.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + + } + + @Test(expected = MalformedURLException.class) + public void testInvalidPathUrl() throws MalformedURLException { + defaultUrl = new ApiUrl( + new URL("https://api.cloud.coveo.com/push/v1/organizations/my-org-id/providers/provider-id/mappings")); + + } + + @Test(expected = MalformedURLException.class) + public void testInvalidHostUrl() throws MalformedURLException { + defaultUrl = new ApiUrl( + new URL("https://platform.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + + } +} From 419d1773e90cb0ff9364d6125db08f44126bf8f4 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Fri, 19 May 2023 13:58:57 -0400 Subject: [PATCH 02/23] add boilerplate for stream service --- .../pushapiclient/DocumentUploadQueue.java | 25 ++++++++ .../coveo/pushapiclient/PlatformClient.java | 20 +++++-- .../coveo/pushapiclient/StreamService.java | 60 +++++++++++++++++++ .../coveo/pushapiclient/UpdloadStrategy.java | 9 +++ 4 files changed, 110 insertions(+), 4 deletions(-) create mode 100644 src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java create mode 100644 src/main/java/com/coveo/pushapiclient/StreamService.java create mode 100644 src/main/java/com/coveo/pushapiclient/UpdloadStrategy.java 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..d3daf06f --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -0,0 +1,25 @@ +package com.coveo.pushapiclient; + +import java.io.IOException; + +public class DocumentUploadQueue { + private final UpdloadStrategy uploader; + + public DocumentUploadQueue(UpdloadStrategy uploader) { + this.uploader = uploader; + } + + public void flush() throws IOException, InterruptedException { + } + + public void add(DocumentBuilder documentToAdd, DeleteDocument documentToDelete) + throws IOException, InterruptedException { + } + + public void add(DocumentBuilder document) throws IOException, InterruptedException { + // Once batch is ready, send it like: + // this.uploader.apply(batchUpdate); + throw new UnsupportedOperationException("Unimplemented method"); + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/PlatformClient.java b/src/main/java/com/coveo/pushapiclient/PlatformClient.java index f54aca7b..12369003 100644 --- a/src/main/java/com/coveo/pushapiclient/PlatformClient.java +++ b/src/main/java/com/coveo/pushapiclient/PlatformClient.java @@ -91,10 +91,10 @@ public HttpResponse createSource(String name, SourceVisibility sourceVis String[] headers = this.getHeaders(this.getAuthorizationHeader(), this.getContentTypeApplicationJSONHeader()); String json = this.toJSON(new HashMap<>() {{ - put("sourceType", "PUSH"); - put("pushEnabled", true); - put("name", name); - put("sourceVisibility", sourceVisibility); + put("sourceType", "PUSH"); + put("pushEnabled", true); + put("name", name); + put("sourceVisibility", sourceVisibility); }}); HttpRequest request = HttpRequest.newBuilder() @@ -282,6 +282,18 @@ public HttpResponse deleteDocument(String sourceId, String documentId, B return this.httpClient.send(request, HttpResponse.BodyHandlers.ofString()); } + public HttpResponse openStream() throws IOException, InterruptedException { + throw new UnsupportedOperationException("Unimplemented method"); + } + + public HttpResponse closeStream(String streamId) throws IOException, InterruptedException { + throw new UnsupportedOperationException("Unimplemented method"); + } + + public HttpResponse requireStreamChunk() throws IOException, InterruptedException { + throw new UnsupportedOperationException("Unimplemented method"); + } + /** * Create a file container. See [Creating a File Container](https://docs.coveo.com/en/43). * 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..ac4697ac --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -0,0 +1,60 @@ +package com.coveo.pushapiclient; + +import java.io.IOException; +import java.net.http.HttpResponse; +import com.google.gson.Gson; + +public class StreamService { + private StreamEnabledSource source; + private String streamId; + private DocumentUploadQueue queue; + + public StreamService(StreamEnabledSource source) { + UpdloadStrategy uploader = this.getUploadStrategy(); + this.source = source; + this.queue = new DocumentUploadQueue(uploader); + } + + /** + * Pushes document to the source. + * If multiple documents are added, the class will ensure documents are + * automatically batched into chunks that do not exceed API limit size. + * + * @param document + * @throws InterruptedException + * @throws IOException + */ + 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 { + if (this.streamId == null) { + throw new java.lang.UnsupportedOperationException("TODO: custom error: No stream was open yet"); + } + queue.flush(); + PlatformClient platformClient = source.getPlatformClient(); + return platformClient.closeStream(this.streamId); + } + + private UpdloadStrategy getUploadStrategy() { + return (batchUpdate) -> { + PlatformClient platformClient = source.getPlatformClient(); + HttpResponse resFileContainer = platformClient.requireStreamChunk(); + FileContainer fileContainer = new Gson().fromJson(resFileContainer.body(), FileContainer.class); + String batchUpdateJson = new Gson().toJson(batchUpdate.marshal()); + return platformClient.uploadContentToFileContainer(fileContainer, + batchUpdateJson); + + }; + } + + private String getStreamId() throws IOException, InterruptedException { + HttpResponse response = this.source.getPlatformClient().openStream(); + return "TODO: get streamID from response"; + } + +} diff --git a/src/main/java/com/coveo/pushapiclient/UpdloadStrategy.java b/src/main/java/com/coveo/pushapiclient/UpdloadStrategy.java new file mode 100644 index 00000000..f5bb812e --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/UpdloadStrategy.java @@ -0,0 +1,9 @@ +package com.coveo.pushapiclient; + +import java.io.IOException; +import java.net.http.HttpResponse; + +@FunctionalInterface +public interface UpdloadStrategy { + HttpResponse apply(BatchUpdate batchUpdate) throws IOException, InterruptedException; +} From 1c5adec54710f868995090a04c0f4965a3b1e440 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Fri, 19 May 2023 14:29:47 -0400 Subject: [PATCH 03/23] revert master merge --- .github/workflows/codeql.yml | 15 --------------- 1 file changed, 15 deletions(-) delete mode 100644 .github/workflows/codeql.yml diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml deleted file mode 100644 index dea2cef6..00000000 --- a/.github/workflows/codeql.yml +++ /dev/null @@ -1,15 +0,0 @@ -name: 'CodeQL' - -on: - push: - branches: ['main'] - - pull_request: - branches: ['main'] - - schedule: - - cron: '29 3 * * 6' # Runs at 03:29, only on Saturday. - -jobs: - analyze-java: - uses: coveo/actions/.github/workflows/java-maven-openjdk11-codeql.yml@main From f5a25576067794a301d711d1345ac7fd851837d3 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Fri, 19 May 2023 15:43:23 -0400 Subject: [PATCH 04/23] add required POST calls --- .../coveo/pushapiclient/PlatformClient.java | 39 ++++++++++++++++--- 1 file changed, 33 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/PlatformClient.java b/src/main/java/com/coveo/pushapiclient/PlatformClient.java index 12369003..0659a1ff 100644 --- a/src/main/java/com/coveo/pushapiclient/PlatformClient.java +++ b/src/main/java/com/coveo/pushapiclient/PlatformClient.java @@ -282,16 +282,43 @@ public HttpResponse deleteDocument(String sourceId, String documentId, B return this.httpClient.send(request, HttpResponse.BodyHandlers.ofString()); } - public HttpResponse openStream() throws IOException, InterruptedException { - throw new UnsupportedOperationException("Unimplemented method"); + public HttpResponse openStream(String sourceId) throws IOException, InterruptedException { + String[] headers = this.getHeaders(this.getAuthorizationHeader(), this.getContentTypeApplicationJSONHeader()); + URI uri = URI.create(this.getBasePushURL() + String.format("/sources/%s/stream/open", sourceId)); + + 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 streamId) throws IOException, InterruptedException { - throw new UnsupportedOperationException("Unimplemented method"); + 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() throws IOException, InterruptedException { - throw new UnsupportedOperationException("Unimplemented method"); + 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()); } /** From 673b76a5712168f2b686e69f2ea64b2c48377e17 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Fri, 19 May 2023 15:44:29 -0400 Subject: [PATCH 05/23] update streamService --- src/main/java/com/coveo/pushapiclient/StreamService.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/StreamService.java b/src/main/java/com/coveo/pushapiclient/StreamService.java index ac4697ac..60cf6af9 100644 --- a/src/main/java/com/coveo/pushapiclient/StreamService.java +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -37,13 +37,15 @@ public HttpResponse close() throws IOException, InterruptedException { } queue.flush(); PlatformClient platformClient = source.getPlatformClient(); - return platformClient.closeStream(this.streamId); + String sourceId = this.source.getId(); + return platformClient.closeStream(sourceId, this.streamId); } private UpdloadStrategy getUploadStrategy() { return (batchUpdate) -> { + String sourceId = this.source.getId(); PlatformClient platformClient = source.getPlatformClient(); - HttpResponse resFileContainer = platformClient.requireStreamChunk(); + HttpResponse resFileContainer = platformClient.requireStreamChunk(sourceId, this.streamId); FileContainer fileContainer = new Gson().fromJson(resFileContainer.body(), FileContainer.class); String batchUpdateJson = new Gson().toJson(batchUpdate.marshal()); return platformClient.uploadContentToFileContainer(fileContainer, @@ -53,7 +55,8 @@ private UpdloadStrategy getUploadStrategy() { } private String getStreamId() throws IOException, InterruptedException { - HttpResponse response = this.source.getPlatformClient().openStream(); + String sourceId = this.source.getId(); + HttpResponse response = this.source.getPlatformClient().openStream(sourceId); return "TODO: get streamID from response"; } From 5ef67724a2ec227b7137fa64c69fddb69e468aa1 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 23 May 2023 10:24:38 -0400 Subject: [PATCH 06/23] apply ApiUrl corrections --- .../java/com/coveo/pushapiclient/ApiUrl.java | 7 +- .../com/coveo/pushapiclient/ApiUrlTest.java | 87 +++++++++---------- 2 files changed, 44 insertions(+), 50 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/ApiUrl.java b/src/main/java/com/coveo/pushapiclient/ApiUrl.java index 508e2c7c..be761eb5 100644 --- a/src/main/java/com/coveo/pushapiclient/ApiUrl.java +++ b/src/main/java/com/coveo/pushapiclient/ApiUrl.java @@ -9,9 +9,10 @@ import java.util.regex.Pattern; /** - * Private util class to extract dynamic parts from a Push API URL + * Private util class to extract dynamic parts from a API URL * * @See https://docs.coveo.com/en/1546#push-api-url + * https://docs.coveo.com/en/3295#stream-api-url */ class ApiUrl { private final String organizationId; @@ -39,7 +40,7 @@ public PlatformUrl getPlatformUrl() { private List extractIdentifiers(URL sourceUrl) throws MalformedURLException { String host = sourceUrl.getPath(); - Pattern pattern = Pattern.compile("/push/v1/organizations/([^/]+)/sources/([^/]+)", Pattern.CASE_INSENSITIVE); + Pattern pattern = Pattern.compile("/push/v1/organizations/([^/]+)/sources/([^/]+)"); Matcher matcher = pattern.matcher(host); if (matcher.find()) { @@ -55,7 +56,7 @@ private List extractIdentifiers(URL sourceUrl) throws MalformedURLExcept private PlatformUrl extractPlatformUrl(URL sourceUrl) throws MalformedURLException { String host = sourceUrl.getHost(); - Pattern pattern = Pattern.compile("api([a-z]*)([a-z-]*)\\.cloud\\.coveo\\.com", Pattern.CASE_INSENSITIVE); + Pattern pattern = Pattern.compile("api([a-z]*)([a-z-]*)\\.cloud\\.coveo\\.com"); Matcher matcher = pattern.matcher(host); if (matcher.find()) { diff --git a/src/test/java/com/coveo/pushapiclient/ApiUrlTest.java b/src/test/java/com/coveo/pushapiclient/ApiUrlTest.java index 1cdda355..964dcb55 100644 --- a/src/test/java/com/coveo/pushapiclient/ApiUrlTest.java +++ b/src/test/java/com/coveo/pushapiclient/ApiUrlTest.java @@ -5,89 +5,82 @@ import java.net.MalformedURLException; import java.net.URL; -import org.junit.Before; import org.junit.Test; public class ApiUrlTest { - private ApiUrl defaultUrl; - private ApiUrl regionOnlyUrl; - private ApiUrl regionOnlyUrlCaseInsensitive; - private ApiUrl environmentOnlyUrl; - private ApiUrl environmentAndRegionUrl; - private ApiUrl streamURL; - - @Before - public void setUp() throws MalformedURLException { - defaultUrl = new ApiUrl( + @Test + public void testSourceId() throws MalformedURLException { + ApiUrl url = new ApiUrl( new URL("https://api.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); - regionOnlyUrl = new ApiUrl( - new URL("https://api-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); - regionOnlyUrlCaseInsensitive = new ApiUrl( - new URL("https://api-EU.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); - environmentOnlyUrl = new ApiUrl( - new URL("https://apidev.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); - environmentAndRegionUrl = new ApiUrl( - new URL("https://apidev-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); - streamURL = new ApiUrl( - new URL("https://apidev-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/stream/open")); - + assertEquals(url.getSourceId(), "my-source-id"); } @Test - public void testSourceId() { - assertEquals(defaultUrl.getSourceId(), "my-source-id"); - assertEquals(regionOnlyUrl.getSourceId(), "my-source-id"); - assertEquals(regionOnlyUrlCaseInsensitive.getSourceId(), "my-source-id"); - assertEquals(environmentOnlyUrl.getSourceId(), "my-source-id"); - assertEquals(environmentAndRegionUrl.getSourceId(), "my-source-id"); - assertEquals(streamURL.getSourceId(), "my-source-id"); + public void testOrganizationId() throws MalformedURLException { + ApiUrl url = new ApiUrl( + new URL("https://api.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + assertEquals(url.getOrganizationId(), "my-org-id"); } @Test - public void testOrganizationId() { - assertEquals(defaultUrl.getOrganizationId(), "my-org-id"); - assertEquals(regionOnlyUrl.getOrganizationId(), "my-org-id"); - assertEquals(regionOnlyUrlCaseInsensitive.getOrganizationId(), "my-org-id"); - assertEquals(environmentOnlyUrl.getOrganizationId(), "my-org-id"); - assertEquals(environmentAndRegionUrl.getOrganizationId(), "my-org-id"); - assertEquals(streamURL.getOrganizationId(), "my-org-id"); - } + public void testPlatformUrl() throws MalformedURLException { + ApiUrl defaultUrl = new ApiUrl( + new URL("https://api.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + ApiUrl regionOnlyUrl = new ApiUrl( + new URL("https://api-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + ApiUrl environmentOnlyUrl = new ApiUrl( + new URL("https://apidev.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + ApiUrl environmentAndRegionUrl = new ApiUrl( + new URL("https://apidev-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); - @Test - public void testPlatformUrl() { assertEquals(defaultUrl.getPlatformUrl().getApiUrl(), "https://api.cloud.coveo.com"); assertEquals(regionOnlyUrl.getPlatformUrl().getApiUrl(), "https://api-au.cloud.coveo.com"); - assertEquals(regionOnlyUrlCaseInsensitive.getPlatformUrl().getApiUrl(), "https://api-eu.cloud.coveo.com"); assertEquals(environmentOnlyUrl.getPlatformUrl().getApiUrl(), "https://apidev.cloud.coveo.com"); assertEquals(environmentAndRegionUrl.getPlatformUrl().getApiUrl(), "https://apidev-au.cloud.coveo.com"); - assertEquals(streamURL.getPlatformUrl().getApiUrl(), "https://apidev-au.cloud.coveo.com"); } + @Test + public void testStreamApiUrl() throws MalformedURLException { + ApiUrl url = new ApiUrl( + new URL("https://apidev-au.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/stream/open")); + + assertEquals(url.getPlatformUrl().getApiUrl(), "https://apidev-au.cloud.coveo.com"); + assertEquals(url.getOrganizationId(), "my-org-id"); + assertEquals(url.getSourceId(), "my-source-id"); + } + + @Test(expected = MalformedURLException.class) + public void testInvalidEnvironmentUrl() throws MalformedURLException { + new ApiUrl( + new URL("https://apifoo.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + + } @Test(expected = MalformedURLException.class) - public void testInvalidEnvironementUrl() throws MalformedURLException { - defaultUrl = new ApiUrl( + public void testInvalidUrl() throws MalformedURLException { + new ApiUrl( new URL("https://apifoo.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); } @Test(expected = MalformedURLException.class) public void testInvalidRegionUrl() throws MalformedURLException { - defaultUrl = new ApiUrl( - new URL("https://api-bar.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); + new ApiUrl( + new URL( + "https://api-bar.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); } @Test(expected = MalformedURLException.class) public void testInvalidPathUrl() throws MalformedURLException { - defaultUrl = new ApiUrl( + new ApiUrl( new URL("https://api.cloud.coveo.com/push/v1/organizations/my-org-id/providers/provider-id/mappings")); } @Test(expected = MalformedURLException.class) public void testInvalidHostUrl() throws MalformedURLException { - defaultUrl = new ApiUrl( + new ApiUrl( new URL("https://platform.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/documents")); } From 8e1e3e7d1949d8d3ca67a9f6087af2124a1a1c94 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 23 May 2023 14:35:15 -0400 Subject: [PATCH 07/23] document StreamService Update src/main/java/com/coveo/pushapiclient/ApiUrl.java Co-authored-by: Benjamin Taillon <54454747+btaillon@users.noreply.github.com> add new getter to source classes feat: create `PushSource` and `catalogSource` classes (#30) https://coveord.atlassian.net/browse/LENS-839 update StreamService add platformClient tests add PowerMockito dependencies remove unused import remove unecessary method downgrade build to Java 11 revert codeql deletion lint lint --- .github/workflows/build.yml | 4 +- .github/workflows/codeql.yml | 15 +++ pom.xml | 16 ++- .../java/com/coveo/pushapiclient/ApiUrl.java | 15 +++ .../com/coveo/pushapiclient/BaseSource.java | 20 +++- .../coveo/pushapiclient/CatalogSource.java | 51 +++++---- .../pushapiclient/DocumentUploadQueue.java | 12 +- .../coveo/pushapiclient/PlatformClient.java | 8 +- .../com/coveo/pushapiclient/PushSource.java | 66 ++++++----- .../coveo/pushapiclient/StreamResponse.java | 11 ++ .../coveo/pushapiclient/StreamService.java | 108 +++++++++++++++--- .../exceptions/NoOpenStreamException.java | 7 ++ .../pushapiclient/PlatformClientTest.java | 33 ++++++ .../pushapiclient/StreamServiceTest.java | 103 +++++++++++++++++ 14 files changed, 387 insertions(+), 82 deletions(-) create mode 100644 .github/workflows/codeql.yml create mode 100644 src/main/java/com/coveo/pushapiclient/StreamResponse.java create mode 100644 src/main/java/com/coveo/pushapiclient/exceptions/NoOpenStreamException.java create mode 100644 src/test/java/com/coveo/pushapiclient/StreamServiceTest.java 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 new file mode 100644 index 00000000..dea2cef6 --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,15 @@ +name: 'CodeQL' + +on: + push: + branches: ['main'] + + pull_request: + branches: ['main'] + + schedule: + - cron: '29 3 * * 6' # Runs at 03:29, only on Saturday. + +jobs: + analyze-java: + uses: coveo/actions/.github/workflows/java-maven-openjdk11-codeql.yml@main diff --git a/pom.xml b/pom.xml index 142e7175..7580afeb 100644 --- a/pom.xml +++ b/pom.xml @@ -137,12 +137,26 @@ joda-time 2.10.10 + + + org.powermock + powermock-module-junit4 + 2.0.9 + test + + + org.powermock + powermock-api-mockito2 + 2.0.9 + test + org.mockito mockito-core - 4.7.0 + 2.23.0 test + junit junit diff --git a/src/main/java/com/coveo/pushapiclient/ApiUrl.java b/src/main/java/com/coveo/pushapiclient/ApiUrl.java index be761eb5..b416576d 100644 --- a/src/main/java/com/coveo/pushapiclient/ApiUrl.java +++ b/src/main/java/com/coveo/pushapiclient/ApiUrl.java @@ -10,6 +10,7 @@ /** * Private util class to extract dynamic parts from a API URL + * Handles extraction of identifiers and platform URL from a source URL. * * @See https://docs.coveo.com/en/1546#push-api-url * https://docs.coveo.com/en/3295#stream-api-url @@ -18,14 +19,28 @@ class ApiUrl { private final String organizationId; private final String sourceId; private final PlatformUrl platformUrl; + private final String sourceUrl; public ApiUrl(URL sourceUrl) throws MalformedURLException { List identifiers = this.extractIdentifiers(sourceUrl); this.organizationId = identifiers.get(0); this.sourceId = identifiers.get(1); + this.sourceUrl = sourceUrl.toString(); this.platformUrl = this.extractPlatformUrl(sourceUrl); } + public ApiUrl(String organizationId, String sourceId, PlatformUrl platformUrl) { + this.organizationId = organizationId; + this.sourceId = sourceId; + this.platformUrl = platformUrl; + this.sourceUrl = String.format("https://api.cloud.coveo.com/push/v1/organizations/%s/sources/%s", + this.organizationId, this.sourceId); + } + + public String getUrl() { + return this.sourceUrl; + } + public String getOrganizationId() { return this.organizationId; } diff --git a/src/main/java/com/coveo/pushapiclient/BaseSource.java b/src/main/java/com/coveo/pushapiclient/BaseSource.java index 26450a9a..a3edd80c 100644 --- a/src/main/java/com/coveo/pushapiclient/BaseSource.java +++ b/src/main/java/com/coveo/pushapiclient/BaseSource.java @@ -2,14 +2,28 @@ public interface BaseSource { /** - * Return an instance of {@link PlatformClient} + * Returns the API key used for all operations regarding your source. * * @return */ - PlatformClient getPlatformClient(); + String getApiKey(); /** - * Returns the unique identifier of the source + * Returns the {@link PlatformUrl} object associated to the source. + * + * @return + */ + PlatformUrl getPlatformUrl(); + + /** + * The unique identifier of your organization. + * + * @return + */ + String getOrganizationId(); + + /** + * The unique identifier of your source. * * @return */ diff --git a/src/main/java/com/coveo/pushapiclient/CatalogSource.java b/src/main/java/com/coveo/pushapiclient/CatalogSource.java index 2c4a2c71..7fc9b08a 100644 --- a/src/main/java/com/coveo/pushapiclient/CatalogSource.java +++ b/src/main/java/com/coveo/pushapiclient/CatalogSource.java @@ -5,8 +5,8 @@ // TODO: LENS-851 - Make public when ready class CatalogSource implements StreamEnabledSource { - private final PlatformClient platformClient; - private final String sourceId; + private final String apiKey; + private final ApiUrl urlExtractor; /** * Create a Catalog source instance from its @@ -41,12 +41,8 @@ class CatalogSource implements StreamEnabledSource { * @throws MalformedURLException */ public CatalogSource(String apiKey, URL sourceUrl) throws MalformedURLException { - ApiUrl parser = new ApiUrl(sourceUrl); - PlatformUrl platformUrl = parser.getPlatformUrl(); - String organizationId = parser.getOrganizationId(); - this.sourceId = parser.getSourceId(); - this.platformClient = new PlatformClient(apiKey, organizationId, - platformUrl); + this.apiKey = apiKey; + this.urlExtractor = new ApiUrl(sourceUrl); } /** @@ -80,11 +76,9 @@ public CatalogSource(String apiKey, URL sourceUrl) throws MalformedURLException * Administration Console * */ - public CatalogSource(String apiKey, String organizationId, String sourceId) { + public static CatalogSource fromPlatformUrl(String apiKey, String organizationId, String sourceId) { PlatformUrl platformUrl = new PlatformUrl(PlatformUrl.DEFAULT_ENVIRONMENT, PlatformUrl.DEFAULT_REGION); - this.sourceId = sourceId; - this.platformClient = new PlatformClient(apiKey, organizationId, - platformUrl); + return new CatalogSource(apiKey, organizationId, sourceId, platformUrl); } /** @@ -121,23 +115,40 @@ public CatalogSource(String apiKey, String organizationId, String sourceId) { * URL endpoint. * You can use the {@link PlatformUrl} when your * organization is located in a non-default Coveo - * environement and/or region. + * environement and/or region. When not specified, the + * default platform URL values will be used: + * {@link PlatformUrl#DEFAULT_ENVIRONMENT} and + * {@link PlatformUrl#DEFAULT_REGION} * */ - public CatalogSource(String apiKey, String organizationId, String sourceId, PlatformUrl platformUrl) { - this.sourceId = sourceId; - this.platformClient = new PlatformClient(apiKey, organizationId, - platformUrl); + public static CatalogSource fromPlatformUrl(String apiKey, String organizationId, String sourceId, + PlatformUrl platformUrl) { + return new CatalogSource(apiKey, organizationId, sourceId, platformUrl); + } + + private CatalogSource(String apiKey, String organizationId, String sourceId, PlatformUrl platformUrl) { + this.apiKey = apiKey; + this.urlExtractor = new ApiUrl(organizationId, sourceId, platformUrl); + } + + @Override + public String getOrganizationId() { + return this.urlExtractor.getOrganizationId(); + } + + @Override + public PlatformUrl getPlatformUrl() { + return this.urlExtractor.getPlatformUrl(); } @Override public String getId() { - return this.sourceId; + return this.urlExtractor.getSourceId(); } @Override - public PlatformClient getPlatformClient() { - return this.platformClient; + public String getApiKey() { + return this.apiKey; } } diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index d3daf06f..89d11da4 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -2,7 +2,8 @@ import java.io.IOException; -public class DocumentUploadQueue { +// TODO: LENS-851 - Make public +class DocumentUploadQueue { private final UpdloadStrategy uploader; public DocumentUploadQueue(UpdloadStrategy uploader) { @@ -10,16 +11,11 @@ public DocumentUploadQueue(UpdloadStrategy uploader) { } public void flush() throws IOException, InterruptedException { - } - - public void add(DocumentBuilder documentToAdd, DeleteDocument documentToDelete) - throws IOException, InterruptedException { + throw new UnsupportedOperationException("Unimplemented method (TODO: LENS-856)"); } public void add(DocumentBuilder document) throws IOException, InterruptedException { - // Once batch is ready, send it like: - // this.uploader.apply(batchUpdate); - throw new UnsupportedOperationException("Unimplemented method"); + 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 0659a1ff..fad82219 100644 --- a/src/main/java/com/coveo/pushapiclient/PlatformClient.java +++ b/src/main/java/com/coveo/pushapiclient/PlatformClient.java @@ -91,10 +91,10 @@ public HttpResponse createSource(String name, SourceVisibility sourceVis String[] headers = this.getHeaders(this.getAuthorizationHeader(), this.getContentTypeApplicationJSONHeader()); String json = this.toJSON(new HashMap<>() {{ - put("sourceType", "PUSH"); - put("pushEnabled", true); - put("name", name); - put("sourceVisibility", sourceVisibility); + put("sourceType", "PUSH"); + put("pushEnabled", true); + put("name", name); + put("sourceVisibility", sourceVisibility); }}); HttpRequest request = HttpRequest.newBuilder() diff --git a/src/main/java/com/coveo/pushapiclient/PushSource.java b/src/main/java/com/coveo/pushapiclient/PushSource.java index 92c777bf..4a17f9a5 100644 --- a/src/main/java/com/coveo/pushapiclient/PushSource.java +++ b/src/main/java/com/coveo/pushapiclient/PushSource.java @@ -9,17 +9,28 @@ // TODO: LENS-851 - Make public when ready class PushSource implements PushEnabledSource { + private final String apiKey; + private final ApiUrl urlExtractor; private final PlatformClient platformClient; - private final String sourceId; @Override - public PlatformClient getPlatformClient() { - return this.platformClient; + public String getOrganizationId() { + return this.urlExtractor.getOrganizationId(); + } + + @Override + public PlatformUrl getPlatformUrl() { + return this.urlExtractor.getPlatformUrl(); } @Override public String getId() { - return this.sourceId; + return this.urlExtractor.getSourceId(); + } + + @Override + public String getApiKey() { + return this.apiKey; } /** @@ -55,17 +66,15 @@ public String getId() { * @throws MalformedURLException */ public PushSource(String apiKey, URL sourceUrl) throws MalformedURLException { - ApiUrl parser = new ApiUrl(sourceUrl); - PlatformUrl platformUrl = parser.getPlatformUrl(); - String organizationId = parser.getOrganizationId(); - this.sourceId = parser.getSourceId(); - this.platformClient = new PlatformClient(apiKey, organizationId, - platformUrl); + this.apiKey = apiKey; + this.urlExtractor = new ApiUrl(sourceUrl); + String organizationId = urlExtractor.getOrganizationId(); + PlatformUrl platformUrl = urlExtractor.getPlatformUrl(); + this.platformClient = new PlatformClient(apiKey, organizationId, platformUrl); } /** - * Create a Push source instance from its - * Stream API URL + * Create a Push source instance * * @param apiKey The API key used for all operations regarding your * source. @@ -94,16 +103,13 @@ public PushSource(String apiKey, URL sourceUrl) throws MalformedURLException { * Administration Console * */ - public PushSource(String apiKey, String organizationId, String sourceId) { + public static PushSource fromPlatformUrl(String apiKey, String organizationId, String sourceId) { PlatformUrl platformUrl = new PlatformUrl(PlatformUrl.DEFAULT_ENVIRONMENT, PlatformUrl.DEFAULT_REGION); - this.sourceId = sourceId; - this.platformClient = new PlatformClient(apiKey, organizationId, - platformUrl); + return new PushSource(apiKey, organizationId, sourceId, platformUrl); } /** - * Create a Push source instance from its - * Stream API URL + * Create a Push source instance * * @param apiKey The API key used for all operations regarding your * source. @@ -135,13 +141,21 @@ public PushSource(String apiKey, String organizationId, String sourceId) { * URL endpoint. * You can use the {@link PlatformUrl} when your * organization is located in a non-default Coveo - * environement and/or region. + * environement and/or region. When not specified, the + * default platform URL values will be used: + * {@link PlatformUrl#DEFAULT_ENVIRONMENT} and + * {@link PlatformUrl#DEFAULT_REGION} * */ - public PushSource(String apiKey, String organizationId, String sourceId, PlatformUrl platformUrl) { - this.sourceId = sourceId; - this.platformClient = new PlatformClient(apiKey, organizationId, - platformUrl); + public static PushSource fromPlatformUrl(String apiKey, String organizationId, String sourceId, + PlatformUrl platformUrl) { + return new PushSource(apiKey, organizationId, sourceId, platformUrl); + } + + private PushSource(String apiKey, String organizationId, String sourceId, PlatformUrl platformUrl) { + this.apiKey = apiKey; + this.urlExtractor = new ApiUrl(organizationId, sourceId, platformUrl); + this.platformClient = new PlatformClient(apiKey, organizationId, platformUrl); } /** @@ -202,7 +216,7 @@ public HttpResponse deleteSecurityIdentity(String securityProviderId, */ public HttpResponse updateSourceStatus(PushAPIStatus status) throws IOException, InterruptedException { - return this.platformClient.updateSourceStatus(this.sourceId, status); + return this.platformClient.updateSourceStatus(this.getId(), status); } /** @@ -277,7 +291,7 @@ public HttpResponse addOrUpdateDocument(DocumentBuilder docBuilder) CompressionType compressionType = docBuilder.getDocument().compressedBinaryData != null ? docBuilder.getDocument().compressedBinaryData.getCompressionType() : CompressionType.UNCOMPRESSED; - return this.platformClient.pushDocument(this.sourceId, docBuilder.marshal(), docBuilder.getDocument().uri, + return this.platformClient.pushDocument(this.getId(), docBuilder.marshal(), docBuilder.getDocument().uri, compressionType); } @@ -294,7 +308,7 @@ public HttpResponse addOrUpdateDocument(DocumentBuilder docBuilder) */ public HttpResponse deleteDocument(String documentId, Boolean deleteChildren) throws IOException, InterruptedException { - return this.platformClient.deleteDocument(this.sourceId, documentId, deleteChildren); + return this.platformClient.deleteDocument(this.getId(), documentId, deleteChildren); } } 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 index 60cf6af9..c949176d 100644 --- a/src/main/java/com/coveo/pushapiclient/StreamService.java +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -2,25 +2,71 @@ import java.io.IOException; import java.net.http.HttpResponse; + +import com.coveo.pushapiclient.exceptions.NoOpenStreamException; import com.google.gson.Gson; -public class StreamService { - private StreamEnabledSource source; +// TODO: LENS-851 - Make public +class StreamService { + private final StreamEnabledSource source; + private final PlatformClient platformClient; private String streamId; private DocumentUploadQueue queue; + /** + * Creates a service to stream your documents to to 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 document to. + */ public StreamService(StreamEnabledSource source) { + String apiKey = source.getApiKey(); + String organizationId = source.getOrganizationId(); + PlatformUrl platformUrl = source.getPlatformUrl(); UpdloadStrategy uploader = this.getUploadStrategy(); + this.source = source; this.queue = new DocumentUploadQueue(uploader); + this.platformClient = new PlatformClient(apiKey, organizationId, platformUrl); } /** - * Pushes document to the source. - * If multiple documents are added, the class will ensure documents are - * automatically batched into chunks that do not exceed API limit size. + * Adds documents to the previously provided 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 be added, 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 + * @param document The documentBuilder to add to your source * @throws InterruptedException * @throws IOException */ @@ -31,33 +77,59 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti queue.add(document); } - public HttpResponse close() throws IOException, InterruptedException { + /** + * Sends any buffered documents and close 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. + * + *

+ * 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 { if (this.streamId == null) { - throw new java.lang.UnsupportedOperationException("TODO: custom error: No stream was open yet"); + throw new NoOpenStreamException( + "No open stream detected. A stream will automatically be opened once you start adding documents."); } queue.flush(); - PlatformClient platformClient = source.getPlatformClient(); - String sourceId = this.source.getId(); - return platformClient.closeStream(sourceId, this.streamId); + String sourceId = this.getSourceId(); + return this.platformClient.closeStream(sourceId, this.streamId); } private UpdloadStrategy getUploadStrategy() { return (batchUpdate) -> { - String sourceId = this.source.getId(); - PlatformClient platformClient = source.getPlatformClient(); - HttpResponse resFileContainer = platformClient.requireStreamChunk(sourceId, this.streamId); + 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 platformClient.uploadContentToFileContainer(fileContainer, + return this.platformClient.uploadContentToFileContainer(fileContainer, batchUpdateJson); }; } private String getStreamId() throws IOException, InterruptedException { - String sourceId = this.source.getId(); - HttpResponse response = this.source.getPlatformClient().openStream(sourceId); - return "TODO: get streamID from response"; + 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/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/StreamServiceTest.java b/src/test/java/com/coveo/pushapiclient/StreamServiceTest.java new file mode 100644 index 00000000..9c4aadcf --- /dev/null +++ b/src/test/java/com/coveo/pushapiclient/StreamServiceTest.java @@ -0,0 +1,103 @@ +package com.coveo.pushapiclient; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import com.coveo.pushapiclient.exceptions.NoOpenStreamException; + +import java.io.IOException; +import java.net.URL; +import java.net.http.HttpResponse; + +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +@RunWith(PowerMockRunner.class) +@PrepareForTest(StreamService.class) +public class StreamServiceTest { + StreamService service; + + DocumentUploadQueue queueMock; + PlatformClient platformClientMock; + HttpResponse httpResponseMock; + + private DocumentBuilder documentA; + private DocumentBuilder documentB; + + @Before + public void setUp() throws Exception { + this.queueMock = PowerMockito.mock(DocumentUploadQueue.class); + this.platformClientMock = PowerMockito.mock(PlatformClient.class); + this.httpResponseMock = PowerMockito.mock(HttpResponse.class); + + PowerMockito + .whenNew(PlatformClient.class) + .withAnyArguments() + .thenReturn(this.platformClientMock); + PowerMockito + .whenNew(DocumentUploadQueue.class) + .withAnyArguments() + .thenReturn(this.queueMock); + + PowerMockito + .when(this.httpResponseMock.body()) + .thenReturn("{\"streamId\": \"stream-id\"}"); + PowerMockito + .when(this.platformClientMock.openStream("my-source-id")) + .thenReturn(this.httpResponseMock); + + 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"); + + URL sourceUrl = new URL( + "https://api.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/stream/open"); + CatalogSource source = new CatalogSource("api_key", sourceUrl); + service = new StreamService(source); + } + + @Test + public void testAddShouldOpenANewStream() throws IOException, InterruptedException { + service.add(documentA); + service.add(documentB); + + verify(this.platformClientMock, times(1)).openStream("my-source-id"); + } + + @Test + public void testAddShouldAddDocumentToQueue() throws IOException, InterruptedException { + service.add(documentA); + service.add(documentB); + + verify(this.queueMock, times(1)).add(documentA); + verify(this.queueMock, times(1)).add(documentB); + } + + @Test + public void testCloseShouldCloseOpenStream() throws IOException, InterruptedException, NoOpenStreamException { + service.add(documentA); + service.close(); + + verify(this.platformClientMock, times(1)).closeStream("my-source-id", + "stream-id"); + } + + @Test + public void testCloseShouldFlushBufferedDocuments() + throws IOException, InterruptedException, NoOpenStreamException { + service.add(documentA); + service.close(); + + verify(this.queueMock, times(1)).flush(); + } + + @Test(expected = NoOpenStreamException.class) + public void givenNoOpenStream_whenClose_thenShouldThrow() + throws IOException, InterruptedException, NoOpenStreamException { + service.close(); + } + +} \ No newline at end of file From 7f9ad5e4e29603cd1bac72c93ea3c5806ea4417c Mon Sep 17 00:00:00 2001 From: Yassine Date: Fri, 26 May 2023 08:30:41 -0400 Subject: [PATCH 08/23] Update src/main/java/com/coveo/pushapiclient/StreamService.java Co-authored-by: Benjamin Taillon <54454747+btaillon@users.noreply.github.com> --- src/main/java/com/coveo/pushapiclient/StreamService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/com/coveo/pushapiclient/StreamService.java b/src/main/java/com/coveo/pushapiclient/StreamService.java index c949176d..fbb2ac24 100644 --- a/src/main/java/com/coveo/pushapiclient/StreamService.java +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -14,7 +14,7 @@ class StreamService { private DocumentUploadQueue queue; /** - * Creates a service to stream your documents to to provided source by + * Creates a service to stream your documents to the provided source by * interacting with * the Stream API. * From 3499404442565681ea7c6a35285f0f851d0132e1 Mon Sep 17 00:00:00 2001 From: Yassine Date: Fri, 26 May 2023 08:31:26 -0400 Subject: [PATCH 09/23] Update src/main/java/com/coveo/pushapiclient/StreamService.java Co-authored-by: jpmarceau <39384459+jpmarceau@users.noreply.github.com> --- src/main/java/com/coveo/pushapiclient/StreamService.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/StreamService.java b/src/main/java/com/coveo/pushapiclient/StreamService.java index fbb2ac24..c87df4d0 100644 --- a/src/main/java/com/coveo/pushapiclient/StreamService.java +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -15,8 +15,7 @@ class StreamService { /** * Creates a service to stream your documents to the provided source by - * interacting with - * the Stream API. + * interacting with the Stream API. * *

* To perform full document From f3ec845675e351ac53571c8a91ab1c598249956b Mon Sep 17 00:00:00 2001 From: Yassine Date: Fri, 26 May 2023 08:32:03 -0400 Subject: [PATCH 10/23] Apply suggestions from code review Co-authored-by: jpmarceau <39384459+jpmarceau@users.noreply.github.com> --- .../com/coveo/pushapiclient/StreamService.java | 14 +++++--------- 1 file changed, 5 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/StreamService.java b/src/main/java/com/coveo/pushapiclient/StreamService.java index c87df4d0..aadc3848 100644 --- a/src/main/java/com/coveo/pushapiclient/StreamService.java +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -19,11 +19,11 @@ class StreamService { * *

* To perform full document - * updates, use the {@PushService} since pushing documents with the + * 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 document to. + * @param source The source to which you want to send your documents. */ public StreamService(StreamEnabledSource source) { String apiKey = source.getApiKey(); @@ -37,7 +37,7 @@ public StreamService(StreamEnabledSource source) { } /** - * Adds documents to the previously provided source. + * Adds documents to the previously specified source. * This function will open a stream before uploading documents into it. * *

@@ -47,7 +47,7 @@ public StreamService(StreamEnabledSource source) { * set for the Stream API. * *

- * Once there are no more documents to be added, it is important to call the {@link StreamService#close} function + * 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. * @@ -77,7 +77,7 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti } /** - * Sends any buffered documents and close the stream. + * 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. @@ -89,10 +89,6 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti * of the source completely. Expect a 15-minute delay for the removal of the old * items from the index. * - *

- * Expect a 15-minute delay for the removal of the old - * items from the index. - * * @return * @throws IOException * @throws InterruptedException From 508564e29fa05ce7d5a68c779e0e817aa400314a Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Fri, 26 May 2023 10:04:02 -0400 Subject: [PATCH 11/23] refactor unit tests --- pom.xml | 16 +-- .../coveo/pushapiclient/StreamService.java | 22 +--- .../pushapiclient/StreamServiceInternal.java | 52 +++++++++ .../StreamServiceInternalTest.java | 96 ++++++++++++++++ .../pushapiclient/StreamServiceTest.java | 103 ------------------ 5 files changed, 153 insertions(+), 136 deletions(-) create mode 100644 src/main/java/com/coveo/pushapiclient/StreamServiceInternal.java create mode 100644 src/test/java/com/coveo/pushapiclient/StreamServiceInternalTest.java delete mode 100644 src/test/java/com/coveo/pushapiclient/StreamServiceTest.java diff --git a/pom.xml b/pom.xml index 7580afeb..142e7175 100644 --- a/pom.xml +++ b/pom.xml @@ -137,26 +137,12 @@ joda-time 2.10.10 - - - org.powermock - powermock-module-junit4 - 2.0.9 - test - - - org.powermock - powermock-api-mockito2 - 2.0.9 - test - org.mockito mockito-core - 2.23.0 + 4.7.0 test - junit junit diff --git a/src/main/java/com/coveo/pushapiclient/StreamService.java b/src/main/java/com/coveo/pushapiclient/StreamService.java index c949176d..07876bb4 100644 --- a/src/main/java/com/coveo/pushapiclient/StreamService.java +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -10,6 +10,7 @@ class StreamService { private final StreamEnabledSource source; private final PlatformClient platformClient; + private StreamServiceInternal service; private String streamId; private DocumentUploadQueue queue; @@ -35,6 +36,7 @@ public StreamService(StreamEnabledSource source) { 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); } /** @@ -71,10 +73,7 @@ public StreamService(StreamEnabledSource source) { * @throws IOException */ public void add(DocumentBuilder document) throws IOException, InterruptedException { - if (this.streamId == null) { - this.streamId = this.getStreamId(); - } - queue.add(document); + this.service.add(document); } /** @@ -100,13 +99,7 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti * @throws NoOpenStreamException */ 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); + return this.service.close(); } private UpdloadStrategy getUploadStrategy() { @@ -121,13 +114,6 @@ private UpdloadStrategy getUploadStrategy() { }; } - 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/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/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 diff --git a/src/test/java/com/coveo/pushapiclient/StreamServiceTest.java b/src/test/java/com/coveo/pushapiclient/StreamServiceTest.java deleted file mode 100644 index 9c4aadcf..00000000 --- a/src/test/java/com/coveo/pushapiclient/StreamServiceTest.java +++ /dev/null @@ -1,103 +0,0 @@ -package com.coveo.pushapiclient; - -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.powermock.api.mockito.PowerMockito; -import org.powermock.core.classloader.annotations.PrepareForTest; -import org.powermock.modules.junit4.PowerMockRunner; - -import com.coveo.pushapiclient.exceptions.NoOpenStreamException; - -import java.io.IOException; -import java.net.URL; -import java.net.http.HttpResponse; - -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; - -@RunWith(PowerMockRunner.class) -@PrepareForTest(StreamService.class) -public class StreamServiceTest { - StreamService service; - - DocumentUploadQueue queueMock; - PlatformClient platformClientMock; - HttpResponse httpResponseMock; - - private DocumentBuilder documentA; - private DocumentBuilder documentB; - - @Before - public void setUp() throws Exception { - this.queueMock = PowerMockito.mock(DocumentUploadQueue.class); - this.platformClientMock = PowerMockito.mock(PlatformClient.class); - this.httpResponseMock = PowerMockito.mock(HttpResponse.class); - - PowerMockito - .whenNew(PlatformClient.class) - .withAnyArguments() - .thenReturn(this.platformClientMock); - PowerMockito - .whenNew(DocumentUploadQueue.class) - .withAnyArguments() - .thenReturn(this.queueMock); - - PowerMockito - .when(this.httpResponseMock.body()) - .thenReturn("{\"streamId\": \"stream-id\"}"); - PowerMockito - .when(this.platformClientMock.openStream("my-source-id")) - .thenReturn(this.httpResponseMock); - - 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"); - - URL sourceUrl = new URL( - "https://api.cloud.coveo.com/push/v1/organizations/my-org-id/sources/my-source-id/stream/open"); - CatalogSource source = new CatalogSource("api_key", sourceUrl); - service = new StreamService(source); - } - - @Test - public void testAddShouldOpenANewStream() throws IOException, InterruptedException { - service.add(documentA); - service.add(documentB); - - verify(this.platformClientMock, times(1)).openStream("my-source-id"); - } - - @Test - public void testAddShouldAddDocumentToQueue() throws IOException, InterruptedException { - service.add(documentA); - service.add(documentB); - - verify(this.queueMock, times(1)).add(documentA); - verify(this.queueMock, times(1)).add(documentB); - } - - @Test - public void testCloseShouldCloseOpenStream() throws IOException, InterruptedException, NoOpenStreamException { - service.add(documentA); - service.close(); - - verify(this.platformClientMock, times(1)).closeStream("my-source-id", - "stream-id"); - } - - @Test - public void testCloseShouldFlushBufferedDocuments() - throws IOException, InterruptedException, NoOpenStreamException { - service.add(documentA); - service.close(); - - verify(this.queueMock, times(1)).flush(); - } - - @Test(expected = NoOpenStreamException.class) - public void givenNoOpenStream_whenClose_thenShouldThrow() - throws IOException, InterruptedException, NoOpenStreamException { - service.close(); - } - -} \ No newline at end of file From b1219528dd12746cf1e2c551b28f0379b6d8f676 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Mon, 29 May 2023 14:33:35 -0400 Subject: [PATCH 12/23] create batch update accumulator --- .../pushapiclient/BatchUpdateAccumulator.java | 49 ++++++++++++++ .../pushapiclient/DocumentUploadQueue.java | 65 +++++++++++++++++-- 2 files changed, 110 insertions(+), 4 deletions(-) create mode 100644 src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java diff --git a/src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java b/src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java new file mode 100644 index 00000000..38305382 --- /dev/null +++ b/src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java @@ -0,0 +1,49 @@ +package com.coveo.pushapiclient; + +import java.util.ArrayList; +import java.util.List; + +/** + * Accumulates documents to be added or deleted in a queue for batch updates. + */ +class BatchUpdateAccumulator { + static final int maxContentLength = 5 * 1024 * 1024; + private List documentToAddList; + private List documentToDeleteList; + private int size; + + public BatchUpdateAccumulator() { + documentToAddList = new ArrayList<>(); + documentToDeleteList = new ArrayList<>(); + // TODO: LENS-843: include partial document updates + } + + public void resetBatch() { + this.size = 0; + } + + public BatchUpdate getBatch() { + return new BatchUpdate(this.documentToAddList, this.documentToDeleteList); + } + + public void addToBatch(DocumentBuilder document) { + documentToAddList.add(document); + } + + public void addToBatch(DeleteDocument document) { + documentToDeleteList.add(document); + } + + public boolean isEmpty() { + // TODO: LENS-843: include partial document updates + return documentToAddList.isEmpty() && documentToDeleteList.isEmpty(); + } + + public int getSize() { + return this.size; + } + + public void setSize(int size) { + this.size = size; + } +} \ 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 89d11da4..034dd91a 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -2,20 +2,77 @@ import java.io.IOException; -// TODO: LENS-851 - Make public +/** + * Represents a queue for uploading documents using a specified upload strategy + * and accumulator. + */ class DocumentUploadQueue { private final UpdloadStrategy uploader; + private final BatchUpdateAccumulator accumulator; - public DocumentUploadQueue(UpdloadStrategy uploader) { + /** + * Constructs a new DocumentUploadQueue object. + * + * @param uploader The upload strategy to be used for document uploads. + * @param accumulator The accumulator for queuing documents to be uploaded. + */ + public DocumentUploadQueue(UpdloadStrategy uploader, BatchUpdateAccumulator accumulator) { this.uploader = uploader; + this.accumulator = accumulator; } + /** + * 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.accumulator.isEmpty()) { + BatchUpdate batch = this.accumulator.getBatch(); + // TODO: LENS-871: support concurrent requests + this.uploader.apply(batch); + } + this.accumulator.resetBatch(); + this.accumulator.setSize(0); } + /** + * 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)"); + final int sizeOfDoc = document.marshal().getBytes().length; + if (accumulator.getSize() + sizeOfDoc >= BatchUpdateAccumulator.maxContentLength) { + this.flush(); + } + this.accumulator.addToBatch(document); + this.accumulator.setSize(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 { + final int sizeOfDoc = document.marshalJsonObject().toString().getBytes().length; + if (accumulator.getSize() + sizeOfDoc >= BatchUpdateAccumulator.maxContentLength) { + this.flush(); + } + this.accumulator.addToBatch(document); + this.accumulator.setSize(sizeOfDoc); + } + + // TODO: LENS-843: include partial document updates + } From 476a09dde98ff1ad371f4eea51d4650f7c68584f Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Mon, 29 May 2023 15:44:54 -0400 Subject: [PATCH 13/23] rework queue --- .../pushapiclient/BatchUpdateAccumulator.java | 49 --------------- .../pushapiclient/DocumentUploadQueue.java | 59 ++++++++++++------- 2 files changed, 37 insertions(+), 71 deletions(-) delete mode 100644 src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java diff --git a/src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java b/src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java deleted file mode 100644 index 38305382..00000000 --- a/src/main/java/com/coveo/pushapiclient/BatchUpdateAccumulator.java +++ /dev/null @@ -1,49 +0,0 @@ -package com.coveo.pushapiclient; - -import java.util.ArrayList; -import java.util.List; - -/** - * Accumulates documents to be added or deleted in a queue for batch updates. - */ -class BatchUpdateAccumulator { - static final int maxContentLength = 5 * 1024 * 1024; - private List documentToAddList; - private List documentToDeleteList; - private int size; - - public BatchUpdateAccumulator() { - documentToAddList = new ArrayList<>(); - documentToDeleteList = new ArrayList<>(); - // TODO: LENS-843: include partial document updates - } - - public void resetBatch() { - this.size = 0; - } - - public BatchUpdate getBatch() { - return new BatchUpdate(this.documentToAddList, this.documentToDeleteList); - } - - public void addToBatch(DocumentBuilder document) { - documentToAddList.add(document); - } - - public void addToBatch(DeleteDocument document) { - documentToDeleteList.add(document); - } - - public boolean isEmpty() { - // TODO: LENS-843: include partial document updates - return documentToAddList.isEmpty() && documentToDeleteList.isEmpty(); - } - - public int getSize() { - return this.size; - } - - public void setSize(int size) { - this.size = size; - } -} \ 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 034dd91a..9ba4074b 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -1,24 +1,29 @@ package com.coveo.pushapiclient; import java.io.IOException; +import java.util.ArrayList; +import java.util.List; /** * Represents a queue for uploading documents using a specified upload strategy - * and accumulator. */ class DocumentUploadQueue { + static final int maxContentLength = 5 * 1024 * 1024; private final UpdloadStrategy uploader; - private final BatchUpdateAccumulator accumulator; + + private List documentToAddList; + private List documentToDeleteList; + private int size; /** * Constructs a new DocumentUploadQueue object. * - * @param uploader The upload strategy to be used for document uploads. - * @param accumulator The accumulator for queuing documents to be uploaded. + * @param uploader The upload strategy to be used for document uploads. */ - public DocumentUploadQueue(UpdloadStrategy uploader, BatchUpdateAccumulator accumulator) { + public DocumentUploadQueue(UpdloadStrategy uploader) { + this.documentToAddList = new ArrayList<>(); + this.documentToDeleteList = new ArrayList<>(); this.uploader = uploader; - this.accumulator = accumulator; } /** @@ -28,13 +33,12 @@ public DocumentUploadQueue(UpdloadStrategy uploader, BatchUpdateAccumulator accu * @throws InterruptedException If the upload process is interrupted. */ public void flush() throws IOException, InterruptedException { - if (!this.accumulator.isEmpty()) { - BatchUpdate batch = this.accumulator.getBatch(); - // TODO: LENS-871: support concurrent requests - this.uploader.apply(batch); - } - this.accumulator.resetBatch(); - this.accumulator.setSize(0); + BatchUpdate batch = this.getBatch(); + // TODO: LENS-871: support concurrent requests + this.uploader.apply(batch); + this.size = 0; + this.documentToAddList.clear(); + this.documentToDeleteList.clear(); } /** @@ -42,17 +46,19 @@ public void flush() throws IOException, InterruptedException { * it exceeds the maximum content length. * See {@link DocumentUploadQueue#flush}. * - * @param document The document to be added to the index. + * @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 { final int sizeOfDoc = document.marshal().getBytes().length; - if (accumulator.getSize() + sizeOfDoc >= BatchUpdateAccumulator.maxContentLength) { + if (!this.isEmpty() && this.size + sizeOfDoc >= maxContentLength) { this.flush(); } - this.accumulator.addToBatch(document); - this.accumulator.setSize(sizeOfDoc); + if (document != null) { + documentToAddList.add(document); + this.size += sizeOfDoc; + } } /** @@ -60,19 +66,28 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti * 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 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 { final int sizeOfDoc = document.marshalJsonObject().toString().getBytes().length; - if (accumulator.getSize() + sizeOfDoc >= BatchUpdateAccumulator.maxContentLength) { + if (!this.isEmpty() && this.size + sizeOfDoc >= maxContentLength) { this.flush(); } - this.accumulator.addToBatch(document); - this.accumulator.setSize(sizeOfDoc); + if (document != null) { + documentToDeleteList.add(document); + this.size += sizeOfDoc; + } } - // TODO: LENS-843: include partial document updates + private BatchUpdate getBatch() { + return new BatchUpdate(this.documentToAddList, this.documentToDeleteList); + } + + private boolean isEmpty() { + // TODO: LENS-843: include partial document updates + return documentToAddList.isEmpty() && documentToDeleteList.isEmpty(); + } } From 141889d9377410b69acf9e0fedf608c313043b65 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 13:03:01 -0400 Subject: [PATCH 14/23] complete unit tests --- .../pushapiclient/DocumentUploadQueue.java | 51 ++++- .../DocumentUploadQueueTest.java | 192 ++++++++++++++++++ 2 files changed, 232 insertions(+), 11 deletions(-) create mode 100644 src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index 9ba4074b..e5e9cc37 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -2,21 +2,21 @@ import java.io.IOException; import java.util.ArrayList; -import java.util.List; /** * Represents a queue for uploading documents using a specified upload strategy */ class DocumentUploadQueue { - static final int maxContentLength = 5 * 1024 * 1024; + static final int defaultMaxQueueSize = 5 * 1024 * 1024; private final UpdloadStrategy uploader; - - private List documentToAddList; - private List documentToDeleteList; + private final int maxQueueSize; + private ArrayList documentToAddList; + private ArrayList documentToDeleteList; private int size; /** - * Constructs a new DocumentUploadQueue object. + * 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. */ @@ -24,8 +24,24 @@ public DocumentUploadQueue(UpdloadStrategy uploader) { this.documentToAddList = new ArrayList<>(); this.documentToDeleteList = new ArrayList<>(); this.uploader = uploader; + this.maxQueueSize = defaultMaxQueueSize; } + // /** + // * Constructs a new DocumentUploadQueue object with the specified uploader and + // * maximum queue size. + // * + // * @param uploader The upload strategy to be used for document uploads. + // * @param maxQueueSize The maximum size of the upload queue before it gets + // * automatically flushed. + // */ + // public DocumentUploadQueue(UpdloadStrategy uploader, int maxQueueSize) { + // this.documentToAddList = new ArrayList<>(); + // this.documentToDeleteList = new ArrayList<>(); + // this.uploader = uploader; + // this.maxQueueSize = maxQueueSize; + // } + /** * Flushes the accumulated documents by applying the upload strategy. * @@ -33,6 +49,9 @@ public DocumentUploadQueue(UpdloadStrategy uploader) { * @throws InterruptedException If the upload process is interrupted. */ public void flush() throws IOException, InterruptedException { + if (this.isEmpty()) { + return; + } BatchUpdate batch = this.getBatch(); // TODO: LENS-871: support concurrent requests this.uploader.apply(batch); @@ -51,8 +70,12 @@ public void flush() throws IOException, InterruptedException { * @throws InterruptedException If the upload process is interrupted. */ public void add(DocumentBuilder document) throws IOException, InterruptedException { + if (document == null) { + return; + } + final int sizeOfDoc = document.marshal().getBytes().length; - if (!this.isEmpty() && this.size + sizeOfDoc >= maxContentLength) { + if (this.size + sizeOfDoc >= this.maxQueueSize) { this.flush(); } if (document != null) { @@ -71,8 +94,12 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti * @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.isEmpty() && this.size + sizeOfDoc >= maxContentLength) { + if (this.size + sizeOfDoc >= this.maxQueueSize) { this.flush(); } if (document != null) { @@ -81,11 +108,13 @@ public void add(DeleteDocument document) throws IOException, InterruptedExceptio } } - private BatchUpdate getBatch() { - return new BatchUpdate(this.documentToAddList, this.documentToDeleteList); + public BatchUpdate getBatch() { + return new BatchUpdate( + new ArrayList(this.documentToAddList), + new ArrayList(this.documentToDeleteList)); } - private boolean isEmpty() { + 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..7f5e2ef3 --- /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 UpdloadStrategy updloadStrategy; + + @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(updloadStrategy, 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 the same 2MB document 3 times to the queue. After adding the second + // document, the queue size will reach 6MB, which exceeds the maximum queue size + // limit. Therefore the added documents will automatically be uploaded to the + // source. + queue.add(firstBulkyDocument); + queue.add(secondBulkyDocument); + + // The 3rd document to be added 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(updloadStrategy, times(1)).apply(any(BatchUpdate.class)); + verify(updloadStrategy, 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 the same 2MB document 3 times to the queue. After adding the second + // document, the queue size will reach 6MB, which exceeds the maximum queue size + // limit. Therefore the 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(updloadStrategy, times(2)).apply(any(BatchUpdate.class)); + verify(updloadStrategy, times(1)).apply(firstBatch); + verify(updloadStrategy, times(1)).apply(secondBatch); + } + + @Test + public void testAddingEmptyDocument() throws IOException, InterruptedException { + DocumentBuilder nullDocument = null; + + queue.add(nullDocument); + queue.flush(); + + verify(updloadStrategy, times(0)).apply(any(BatchUpdate.class)); + } +} From aa682402b95bab19af59bb3fab9506525cc374de Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 13:07:53 -0400 Subject: [PATCH 15/23] remove comments --- .../pushapiclient/DocumentUploadQueue.java | 19 +------------------ 1 file changed, 1 insertion(+), 18 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index e5e9cc37..fa7589b5 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -7,9 +7,8 @@ * Represents a queue for uploading documents using a specified upload strategy */ class DocumentUploadQueue { - static final int defaultMaxQueueSize = 5 * 1024 * 1024; private final UpdloadStrategy uploader; - private final int maxQueueSize; + private final int maxQueueSize = 5 * 1024 * 1024; private ArrayList documentToAddList; private ArrayList documentToDeleteList; private int size; @@ -24,24 +23,8 @@ public DocumentUploadQueue(UpdloadStrategy uploader) { this.documentToAddList = new ArrayList<>(); this.documentToDeleteList = new ArrayList<>(); this.uploader = uploader; - this.maxQueueSize = defaultMaxQueueSize; } - // /** - // * Constructs a new DocumentUploadQueue object with the specified uploader and - // * maximum queue size. - // * - // * @param uploader The upload strategy to be used for document uploads. - // * @param maxQueueSize The maximum size of the upload queue before it gets - // * automatically flushed. - // */ - // public DocumentUploadQueue(UpdloadStrategy uploader, int maxQueueSize) { - // this.documentToAddList = new ArrayList<>(); - // this.documentToDeleteList = new ArrayList<>(); - // this.uploader = uploader; - // this.maxQueueSize = maxQueueSize; - // } - /** * Flushes the accumulated documents by applying the upload strategy. * From 369525a18303a225ce62699d94d7418c814a9c61 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 13:10:54 -0400 Subject: [PATCH 16/23] update comment --- .../DocumentUploadQueueTest.java | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java b/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java index 7f5e2ef3..a0ed31b9 100644 --- a/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java +++ b/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java @@ -125,16 +125,16 @@ public void testShouldAutomaticallyFlushAccumulatedDocuments() throws IOExceptio } }, emptyList); - // Adding the same 2MB document 3 times to the queue. After adding the second - // document, the queue size will reach 6MB, which exceeds the maximum queue size - // limit. Therefore the added documents will automatically be uploaded to the - // source. + // 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 to be added 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 + // 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(updloadStrategy, times(1)).apply(any(BatchUpdate.class)); @@ -162,10 +162,10 @@ public void testShouldManuallyFlushAccumulatedDocuments() throws IOException, In } }, emptyList); - // Adding the same 2MB document 3 times to the queue. After adding the second - // document, the queue size will reach 6MB, which exceeds the maximum queue size - // limit. Therefore the added documents will automatically be uploaded to the - // source. + // 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); From 7d5dc2c56992f66dc364f6dec66c0ec498ce59f6 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 14:30:30 -0400 Subject: [PATCH 17/23] draft PushService --- .../com/coveo/pushapiclient/PushService.java | 51 +++++++++++++++++++ .../pushapiclient/PushServiceInternal.java | 24 +++++++++ 2 files changed, 75 insertions(+) create mode 100644 src/main/java/com/coveo/pushapiclient/PushService.java create mode 100644 src/main/java/com/coveo/pushapiclient/PushServiceInternal.java 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..8adc5f07 --- /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(); + UpdloadStrategy 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 UpdloadStrategy 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(); + } + +} From c90446cd3a1f466d4f0987198da73594f80be344 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 15:59:06 -0400 Subject: [PATCH 18/23] apply corrections --- .../java/com/coveo/pushapiclient/DocumentUploadQueue.java | 4 ++-- src/main/java/com/coveo/pushapiclient/PlatformClient.java | 2 ++ src/main/java/com/coveo/pushapiclient/StreamService.java | 4 ++-- .../{UpdloadStrategy.java => UploadStrategy.java} | 2 +- 4 files changed, 7 insertions(+), 5 deletions(-) rename src/main/java/com/coveo/pushapiclient/{UpdloadStrategy.java => UploadStrategy.java} (86%) diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index 89d11da4..b21e0dd5 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -4,9 +4,9 @@ // TODO: LENS-851 - Make public class DocumentUploadQueue { - private final UpdloadStrategy uploader; + private final UploadStrategy uploader; - public DocumentUploadQueue(UpdloadStrategy uploader) { + public DocumentUploadQueue(UploadStrategy uploader) { this.uploader = uploader; } diff --git a/src/main/java/com/coveo/pushapiclient/PlatformClient.java b/src/main/java/com/coveo/pushapiclient/PlatformClient.java index fad82219..7f0a1705 100644 --- a/src/main/java/com/coveo/pushapiclient/PlatformClient.java +++ b/src/main/java/com/coveo/pushapiclient/PlatformClient.java @@ -284,8 +284,10 @@ public HttpResponse deleteDocument(String sourceId, String documentId, B 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) diff --git a/src/main/java/com/coveo/pushapiclient/StreamService.java b/src/main/java/com/coveo/pushapiclient/StreamService.java index 5a79806b..b9212e20 100644 --- a/src/main/java/com/coveo/pushapiclient/StreamService.java +++ b/src/main/java/com/coveo/pushapiclient/StreamService.java @@ -30,7 +30,7 @@ public StreamService(StreamEnabledSource source) { String apiKey = source.getApiKey(); String organizationId = source.getOrganizationId(); PlatformUrl platformUrl = source.getPlatformUrl(); - UpdloadStrategy uploader = this.getUploadStrategy(); + UploadStrategy uploader = this.getUploadStrategy(); this.source = source; this.queue = new DocumentUploadQueue(uploader); @@ -97,7 +97,7 @@ public HttpResponse close() throws IOException, InterruptedException, No return this.service.close(); } - private UpdloadStrategy getUploadStrategy() { + private UploadStrategy getUploadStrategy() { return (batchUpdate) -> { String sourceId = this.getSourceId(); HttpResponse resFileContainer = this.platformClient.requireStreamChunk(sourceId, this.streamId); diff --git a/src/main/java/com/coveo/pushapiclient/UpdloadStrategy.java b/src/main/java/com/coveo/pushapiclient/UploadStrategy.java similarity index 86% rename from src/main/java/com/coveo/pushapiclient/UpdloadStrategy.java rename to src/main/java/com/coveo/pushapiclient/UploadStrategy.java index f5bb812e..e77dd01c 100644 --- a/src/main/java/com/coveo/pushapiclient/UpdloadStrategy.java +++ b/src/main/java/com/coveo/pushapiclient/UploadStrategy.java @@ -4,6 +4,6 @@ import java.net.http.HttpResponse; @FunctionalInterface -public interface UpdloadStrategy { +public interface UploadStrategy { HttpResponse apply(BatchUpdate batchUpdate) throws IOException, InterruptedException; } From cad5aac1d0ee06c234a1cc4fb045cd3065815520 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 16:01:02 -0400 Subject: [PATCH 19/23] add gitignore --- .gitignore | 1 + 1 file changed, 1 insertion(+) 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 From 0b5baa11fff0858507a2263ab2ecc360a6d7c8fa Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 16:04:43 -0400 Subject: [PATCH 20/23] fix merge conflict --- .../pushapiclient/DocumentUploadQueueTest.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java b/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java index a0ed31b9..5e346a41 100644 --- a/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java +++ b/src/test/java/com/coveo/pushapiclient/DocumentUploadQueueTest.java @@ -20,7 +20,7 @@ public class DocumentUploadQueueTest { @Mock - private UpdloadStrategy updloadStrategy; + private UploadStrategy uploadStrategy; @InjectMocks private DocumentUploadQueue queue; @@ -108,7 +108,7 @@ public void testFlushShouldNotUploadDocumentaWhenRequiredSizeIsNotMet() throws I queue.add(documentToAdd); queue.add(documentToDelete); - verify(updloadStrategy, times(0)).apply(any(BatchUpdate.class)); + verify(uploadStrategy, times(0)).apply(any(BatchUpdate.class)); } @Test @@ -137,8 +137,8 @@ public void testShouldAutomaticallyFlushAccumulatedDocuments() throws IOExceptio // queue size limit has been reached queue.add(thirdBulkyDocument); - verify(updloadStrategy, times(1)).apply(any(BatchUpdate.class)); - verify(updloadStrategy, times(1)).apply(firstBatch); + verify(uploadStrategy, times(1)).apply(any(BatchUpdate.class)); + verify(uploadStrategy, times(1)).apply(firstBatch); } @Test @@ -175,9 +175,9 @@ public void testShouldManuallyFlushAccumulatedDocuments() throws IOException, In // Additional flush will have no effect if documents where already flushed queue.flush(); - verify(updloadStrategy, times(2)).apply(any(BatchUpdate.class)); - verify(updloadStrategy, times(1)).apply(firstBatch); - verify(updloadStrategy, times(1)).apply(secondBatch); + verify(uploadStrategy, times(2)).apply(any(BatchUpdate.class)); + verify(uploadStrategy, times(1)).apply(firstBatch); + verify(uploadStrategy, times(1)).apply(secondBatch); } @Test @@ -187,6 +187,6 @@ public void testAddingEmptyDocument() throws IOException, InterruptedException { queue.add(nullDocument); queue.flush(); - verify(updloadStrategy, times(0)).apply(any(BatchUpdate.class)); + verify(uploadStrategy, times(0)).apply(any(BatchUpdate.class)); } } From 40a73ddbb91a7171ffca1f239e53c281bf2e1a58 Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Tue, 30 May 2023 16:15:06 -0400 Subject: [PATCH 21/23] add unit tests --- .../com/coveo/pushapiclient/PushService.java | 4 +- .../PushServiceInternalTest.java | 75 +++++++++++++++++++ 2 files changed, 77 insertions(+), 2 deletions(-) create mode 100644 src/test/java/com/coveo/pushapiclient/PushServiceInternalTest.java diff --git a/src/main/java/com/coveo/pushapiclient/PushService.java b/src/main/java/com/coveo/pushapiclient/PushService.java index 8adc5f07..185203f0 100644 --- a/src/main/java/com/coveo/pushapiclient/PushService.java +++ b/src/main/java/com/coveo/pushapiclient/PushService.java @@ -14,7 +14,7 @@ public PushService(PushEnabledSource source) { String apiKey = source.getApiKey(); String organizationId = source.getOrganizationId(); PlatformUrl platformUrl = source.getPlatformUrl(); - UpdloadStrategy uploader = this.getUploadStrategy(); + UploadStrategy uploader = this.getUploadStrategy(); DocumentUploadQueue queue = new DocumentUploadQueue(uploader); this.platformClient = new PlatformClient(apiKey, organizationId, platformUrl); @@ -35,7 +35,7 @@ public void close() throws IOException, InterruptedException { this.service.close(); } - private UpdloadStrategy getUploadStrategy() { + private UploadStrategy getUploadStrategy() { return (batchUpdate) -> { String sourceId = this.getSourceId(); HttpResponse resFileContainer = this.platformClient.createFileContainer(); 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 From b5bbee0fd19a706931a2991f6bd0828a00412606 Mon Sep 17 00:00:00 2001 From: Yassine Date: Fri, 2 Jun 2023 09:02:26 -0400 Subject: [PATCH 22/23] Apply suggestions from code review Co-authored-by: Mohan Raj Rajamanickam <128537068+mrrajamanickam-coveo@users.noreply.github.com> --- .../java/com/coveo/pushapiclient/DocumentUploadQueue.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index 79014137..9606f800 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -68,11 +68,11 @@ public void add(DocumentBuilder document) throws IOException, InterruptedExcepti } /** - * 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. */ From e429d3e8a1fa47f32758081eae17eec44705ddda Mon Sep 17 00:00:00 2001 From: ylakhdar Date: Fri, 2 Jun 2023 09:33:46 -0400 Subject: [PATCH 23/23] applying corrections --- .../coveo/pushapiclient/DocumentUploadQueue.java | 12 ++++-------- .../pushapiclient/DocumentUploadQueueTest.java | 14 +++++++++++--- 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java index 9606f800..66355664 100644 --- a/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java +++ b/src/main/java/com/coveo/pushapiclient/DocumentUploadQueue.java @@ -61,10 +61,8 @@ 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; } /** @@ -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/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