Client factory spi (#486)

* Add default Serde factory provider

* Improve suspend overloads for kotlin clients

* Spotless apply

Author: slinkydeveloper

client-kotlin/src/main/kotlin/dev/restate/client/kotlin/ingress.kt

@@ -15,7 +15,8 @@ import dev.restate.client.SendResponse
 import dev.restate.common.Output
 import dev.restate.common.Request
 import dev.restate.common.WorkflowRequest
-import dev.restate.serde.Serde
+import dev.restate.serde.TypeTag
+import dev.restate.serde.kotlinx.typeTag
 import kotlin.time.Duration
 import kotlin.time.toJavaDuration
 import kotlinx.coroutines.future.await
@@ -71,11 +72,18 @@ suspend fun <Req, Res> Client.submitSuspend(
 }
 
 suspend fun <T : Any> Client.AwakeableHandle.resolveSuspend(
-    serde: Serde<T>,
+    typeTag: TypeTag<T>,
     payload: T,
     options: ClientRequestOptions = ClientRequestOptions.DEFAULT
 ): ClientResponse<Void> {
-  return this.resolveAsync(serde, payload, options).await()
+  return this.resolveAsync(typeTag, payload, options).await()
+}
+
+suspend inline fun <reified T : Any> Client.AwakeableHandle.resolveSuspend(
+    payload: T,
+    options: ClientRequestOptions = ClientRequestOptions.DEFAULT
+): ClientResponse<Void> {
+  return this.resolveSuspend(typeTag<T>(), payload, options)
 }
 
 suspend fun Client.AwakeableHandle.rejectSuspend(

client/src/main/java/dev/restate/client/Client.java

@@ -12,7 +12,6 @@
 import dev.restate.common.Request;
 import dev.restate.common.Target;
 import dev.restate.common.WorkflowRequest;
-import dev.restate.serde.Serde;
 import dev.restate.serde.SerdeFactory;
 import dev.restate.serde.TypeTag;
 import java.time.Duration;
@@ -123,6 +122,34 @@ default <T> ClientResponse<Void> resolve(
       }
     }
 
+    /**
+     * Complete with success the Awakeable.
+     *
+     * @param clazz used to serialize the Awakeable result payload.
+     * @param payload the result payload. MUST NOT be null.
+     */
+    default <T> ClientResponse<Void> resolve(Class<T> clazz, @NonNull T payload) {
+      return this.resolve(TypeTag.of(clazz), payload, ClientRequestOptions.DEFAULT);
+    }
+
+    /** Same as {@link #resolve(Class, Object)} but async with options. */
+    default <T> CompletableFuture<ClientResponse<Void>> resolveAsync(
+        Class<T> clazz, @NonNull T payload, ClientRequestOptions options) {
+      return this.resolveAsync(TypeTag.of(clazz), payload, options);
+    }
+
+    /** Same as {@link #resolve(TypeTag, Object)} but async. */
+    default <T> CompletableFuture<ClientResponse<Void>> resolveAsync(
+        Class<T> clazz, @NonNull T payload) {
+      return resolveAsync(TypeTag.of(clazz), payload, ClientRequestOptions.DEFAULT);
+    }
+
+    /** Same as {@link #resolve(TypeTag, Object)} with options. */
+    default <T> ClientResponse<Void> resolve(
+        Class<T> clazz, @NonNull T payload, ClientRequestOptions options) {
+      return resolve(TypeTag.of(clazz), payload, options);
+    }
+
     /**
      * Complete with success the Awakeable.
      *
@@ -331,7 +358,7 @@ default ClientResponse<Output<Res>> getOutput() throws IngressException {
    * @param baseUri uri to connect to.
    */
   static Client connect(String baseUri) {
-    return connect(baseUri, SerdeFactory.NOOP, ClientRequestOptions.DEFAULT);
+    return connect(baseUri, null, null);
   }
 
   /**
@@ -341,15 +368,14 @@ static Client connect(String baseUri) {
    * @param options default options to use in all the requests.
    */
   static Client connect(String baseUri, ClientRequestOptions options) {
-    return connect(baseUri, SerdeFactory.NOOP, options);
+    return connect(baseUri, null, options);
   }
 
   /**
    * Create a default JDK client.
    *
    * @param baseUri uri to connect to
-   * @param serdeFactory Serde factory to use. You must provide this when the provided {@link
-   *     TypeTag} are not {@link Serde} instances. If you're just wrapping this client in a
+   * @param serdeFactory Serde factory to use. If you're just wrapping this client in a
    *     code-generated client, you don't need to provide this parameter.
    */
   static Client connect(String baseUri, SerdeFactory serdeFactory) {
@@ -360,8 +386,7 @@ static Client connect(String baseUri, SerdeFactory serdeFactory) {
    * Create a default JDK client.
    *
    * @param baseUri uri to connect to
-   * @param serdeFactory Serde factory to use. You must provide this when the provided {@link
-   *     TypeTag} are not {@link Serde} instances. If you're just wrapping this client in a
+   * @param serdeFactory Serde factory to use. If you're just wrapping this client in a
    *     code-generated client, you don't need to provide this parameter.
    * @param options default options to use in all the requests.
    */

client/src/main/java/dev/restate/client/base/BaseClient.java

@@ -15,7 +15,9 @@
 import dev.restate.common.*;
 import dev.restate.serde.Serde;
 import dev.restate.serde.SerdeFactory;
+import dev.restate.serde.TypeRef;
 import dev.restate.serde.TypeTag;
+import dev.restate.serde.provider.DefaultSerdeFactoryProvider;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -26,6 +28,8 @@
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Stream;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.jetbrains.annotations.NotNull;
 import org.jspecify.annotations.NonNull;
 import org.jspecify.annotations.Nullable;
@@ -42,13 +46,19 @@ public abstract class BaseClient implements Client {
   private final SerdeFactory serdeFactory;
   private final ClientRequestOptions baseOptions;
 
-  protected BaseClient(URI baseUri, SerdeFactory serdeFactory, ClientRequestOptions baseOptions) {
+  protected BaseClient(
+      URI baseUri,
+      @Nullable SerdeFactory serdeFactory,
+      @Nullable ClientRequestOptions baseOptions) {
     this.baseUri = Objects.requireNonNull(baseUri, "Base uri cannot be null");
     if (!this.baseUri.isAbsolute()) {
       throw new IllegalArgumentException(
           "The base uri " + baseUri + " is not absolute. This is not supported.");
     }
-    this.serdeFactory = serdeFactory == null ? SerdeFactory.NOOP : serdeFactory;
+    this.serdeFactory =
+        serdeFactory == null
+            ? DefaultSerdeFactorySingleton.INSTANCE.getLoadedFactory()
+            : serdeFactory;
     this.baseOptions = baseOptions == null ? ClientRequestOptions.DEFAULT : baseOptions;
   }
 
@@ -539,4 +549,50 @@ private static Map<String, String> findStringFieldsInJsonObject(
 
     return resultMap;
   }
+
+  // Machinery to load default serde factory
+
+  private static class DefaultSerdeFactorySingleton {
+    private static final DefaultSerdeFactory INSTANCE = new DefaultSerdeFactory();
+  }
+
+  public static final class DefaultSerdeFactory {
+
+    private static final Logger LOG = LogManager.getLogger(DefaultSerdeFactory.class);
+
+    private final SerdeFactory loadedFactory;
+
+    public DefaultSerdeFactory() {
+      var loadedFactories = ServiceLoader.load(DefaultSerdeFactoryProvider.class).stream().toList();
+      if (loadedFactories.size() == 1) {
+        this.loadedFactory = loadedFactories.get(0).get().create();
+      } else {
+        this.loadedFactory =
+            new SerdeFactory() {
+              @Override
+              public <T> Serde<T> create(TypeRef<T> typeRef) {
+                throw new UnsupportedOperationException(
+                    "No SerdeFactory class was configured. Please configure one, this is required when using TypeTag and Class in client methods.");
+              }
+
+              @Override
+              public <T> Serde<T> create(Class<T> clazz) {
+                throw new UnsupportedOperationException(
+                    "No SerdeFactory class was configured. Please configure one, this is required when using TypeTag and Class in client methods.");
+              }
+            };
+      }
+
+      if (loadedFactories.size() > 1) {
+        LOG.warn(
+            "When creating the Client, more than one SerdeFactory was found.\n"
+                + "To prevent unexpected behavior, the client was configured without SerdeFactory. "
+                + "Please manually provide the SerdeFactory (e.g. JacksonSerdeFactory or KotlinxSerializationSerdeFactory) of your choice in the Client.connect() factory methods.");
+      }
+    }
+
+    public SerdeFactory getLoadedFactory() {
+      return loadedFactory;
+    }
+  }
 }

client/src/main/java/dev/restate/client/jdk/JdkClient.java

@@ -116,14 +116,14 @@ public Map<String, String> toLowercaseMap() {
   public static JdkClient of(
       HttpClient httpClient,
       String baseUri,
-      SerdeFactory serdeFactory,
-      ClientRequestOptions options) {
+      @Nullable SerdeFactory serdeFactory,
+      @Nullable ClientRequestOptions options) {
     return new JdkClient(URI.create(baseUri), serdeFactory, options, httpClient);
   }
 
   /** Create a new JDK Client */
   public static JdkClient of(
-      String baseUri, SerdeFactory serdeFactory, ClientRequestOptions options) {
+      String baseUri, @Nullable SerdeFactory serdeFactory, @Nullable ClientRequestOptions options) {
     return new JdkClient(URI.create(baseUri), serdeFactory, options, HttpClient.newHttpClient());
   }
 }

common/src/main/java/dev/restate/serde/provider/DefaultSerdeFactoryProvider.java

@@ -0,0 +1,20 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.serde.provider;
+
+import dev.restate.serde.SerdeFactory;
+
+/**
+ * This class is used to autoload either JacksonSerdeFactory or KotlinxSerializationSerdeFactory in
+ * the client.
+ */
+public interface DefaultSerdeFactoryProvider {
+
+  SerdeFactory create();
+}

sdk-serde-jackson/src/main/java/dev/restate/serde/jackson/JacksonSerdeFactoryProvider.java

@@ -0,0 +1,19 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.serde.jackson;
+
+import dev.restate.serde.SerdeFactory;
+import dev.restate.serde.provider.DefaultSerdeFactoryProvider;
+
+public class JacksonSerdeFactoryProvider implements DefaultSerdeFactoryProvider {
+  @Override
+  public SerdeFactory create() {
+    return new JacksonSerdeFactory();
+  }
+}

sdk-serde-jackson/src/main/resources/META-INF/services/dev.restate.serde.provider.DefaultSerdeFactoryProvider

@@ -0,0 +1 @@
+dev.restate.serde.jackson.JacksonSerdeFactoryProvider

sdk-serde-kotlinx/src/main/kotlin/dev/restate/serde/kotlinx/KotlinSerializationSerdeFactoryProvider.kt

@@ -0,0 +1,15 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.serde.kotlinx
+
+import dev.restate.serde.provider.DefaultSerdeFactoryProvider
+
+public class KotlinSerializationSerdeFactoryProvider : DefaultSerdeFactoryProvider {
+  override fun create() = KotlinSerializationSerdeFactory()
+}

sdk-serde-kotlinx/src/main/resources/META-INF/services/dev.restate.serde.provider.DefaultSerdeFactoryProvider

@@ -0,0 +1 @@
+dev.restate.serde.kotlinx.KotlinSerializationSerdeFactoryProvider