Merge pull request #60 from ensi-platform/v7-cloudtech-3

V7 async search

Author: MsNatali

src/Aggregating/AggregationsQuery.php

@@ -7,6 +7,7 @@
 use Ensi\LaravelElasticQuery\Contracts\AggregationsBuilder;
 use Ensi\LaravelElasticQuery\Contracts\SearchIndex;
 use Ensi\LaravelElasticQuery\Filtering\BoolQueryBuilder;
+use GuzzleHttp\Ring\Future\FutureArray;
 use Illuminate\Support\Collection;
 
 class AggregationsQuery implements AggregationsBuilder
@@ -29,18 +30,31 @@ public function composite(Closure $callback): static
         return $this;
     }
 
-    public function get(): Collection
+    public function get(?callable $async = null): Collection|FutureArray
     {
         if ($this->aggregations->isEmpty()) {
             return new Collection();
         }
 
-        $response = $this->execute();
+        if ($async) {
+            /** @var FutureArray $promise */
+            $promise = $this->execute(async: true);
+
+            $promise->then(function (array $response) use ($async) {
+                $async($this->responseToResults($response));
+            });
+
+            return $promise;
+        } else {
+            $response = $this->execute(async: false);
+
+            return $this->responseToResults($response);
+        }
 
         return $this->aggregations->parseResults($response['aggregations'] ?? []);
     }
 
-    protected function execute(): array
+    protected function execute(bool $async = false): array|FutureArray
     {
         $dsl = [
             'size' => 0,
@@ -49,6 +63,11 @@ protected function execute(): array
             'aggs' => $this->aggregations->toDSL(),
         ];
 
-        return $this->index->search($dsl);
+        return $async ? $this->index->searchAsync($dsl) : $this->index->search($dsl);
+    }
+
+    protected function responseToResults($response): Collection
+    {
+        return $this->aggregations->parseResults($response['aggregations'] ?? []);
     }
 }

src/Concerns/InteractsWithIndex.php

@@ -8,6 +8,7 @@
 use Ensi\LaravelElasticQuery\Search\SearchQuery;
 use Ensi\LaravelElasticQuery\Suggesting\SuggestQuery;
 use Exception;
+use GuzzleHttp\Ring\Future\FutureArray;
 
 trait InteractsWithIndex
 {
@@ -33,6 +34,14 @@ public function search(array $dsl): array
         return $this->resolveClient()->search($this->indexName(), $dsl);
     }
 
+    /**
+     * @see SearchIndex::search()
+     */
+    public function searchAsync(array $dsl): FutureArray
+    {
+        return $this->resolveClient()->searchAsync($this->indexName(), $dsl);
+    }
+
     /**
      * @see SearchIndex::search()
      */

src/ElasticClient.php

@@ -6,6 +6,7 @@
 use Elasticsearch\ClientBuilder;
 use Ensi\LaravelElasticQuery\Debug\QueryLog;
 use Ensi\LaravelElasticQuery\Debug\QueryLogRecord;
+use GuzzleHttp\Ring\Future\FutureArray;
 use Illuminate\Support\Collection;
 
 class ElasticClient
@@ -26,6 +27,16 @@ public function search(string $indexName, array $dsl): array
         ]);
     }
 
+    public function searchAsync(string $indexName, array $dsl): FutureArray
+    {
+        $this->queryLog?->log($indexName, $dsl);
+
+        return $this->client->search($this->paramsAsync([
+            'index' => $indexName,
+            'body' => $dsl,
+        ]));
+    }
+
     public function deleteByQuery(string $indexName, array $dsl): array
     {
         $this->queryLog?->log($indexName, $dsl);
@@ -133,4 +144,11 @@ public static function fromConfig(array $config): static
 
         return new static($builder->build());
     }
+
+    protected function paramsAsync(array $params): array
+    {
+        return array_merge_recursive($params, [
+            'client' => ['future' => 'lazy'],
+        ]);
+    }
 }

src/Search/SearchQuery.php

@@ -17,6 +17,7 @@
 use Ensi\LaravelElasticQuery\Search\Collapsing\Collapse;
 use Ensi\LaravelElasticQuery\Search\Sorting\SortBuilder;
 use Ensi\LaravelElasticQuery\Search\Sorting\SortCollection;
+use GuzzleHttp\Ring\Future\FutureArray;
 use Illuminate\Support\Collection;
 use InvalidArgumentException;
 use Webmozart\Assert\Assert;
@@ -54,17 +55,32 @@ public function get(): Collection
         return $this->parseHits($response);
     }
 
-    public function paginate(int $size, int $offset = 0): Page
+    public function paginate(int $size, int $offset = 0, ?callable $async = null): Page|FutureArray
     {
         Assert::greaterThan($size, 0);
         Assert::greaterThanEq($offset, 0);
 
-        $response = $this->execute(size: $size, from: $offset, totals: true);
-        $hits = $this->parseHits($response);
+        if ($async) {
+            /** @var FutureArray $promise */
+            $promise = $this->execute(size: $size, from: $offset, totals: true, async: true);
+
+            $promise->then(function (array $response) use ($size, $offset, $async) {
+                $async($this->responseToPage($size, $offset, $response));
+            });
+
+            return $promise;
+        } else {
+            $response = $this->execute(size: $size, from: $offset, totals: true, async: false);
 
+            return $this->responseToPage($size, $offset, $response);
+        }
+    }
+
+    protected function responseToPage(int $size, int $offset, array $response): Page
+    {
         return new Page(
             $size,
-            $hits,
+            $this->parseHits($response),
             aggs: $this->aggregations?->parseResults($response['aggregations'] ?? []),
             offset: $offset,
             total: data_get($response, 'hits.total.value', 0)
@@ -122,8 +138,9 @@ protected function execute(
         ?int $from = null,
         bool $totals = false,
         bool $source = true,
-        ?Cursor $cursor = null
-    ): array {
+        ?Cursor $cursor = null,
+        bool $async = false,
+    ): array|FutureArray {
         $dsl = [
             'size' => $size,
             'from' => $from,
@@ -150,7 +167,9 @@ protected function execute(
             $dsl['search_after'] = $cursor->toDSL();
         }
 
-        return $this->index->search(array_filter($dsl));
+        $dsl = array_filter($dsl);
+
+        return $async ? $this->index->searchAsync($dsl) : $this->index->search($dsl);
     }
 
     protected function sourceToDSL(bool $source): array | bool