Feat: TaskInfo#await

README.md

@@ -89,10 +89,10 @@ meili-search.api-key=MASTER_KEY
 
 ## Usage
 
-#### Reactive(reactor)
+### Searching
 
 ```java
-public static void main(String[] args) {
+public static void reactive() {
     JsonHandler jsonHandler = new JacksonJsonHandler();
     @Cleanup
     ReactiveMSearchClient client = ReactiveMSearchClient.create(builder -> builder
@@ -117,39 +117,8 @@ public static void main(String[] args) {
         )
     ).block();
 }
-```
-
-##### Search With Details API
-
-```java
-ReactiveTypedDetailsSearch<Movie> searcher = client.indexes().searchWithDetails("movies", Movie.class);
-searcher.find(builder ->builder
-    .q("hello world")
-    .filter("id < 10")
-    .showRankingScore(true)
-)
-    .map(response -> {
-        Integer estimatedTotalHits = response.getEstimatedTotalHits();
-        Long processingTimeMs = response.getProcessingTimeMs();
-        return response.getHits();
-    })
-    .doOnNext(hitDetails -> {
-        for(HitDetails<Movie> hitDetail :hitDetails){
-            SearchDetails details = hitDetail.getDetails(); // Search Details
-            Movie source = hitDetail.getSource();           // Source Document
-
-            Map<String, Object> formatted = details.get_formatted();
-            SearchDetails.Geo geo = details.get_geo();
-            Double rankingScore = details.get_rankingScore();
-        }
-    })
-    .subscribe()
-```
-
-#### Blocking
 
-```java
-public static void main(String[] args) {
+public static void blocking() {
     JsonHandler jsonHandler = new JacksonJsonHandler();
     @Cleanup
     MSearchClient client = MSearchClient.create(builder -> builder
@@ -176,24 +145,63 @@ public static void main(String[] args) {
 }
 ```
 
-##### Search With Details API
+### Search With Details API
+
+```java
+public static void reactive() {
+    ReactiveTypedDetailsSearch<Movie> searcher = client.indexes().searchWithDetails("movies", Movie.class);
+    searcher.find(builder ->builder
+            .q("hello world")
+            .filter("id < 10")
+            .showRankingScore(true)
+        )
+        .map(response -> {
+            Integer estimatedTotalHits = response.getEstimatedTotalHits();
+            Long processingTimeMs = response.getProcessingTimeMs();
+            return response.getHits();
+        })
+        .doOnNext(hitDetails -> {
+            for(HitDetails<Movie> hitDetail :hitDetails){
+                SearchDetails details = hitDetail.getDetails(); // Search Details
+                Movie source = hitDetail.getSource();           // Source Document
+
+                Map<String, Object> formatted = details.get_formatted();
+                SearchDetails.Geo geo = details.get_geo();
+                Double rankingScore = details.get_rankingScore();
+            }
+        })
+        .subscribe()
+}
+
+public static void blocking() {
+    TypedDetailsSearch<Movie> searcher = client.indexes().searchWithDetails("movies", Movie.class);
+    SearchDetailsResponse<Movie> response = searcher.find(builder -> builder
+        .q("hello world")
+        .filter("id < 10")
+        .showRankingScore(true)
+    );
+    Integer estimatedTotalHits = response.getEstimatedTotalHits();
+    Long processingTimeMs = response.getProcessingTimeMs();
+    List<HitDetails<Movie>> hitDetails = response.getHits();
+    for(HitDetails<Movie> hitDetail :hitDetails){
+        SearchDetails details = hitDetail.getDetails(); // Search Details
+        Movie source = hitDetail.getSource();           // Source Document
+
+        Map<String, Object> formatted = details.get_formatted();
+        SearchDetails.Geo geo = details.get_geo();
+        Double rankingScore = details.get_rankingScore();
+    }
+}
+```
+
+### Await On TaskInfo, Only in blocking
 
 ```java
-TypedDetailsSearch<Movie> searcher = client.indexes().searchWithDetails("movies", Movie.class);
-SearchDetailsResponse<Movie> response = searcher.find(builder -> builder
-    .q("hello world")
-    .filter("id < 10")
-    .showRankingScore(true)
-);
-Integer estimatedTotalHits = response.getEstimatedTotalHits();
-Long processingTimeMs = response.getProcessingTimeMs();
-List<HitDetails<Movie>> hitDetails = response.getHits();
-for(HitDetails<Movie> hitDetail :hitDetails){
-    SearchDetails details = hitDetail.getDetails(); // Search Details
-    Movie source = hitDetail.getSource();           // Source Document
-
-    Map<String, Object> formatted = details.get_formatted();
-    SearchDetails.Geo geo = details.get_geo();
-    Double rankingScore = details.get_rankingScore();
+public static void blocking() {
+    TaskInfo saveTask = client.indexes(indexes -> indexes
+        .documents("movies").save(jsonQuote("{'id':100}")));
+    TaskView taskView = saveTask.await();
+    TaskStatus status = taskView.getStatus();
+    assert status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED;
 }
 ```

src/main/java/io/github/honhimw/ms/internal/reactive/AbstractReactiveImpl.java

@@ -18,12 +18,20 @@
 import io.github.honhimw.ms.http.ReactiveHttpUtils;
 import io.github.honhimw.ms.json.JsonHandler;
 import io.github.honhimw.ms.json.TypeRef;
