Implement methods for CSV and Ndjson batch adds

src/Endpoints/Delegates/HandlesDocuments.php

@@ -52,11 +52,31 @@ public function addDocumentsNdjson(string $documents, ?string $primaryKey = null
         return $this->http->post(self::PATH.'/'.$this->uid.'/documents', $documents, ['primaryKey' => $primaryKey], 'application/x-ndjson');
     }
 
+    public function addDocumentsNdjsonInBatches(string $documents, ?int $batchSize = 1000, ?string $primaryKey = null)
+    {
+        $promises = [];
+        foreach (self::batchNdjsonString($documents, $batchSize) as $batch) {
+            $promises[] = $this->addDocumentsNdjson($batch, $primaryKey);
+        }
+
+        return $promises;
+    }
+
     public function addDocumentsCsv(string $documents, ?string $primaryKey = null)
     {
         return $this->http->post(self::PATH.'/'.$this->uid.'/documents', $documents, ['primaryKey' => $primaryKey], 'text/csv');
     }
 
+    public function addDocumentsCsvInBatches(string $documents, ?int $batchSize = 1000, ?string $primaryKey = null)
+    {
+        $promises = [];
+        foreach (self::batchCsvString($documents, $batchSize) as $batch) {
+            $promises[] = $this->addDocumentsCsv($batch, $primaryKey);
+        }
+
+        return $promises;
+    }
+
     public function updateDocuments(array $documents, ?string $primaryKey = null)
     {
         return $this->http->put(self::PATH.'/'.$this->uid.'/documents', $documents, ['primaryKey' => $primaryKey]);
@@ -100,6 +120,31 @@ private function assertValidDocumentId($documentId): void
         }
     }
 
+    private static function batchCsvString(string $documents, int $batchSize): Generator
+    {
+        $documents = preg_split("/\r\n|\n|\r/", trim($documents));
+        $csvHeader = $documents[0];
+        array_shift($documents);
+
+        $batches = array_chunk($documents, $batchSize);
+        foreach ($batches as $batch) {
+            array_unshift($batch, $csvHeader);
+            $batch = implode("\n", $batch);
+            yield $batch;
+        }
+    }
+
+    private static function batchNdjsonString(string $documents, int $batchSize): Generator
+    {
+        $documents = preg_split("/\r\n|\n|\r/", trim($documents));
+
+        $batches = array_chunk($documents, $batchSize);
+        foreach ($batches as $batch) {
+            $batch = implode("\n", $batch);
+            yield $batch;
+        }
+    }
+
     private static function batch(array $documents, int $batchSize): Generator
     {
         $batches = array_chunk($documents, $batchSize);

tests/Endpoints/DocumentsTest.php

@@ -256,6 +256,53 @@ public function testUpdateDocumentsInBatches(): void
         $this->assertCount(\count(self::DOCUMENTS), $response);
     }
 
+    public function testAddDocumentsCsvInBatches(): void
+    {
+        $index = $this->client->index('documentCsv');
+
+        $fileCsv = fopen('./tests/datasets/songs.csv', 'r');
+        $documentCsv = fread($fileCsv, filesize('./tests/datasets/songs.csv'));
+        fclose($fileCsv);
+
+        // Total number of lines excluding header
+        $total = \count(preg_split("/\r\n|\n|\r/", trim($documentCsv))) - 1;
+
+        $promises = $index->addDocumentsCsvInBatches($documentCsv, 250);
+
+        $this->assertCount(2, $promises);
+
+        foreach ($promises as $promise) {
+            $this->assertIsValidPromise($promise);
+            $index->waitForTask($promise['taskUid']);
+        }
+
+        $response = $index->getDocuments();
+        $this->assertSame($total, $response->getTotal());
+    }
+
+    public function testAddDocumentsNdjsonInBatches(): void
+    {
+        $index = $this->client->index('documentNdJson');
+
+        $fileNdJson = fopen('./tests/datasets/songs.ndjson', 'r');
+        $documentNdJson = fread($fileNdJson, filesize('./tests/datasets/songs.ndjson'));
+        fclose($fileNdJson);
+
+        $total = \count(preg_split("/\r\n|\n|\r/", trim($documentNdJson)));
+
+        $promises = $index->addDocumentsNdjsonInBatches($documentNdJson, 150);
+
+        $this->assertCount(2, $promises);
+
+        foreach ($promises as $promise) {
+            $this->assertIsValidPromise($promise);
+            $index->waitForTask($promise['taskUid']);
+        }
+
+        $response = $index->getDocuments();
+        $this->assertSame($total, $response->getTotal());
+    }
+
     public function testAddWithUpdateDocuments(): void
     {
         $index = $this->createEmptyIndex($this->safeIndexName('movies'));