Add support to retrieve invocations from idempotency key. Fix #358

restatedev · Jul 18, 2024 · 5f0caa5 · 5f0caa5
1 parent cd23a59
commit 5f0caa5
Show file tree

Hide file tree

Showing 3 changed files with 200 additions and 151 deletions.
diff --git a/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/ingress.kt b/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/ingress.kt
@@ -14,9 +14,9 @@ import dev.restate.sdk.client.SendResponse
 import dev.restate.sdk.common.Output
 import dev.restate.sdk.common.Serde
 import dev.restate.sdk.common.Target
+import kotlinx.coroutines.future.await
 import kotlin.time.Duration
 import kotlin.time.toJavaDuration
-import kotlinx.coroutines.future.await
 
 // Extension methods for the IngressClient
 
@@ -67,6 +67,18 @@ suspend fun <T : Any?> Client.InvocationHandle<T>.getOutputSuspend(
   return this.getOutputAsync(options).await()
 }
 
+suspend fun <T> Client.IdempotentInvocationHandle<T>.attachSuspend(
+    options: RequestOptions = RequestOptions.DEFAULT
+): T {
+  return this.attachAsync(options).await()
+}
+
+suspend fun <T> Client.IdempotentInvocationHandle<T>.getOutputSuspend(
+    options: RequestOptions = RequestOptions.DEFAULT
+): Output<T> {
+  return this.getOutputAsync(options).await()
+}
+
 suspend fun <T> Client.WorkflowHandle<T>.attachSuspend(
     options: RequestOptions = RequestOptions.DEFAULT
 ): T {

diff --git a/sdk-common/src/main/java/dev/restate/sdk/client/Client.java b/sdk-common/src/main/java/dev/restate/sdk/client/Client.java
@@ -211,6 +211,54 @@ default Output<Res> getOutput() throws IngressException {
     }
   }
 
+  <Res> IdempotentInvocationHandle<Res> idempotentInvocationHandle(
+      Target target, String idempotencyKey, Serde<Res> resSerde);
+
+  interface IdempotentInvocationHandle<Res> {
+
+    CompletableFuture<Res> attachAsync(RequestOptions options);
+
+    default CompletableFuture<Res> attachAsync() {
+      return attachAsync(RequestOptions.DEFAULT);
+    }
+
+    default Res attach(RequestOptions options) throws IngressException {
+      try {
+        return attachAsync(options).join();
+      } catch (CompletionException e) {
+        if (e.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) e.getCause();
+        }
+        throw new RuntimeException(e.getCause());
+      }
+    }
+
+    default Res attach() throws IngressException {
+      return attach(RequestOptions.DEFAULT);
+    }
+
+    CompletableFuture<Output<Res>> getOutputAsync(RequestOptions options);
+
+    default CompletableFuture<Output<Res>> getOutputAsync() {
+      return getOutputAsync(RequestOptions.DEFAULT);
+    }
+
+    default Output<Res> getOutput(RequestOptions options) throws IngressException {
+      try {
+        return getOutputAsync(options).join();
+      } catch (CompletionException e) {
+        if (e.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) e.getCause();
+        }
+        throw new RuntimeException(e.getCause());
+      }
+    }
+
+    default Output<Res> getOutput() throws IngressException {
+      return getOutput(RequestOptions.DEFAULT);
+    }
+  }
+
   <Res> WorkflowHandle<Res> workflowHandle(
       String workflowName, String workflowId, Serde<Res> resSerde);