chore: Cleanup JsonSupport

akka-javasdk/src/main/java/akka/javasdk/JsonSupport.java

@@ -5,10 +5,10 @@
 package akka.javasdk;
 
 import akka.Done;
-import akka.annotation.InternalApi;
 import akka.javasdk.annotations.Migration;
 import akka.javasdk.impl.AnySupport;
 import akka.javasdk.impl.ByteStringEncoding;
+import akka.runtime.sdk.spi.BytesPayload;
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.PropertyAccessor;
@@ -82,6 +82,8 @@ public static ObjectMapper getObjectMapper() {
     return objectMapper;
   }
 
+  private static akka.javasdk.impl.serialization.JsonSerializer jsonSerializer = new akka.javasdk.impl.serialization.JsonSerializer();
+
   private JsonSupport() {
   }
 
@@ -96,7 +98,10 @@ private JsonSupport() {
    * for the JSON type instead.
    *
    * @see {{encodeJson(T, String}}
+   *
+   * @deprecated Protobuf Any with JSON is not supported
    */
+  @Deprecated
   public static <T> Any encodeJson(T value) {
     return encodeJson(value, value.getClass().getName());
   }
@@ -111,7 +116,10 @@ public static <T> Any encodeJson(T value) {
    *                 JSON, useful for example when multiple different objects are passed through a pub/sub
    *                 topic.
    * @throws IllegalArgumentException if the given value cannot be turned into JSON
+   *
+   * @deprecated Protobuf Any with JSON is not supported
    */
+  @Deprecated
   public static <T> Any encodeJson(T value, String jsonType) {
     try {
       ByteString bytes = encodeToBytes(value);
@@ -123,26 +131,81 @@ public static <T> Any encodeJson(T value, String jsonType) {
     }
   }
 
-  // FIXME do we really want all these to be public API?
+  /**
+   * @deprecated Use encodeToAkkaByteString
+   */
+  @Deprecated
   public static <T> ByteString encodeToBytes(T value) throws JsonProcessingException {
     return UnsafeByteOperations.unsafeWrap(
       objectMapper.writerFor(value.getClass()).writeValueAsBytes(value));
   }
 
-  public static <T> akka.util.ByteString encodeToAkkaByteString(T value) throws JsonProcessingException {
-    return akka.util.ByteString.fromArrayUnsafe(objectMapper.writerFor(value.getClass()).writeValueAsBytes(value));
+  /**
+   * Encode the given value as JSON using Jackson.
+   *
+   * @param value    the object to encode as JSON, must be an instance of a class properly annotated
+   *                 with the needed Jackson annotations.
+   * @throws IllegalArgumentException if the given value cannot be turned into JSON
+   */
+  public static <T> akka.util.ByteString encodeToAkkaByteString(T value) {
+    try {
+      return akka.util.ByteString.fromArrayUnsafe(objectMapper.writerFor(value.getClass()).writeValueAsBytes(value));
+    } catch (JsonProcessingException ex) {
+      throw new IllegalArgumentException(
+          "Could not encode [" + value.getClass().getName() + "] as JSON", ex);
+    }
   }
 
-  public static akka.util.ByteString encodeDynamicToAkkaByteString(String key, String value) throws JsonProcessingException {
-    ObjectNode dynamicJson = objectMapper.createObjectNode().put(key, value);
-    return akka.util.ByteString.fromArrayUnsafe(objectMapper.writeValueAsBytes(dynamicJson));
+  public static akka.util.ByteString encodeDynamicToAkkaByteString(String key, String value) {
+    try {
+      ObjectNode dynamicJson = objectMapper.createObjectNode().put(key, value);
+      return akka.util.ByteString.fromArrayUnsafe(objectMapper.writeValueAsBytes(dynamicJson));
+    } catch (JsonProcessingException ex) {
+      throw new IllegalArgumentException(
+          "Could not encode dynamic key/value as JSON", ex);
+    }
   }
 
-  public static akka.util.ByteString encodeDynamicCollectionToAkkaByteString(String key, Collection<?> values) throws JsonProcessingException {
-    ObjectNode objectNode = objectMapper.createObjectNode();
-    ArrayNode dynamicJson = objectNode.putArray(key);
-    values.forEach(v -> dynamicJson.add(v.toString()));
-    return akka.util.ByteString.fromArrayUnsafe(objectMapper.writeValueAsBytes(objectNode));
+  public static akka.util.ByteString encodeDynamicCollectionToAkkaByteString(String key, Collection<?> values) {
+    try {
+      ObjectNode objectNode = objectMapper.createObjectNode();
+      ArrayNode dynamicJson = objectNode.putArray(key);
+      values.forEach(v -> dynamicJson.add(v.toString()));
+      return akka.util.ByteString.fromArrayUnsafe(objectMapper.writeValueAsBytes(objectNode));
+    } catch (JsonProcessingException ex) {
+      throw new IllegalArgumentException(
+          "Could not encode dynamic key/values as JSON", ex);
+    }
+  }
+
+  /**
+   * Decode the given bytes to an instance of T using Jackson. The bytes must be
+   * the JSON string as bytes.
+   *
+   * @param valueClass The type of class to deserialize the object to, the class must have the
+   *                   proper Jackson annotations for deserialization.
+   * @param bytes      The bytes to deserialize.
+   * @return The decoded object
+   * @throws IllegalArgumentException if the given value cannot be decoded to a T
+   *
+   */
+  public static <T> T decodeJson(Class<T> valueClass, akka.util.ByteString bytes) {
+    return jsonSerializer.fromBytes(valueClass, new BytesPayload(bytes, jsonSerializer.contentTypeFor(valueClass)));
+  }
+
+  /**
+   * Decode the given bytes to an instance of T using Jackson. The bytes must be
+   * the JSON string as bytes.
+   *
+   * @param valueClass The type of class to deserialize the object to, the class must have the
+   *                   proper Jackson annotations for deserialization.
+   * @param bytes      The bytes to deserialize.
+   * @return The decoded object
+   * @throws IllegalArgumentException if the given value cannot be decoded to a T
+   *
+   */
+  public static <T> T decodeJson(Class<T> valueClass, byte[] bytes) {
+    return decodeJson(valueClass, akka.util.ByteString.fromArrayUnsafe(bytes));
   }
 
   /**
@@ -154,7 +217,10 @@ public static akka.util.ByteString encodeDynamicCollectionToAkkaByteString(Strin
    * @param any        The protobuf Any object to deserialize.
    * @return The decoded object
    * @throws IllegalArgumentException if the given value cannot be decoded to a T
+   *
+   * @deprecated Protobuf Any with JSON is not supported
    */
+  @Deprecated
   public static <T> T decodeJson(Class<T> valueClass, Any any) {
     if (!AnySupport.isJsonTypeUrl(any.getTypeUrl())) {
       throw new IllegalArgumentException(
@@ -177,15 +243,15 @@ public static <T> T decodeJson(Class<T> valueClass, Any any) {
           if (fromVersion < currentVersion) {
             return migrate(valueClass, decodedBytes, fromVersion, migration);
           } else if (fromVersion == currentVersion) {
-            return parseBytes(decodedBytes.toByteArray(), valueClass);
+            return objectMapper.readValue(decodedBytes.toByteArray(), valueClass);
           } else if (fromVersion <= supportedForwardVersion) {
             return migrate(valueClass, decodedBytes, fromVersion, migration);
           } else {
             throw new IllegalStateException("Migration version " + supportedForwardVersion + " is " +
                 "behind version " + fromVersion + " of deserialized type [" + valueClass.getName() + "]");
           }
         } else {
-          return parseBytes(decodedBytes.toByteArray(), valueClass);
+          return objectMapper.readValue(decodedBytes.toByteArray(), valueClass);
         }
       } catch (JsonProcessingException e) {
         throw jsonProcessingException(valueClass, any, e);
@@ -196,6 +262,10 @@ public static <T> T decodeJson(Class<T> valueClass, Any any) {
     }
   }
 
+  /**
+   * @deprecated Use decodeJson
+   */
+  @Deprecated
   public static <T> T parseBytes(byte[] bytes, Class<T> valueClass) throws IOException {
     return objectMapper.readValue(bytes, valueClass);
   }
@@ -236,6 +306,11 @@ private static int parseVersion(String typeUrl) {
     }
   }
 
+
+  /**
+   * @deprecated Protobuf Any with JSON is not supported
+   */
+  @Deprecated
   public static <T, C extends Collection<T>> C decodeJsonCollection(Class<T> valueClass, Class<C> collectionType, Any any) {
     if (!AnySupport.isJsonTypeUrl(any.getTypeUrl())) {
       throw new IllegalArgumentException(
@@ -257,14 +332,25 @@ public static <T, C extends Collection<T>> C decodeJsonCollection(Class<T> value
     }
   }
 
+  public static <T, C extends Collection<T>> C decodeJsonCollection(Class<T> valueClass, Class<C> collectionType, akka.util.ByteString bytes) {
+    return jsonSerializer.fromBytes(valueClass, collectionType, new BytesPayload(bytes, jsonSerializer.contentTypeFor(valueClass)));
+  }
+
+  public static <T, C extends Collection<T>> C decodeJsonCollection(Class<T> valueClass, Class<C> collectionType, byte[] bytes) {
+    return decodeJsonCollection(valueClass, collectionType, akka.util.ByteString.fromArrayUnsafe(bytes));
+  }
+
   /**
    * Decode the given protobuf Any to an instance of T using Jackson but only if the suffix of the
    * type URL matches the given jsonType.
    *
    * @return An Optional containing the successfully decoded value or an empty Optional if the type
    *     suffix does not match.
    * @throws IllegalArgumentException if the suffix matches but the Any cannot be parsed into a T
+   *
+   * @deprecated Protobuf Any with JSON is not supported
    */
+  @Deprecated
   public static <T> Optional<T> decodeJson(Class<T> valueClass, String jsonType, Any any) {
     if (any.getTypeUrl().endsWith(jsonType)) {
       return Optional.of(decodeJson(valueClass, any));
@@ -273,15 +359,6 @@ public static <T> Optional<T> decodeJson(Class<T> valueClass, String jsonType, A
     }
   }
 
-  /**
-   * INTERNAL API
-   * @hidden
-   */
-  @InternalApi
-  public static <T> T decodeJson(Class<T> valueClass, com.google.protobuf.any.Any scalaPbAny) {
-    var javaAny = com.google.protobuf.any.Any.toJavaProto(scalaPbAny);
-    return JsonSupport.decodeJson(valueClass, javaAny);
-  }
 }
 
 class DoneSerializer extends JsonSerializer<Done> {

akka-javasdk/src/main/java/akka/javasdk/http/HttpResponses.java

@@ -60,12 +60,8 @@ public static HttpResponse ok(String text) {
    */
   public static HttpResponse ok(Object object) {
     if (object == null) throw new IllegalArgumentException("object must not be null");
-    try {
-      byte[] body = JsonSupport.encodeToBytes(object).toByteArray();
-      return HttpResponse.create().withEntity(ContentTypes.APPLICATION_JSON, body);
-    } catch (JsonProcessingException e) {
-      throw new RuntimeException(e);
-    }
+    var body = JsonSupport.encodeToAkkaByteString(object);
+    return HttpResponse.create().withEntity(ContentTypes.APPLICATION_JSON, body);
   }
 
   /**

akka-javasdk/src/main/scala/akka/javasdk/impl/http/HttpClientImpl.scala

@@ -117,7 +117,7 @@ private[akka] final case class RequestBuilderImpl[R](
   override def withRequestBody(`object`: AnyRef): RequestBuilder[R] = {
     if (`object` eq null) throw new IllegalArgumentException("object must not be null")
     try {
-      val body = JsonSupport.encodeToBytes(`object`).toByteArray
+      val body = JsonSupport.encodeToAkkaByteString(`object`)
       val requestWithBody = request.withEntity(ContentTypes.APPLICATION_JSON, body)
       withRequest(requestWithBody)
     } catch {
@@ -167,7 +167,7 @@ private[akka] final case class RequestBuilderImpl[R](
             throw new RuntimeException(errorString + ": " + bytes.utf8String)
         }
       } else if (res.entity.getContentType == ContentTypes.APPLICATION_JSON)
-        new StrictResponse[T](res, JsonSupport.parseBytes(bytes.toArrayUnsafe(), `type`))
+        new StrictResponse[T](res, JsonSupport.decodeJson(`type`, bytes))
       else if (!res.entity.getContentType.binary && (`type` eq classOf[String]))
         new StrictResponse[T](
           res,

akka-javasdk/src/test/scala/akka/javasdk/JsonSupportSpec.scala

@@ -19,8 +19,8 @@ class MyJsonable {
   @BeanProperty var field: String = _
 }
 
+@Deprecated
 class JsonSupportSpec extends AnyWordSpec with Matchers {
-  // FIXME move these tests to JsonSerializerSpec
 
   val myJsonable = new MyJsonable
   myJsonable.field = "foo"

docs/src/modules/java/pages/serialization.adoc

@@ -164,5 +164,4 @@ include::example$event-sourced-customer-registry/src/test/java/customer/domain/C
 ----
 <1> Encodes old class in Base64 String.
 <2> Decodes Base64 bytes.
-<3> Parses bytes into `Any` object.
-<4> Verifies JSON deserialization.
+<3> Verifies JSON deserialization.

samples/event-sourced-counter-brokers/src/it/java/counter/application/CounterWithRealKafkaIntegrationTest.java

@@ -55,7 +55,7 @@ public void verifyCounterEventSourcedConsumesFromKafka() {
           ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(200));
           var foundRecord = false;
           for (ConsumerRecord<String, byte[]> r : records) {
-            var increased = JsonSupport.parseBytes(r.value(), CounterEvent.ValueIncreased.class);
+            var increased = JsonSupport.decodeJson(CounterEvent.ValueIncreased.class, r.value());
             String subjectId = new String(r.headers().headers("ce-subject").iterator().next().value(), StandardCharsets.UTF_8);
             if (subjectId.equals(counterId) && increased.value() == 20) {
               foundRecord = true;

samples/event-sourced-customer-registry/src/test/java/customer/domain/CustomerEventSerializationTest.java

@@ -1,9 +1,7 @@
 package customer.domain;
 
-import com.google.protobuf.Any;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.InvalidProtocolBufferException;
 import akka.javasdk.JsonSupport;
+import akka.util.ByteString;
 import org.junit.jupiter.api.Test;
 
 import java.util.Base64;
@@ -17,7 +15,7 @@ class CustomerEventSerializationTest {
   @Test
   public void shouldDeserializeWithMandatoryField() {
     //given
-    Any serialized = JsonSupport.encodeJson(new CustomerEvent.NameChanged("andre"));
+    ByteString serialized = JsonSupport.encodeToAkkaByteString(new CustomerEvent.NameChanged("andre"));
 
     //when
     NameChanged deserialized = JsonSupport.decodeJson(NameChanged.class, serialized);
@@ -32,7 +30,7 @@ public void shouldDeserializeWithMandatoryField() {
   public void shouldDeserializeWithChangedFieldName() {
     //given
     Address address = new Address("Wall Street", "New York");
-    Any serialized = JsonSupport.encodeJson(new CustomerEvent.AddressChanged(address));
+    ByteString serialized = JsonSupport.encodeToAkkaByteString(new CustomerEvent.AddressChanged(address));
 
     //when
     AddressChanged deserialized = JsonSupport.decodeJson(AddressChanged.class, serialized);
@@ -44,7 +42,7 @@ public void shouldDeserializeWithChangedFieldName() {
   @Test
   public void shouldDeserializeWithStructureMigration() {
     //given
-    Any serialized = JsonSupport.encodeJson(new CustomerCreatedOld("[email protected]", "bob", "Wall Street", "New York"));
+    ByteString serialized = JsonSupport.encodeToAkkaByteString(new CustomerCreatedOld("[email protected]", "bob", "Wall Street", "New York"));
 
     //when
     CustomerEvent.CustomerCreated deserialized = JsonSupport.decodeJson(CustomerEvent.CustomerCreated.class, serialized);
@@ -56,21 +54,20 @@ public void shouldDeserializeWithStructureMigration() {
 
   // tag::testing-deserialization[]
   @Test
-  public void shouldDeserializeCustomerCreated_V0() throws InvalidProtocolBufferException {
+  public void shouldDeserializeCustomerCreated_V0() {
     // tag::testing-deserialization-encoding[]
-    Any serialized = JsonSupport.encodeJson(new CustomerCreatedOld("[email protected]", "bob", "Wall Street", "New York"));
-    String encodedBytes = new String(Base64.getEncoder().encode(serialized.toByteArray())); // <1>
+    ByteString serialized = JsonSupport.encodeToAkkaByteString(new CustomerCreatedOld("[email protected]", "bob", "Wall Street", "New York"));
+    String encodedBytes = serialized.encodeBase64().utf8String(); // <1>
     // end::testing-deserialization-encoding[]
 
-    byte[] bytes = Base64.getDecoder().decode(encodedBytes.getBytes()); // <2>
-    Any serializedAny = Any.parseFrom(ByteString.copyFrom(bytes)); // <3>
+    ByteString decodedBytes = ByteString.fromString(encodedBytes).decodeBase64(); // <2>
 
     CustomerEvent.CustomerCreated deserialized = JsonSupport.decodeJson(CustomerEvent.CustomerCreated.class,
-      serializedAny); // <4>
+      decodedBytes); // <3>
 
     assertEquals("Wall Street", deserialized.address().street());
     assertEquals("New York", deserialized.address().city());
   }
   // end::testing-deserialization[]
 
-}
+}