+import io.github.honhimw.ms.model.TaskInfo;
+import io.github.honhimw.ms.model.TaskStatus;
+import io.github.honhimw.ms.model.TaskType;
+import io.github.honhimw.ms.model.TaskView;
+import io.github.honhimw.ms.support.ReactorUtils;
+import io.github.honhimw.ms.support.TypeRefs;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
 
 import java.nio.charset.Charset;
+import java.time.Duration;
+import java.time.LocalDateTime;
 import java.util.Objects;
 import java.util.function.Consumer;
 
@@ -140,7 +148,8 @@
                 } else {
                     return false;
                 }
-            }, throwable -> Mono.empty());
+            }, throwable -> Mono.empty())
+            .map(t -> decorate(t, typeRef));
     }
 
     protected void json(ReactiveHttpUtils.Configurer configurer, Object object) {
@@ -151,4 +160,82 @@
         configurer.body(payload -> payload.raw(raw -> raw.json(json)));
     }
 
+    @SuppressWarnings("unchecked")
+    protected <T> T decorate(T delegate, TypeRef<T> typeRef) {
+        if (typeRef == TypeRefs.TaskInfoRef.INSTANCE) {
+            return (T) delegateTaskInfo((TaskInfo) delegate);
+        }
+        return delegate;
+    }
+
+    protected TaskInfo delegateTaskInfo(TaskInfo taskInfo) {
+        return new DelegateTaskInfo(taskInfo) {
+            @Override
+            public TaskView await(Duration duration) {
+                Mono<TaskView> tasks = _client.tasks(reactiveTasks -> reactiveTasks
+                    .await(this, _client.config.getAwaitAttempts(), _client.config.getAwaitFixedDelay(), duration));
+                return ReactorUtils.blockNonNull(tasks);
+            }
+        };
+    }
+
+    private static abstract class DelegateTaskInfo extends TaskInfo {
+        private final TaskInfo taskInfo;
+
+        public DelegateTaskInfo(TaskInfo taskInfo) {
+            this.taskInfo = taskInfo;
+        }
+
+        @Override
+        public Integer getTaskUid() {
+            return taskInfo.getTaskUid();
+        }
+
+        @Override
+        public String getIndexUid() {
+            return taskInfo.getIndexUid();
+        }
+
+        @Override
+        public TaskStatus getStatus() {
+            return taskInfo.getStatus();
+        }
+
+        @Override
+        public TaskType getType() {
+            return taskInfo.getType();
+        }
+
+        @Override
+        public LocalDateTime getEnqueuedAt() {
+            return taskInfo.getEnqueuedAt();
+        }
+
+        @Override
+        public void setTaskUid(Integer taskUid) {
+            taskInfo.setTaskUid(taskUid);
+        }
+
+        @Override
+        public void setIndexUid(String indexUid) {
+            taskInfo.setIndexUid(indexUid);
+        }
+
+        @Override
+        public void setStatus(TaskStatus status) {
+            taskInfo.setStatus(status);
+        }
+
+        @Override
+        public void setType(TaskType type) {
+            taskInfo.setType(type);
+        }
+
+        @Override
+        public void setEnqueuedAt(LocalDateTime enqueuedAt) {
+            taskInfo.setEnqueuedAt(enqueuedAt);
+        }
+    }
+
+
 }

src/main/java/io/github/honhimw/ms/model/TaskInfo.java

@@ -14,13 +14,15 @@
 
 package io.github.honhimw.ms.model;
 
+import io.github.honhimw.ms.json.JsonHandler;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
 
 import java.io.Serializable;
+import java.time.Duration;
 import java.time.LocalDateTime;
 
 /**
@@ -68,4 +70,18 @@
     @Schema(description = "Represents the date and time in the RFC 3339 format when the task has been enqueued")
     private LocalDateTime enqueuedAt;
 
+    /**
+     * Only used in non-reactive context.
+     */
+    public TaskView await() {
+        return this.await(Duration.ofDays(1));
+    }
+
+    /**
+     * Only used in non-reactive context.
+     */
+    public TaskView await(Duration duration) {
+        throw new UnsupportedOperationException("Not Supported");
+    }
+
 }

src/test/java/io/github/honhimw/ms/client/TaskTests.java

@@ -17,6 +17,7 @@
 import io.github.honhimw.ms.api.Documents;
 import io.github.honhimw.ms.api.Tasks;
 import io.github.honhimw.ms.model.TaskInfo;
+import io.github.honhimw.ms.model.TaskStatus;
 import io.github.honhimw.ms.model.TaskView;
 import io.github.honhimw.ms.support.TestSupport;
 import org.junit.jupiter.api.MethodOrderer;
@@ -46,6 +47,15 @@ void taskView() {
         assert Objects.nonNull(taskView.getUid());
     }
 
+
+    @Order(1)
+    @Test
+    void awaitOnTaskInfo() {
+        TaskInfo saveTask = blockingClient.indexes(indexes -> indexes.documents(INDEX).save(TestSupport.jsonQuote("{'id':100}")));
+        TaskStatus status = saveTask.await().getStatus();
+        assert status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED;
+    }
+
     @Test
     void timeout() {
         TestSupport.assertError(() -> getBlockingTasks().await(1, 1, Duration.ofSeconds(1), Duration.ZERO));