feat: enable push on catalog source (#82)

https://coveord.atlassian.net/browse/CDX-1464

Author: y-lakhdar

samples/StreamDocuments.java

@@ -11,7 +11,10 @@ public static void main(String[] args) throws IOException, InterruptedException,
         PlatformUrl platformUrl = new PlatformUrlBuilder().withEnvironment(Environment.PRODUCTION).withRegion(Region.US).build();
         CatalogSource catalogSource = CatalogSource.fromPlatformUrl("my_api_key","my_org_id","my_source_id", platformUrl);
 
+        // Using the Stream Service will act as a source rebuild, therefore any currently indexed items not contained in the payload will be deleted.
         StreamService streamService = new StreamService(catalogSource);
+        // To perform full document updates, use the PushService instead.
+        // For more info, visit: https://docs.coveo.com/en/l62e0540/coveo-for-commerce/how-to-update-your-catalog#full-document-updates
 
         DocumentBuilder document1 = new DocumentBuilder("https://my.document.uri", "My document title")
                 .withData("these words will be searchable")

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

@@ -5,7 +5,7 @@
 import java.net.URL;
 import java.net.http.HttpResponse;
 
-public class CatalogSource implements StreamEnabledSource {
+public class CatalogSource implements StreamEnabledSource, PushEnabledSource {
   private final String apiKey;
   private final ApiUrl urlExtractor;
 

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

@@ -444,6 +444,29 @@ public HttpResponse<String> pushFileContainerContent(String sourceId, FileContai
     return this.api.put(uri, headers, HttpRequest.BodyPublishers.ofString(""));
   }
 
+  /**
+   * Push a file container into a stream source. See [Push the File Container into a Stream
+   * Source](https://docs.coveo.com/en/l62e0540/coveo-for-commerce/how-to-update-your-catalog#step-3-send-the-file-container-to-update-your-catalog).
+   *
+   * @param sourceId
+   * @param fileContainer
+   * @return
+   * @throws IOException
+   * @throws InterruptedException
+   */
+  public HttpResponse<String> pushFileContainerContentToStreamSource(
+      String sourceId, FileContainer fileContainer) throws IOException, InterruptedException {
+    String[] headers =
+        this.getHeaders(this.getAuthorizationHeader(), this.getContentTypeApplicationJSONHeader());
+    URI uri =
+        URI.create(
+            this.getBasePushURL()
+                + String.format(
+                    "/sources/%s/stream/update?fileId=%s", sourceId, fileContainer.fileId));
+
+    return this.api.put(uri, headers, HttpRequest.BodyPublishers.ofString(""));
+  }
+
   /**
    * Push a binary to a File Container.
    *

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

@@ -387,6 +387,25 @@ public void testPushFileContainerContent() throws IOException, InterruptedExcept
     assertAuthorizationHeader();
   }
 
+  @Test
+  public void testPushFileContainerContentToStream() throws IOException, InterruptedException {
+    client.pushFileContainerContentToStreamSource("my_source", fileContainer());
+    verify(httpClient)
+        .send(argument.capture(), any(HttpResponse.BodyHandlers.ofString().getClass()));
+
+    assertEquals("PUT", argument.getValue().method());
+    assertTrue(
+        argument.getValue().uri().getPath().contains("the_org_id/sources/my_source/stream/update"));
+    assertTrue(
+        argument
+            .getValue()
+            .uri()
+            .getQuery()
+            .contains(String.format("fileId=%s", fileContainer().fileId)));
+    assertApplicationJsonHeader();
+    assertAuthorizationHeader();
+  }
+
   @Test
   public void testOpenStream() throws IOException, InterruptedException {
     client.openStream("my_source");