Add streaming read methods to JsonCodec

They can be more efficient than reading all contents to memory as bytes and then processing them.
ksobolew · Jan 16, 2025 · 65668f4 · 65668f4
1 parent 66ba923
commit 65668f4
Show file tree

Hide file tree

Showing 8 changed files with 129 additions and 6 deletions.
diff --git a/discovery/src/main/java/io/airlift/discovery/client/HttpDiscoveryLookupClient.java b/discovery/src/main/java/io/airlift/discovery/client/HttpDiscoveryLookupClient.java
@@ -138,15 +138,14 @@ public ServiceDescriptors handle(Request request, Response response)
                     throw new DiscoveryException(format("Lookup of %s failed with status code %s", type, response.getStatusCode()));
                 }
 
-                byte[] json;
+                ServiceDescriptorsRepresentation serviceDescriptorsRepresentation;
                 try (InputStream stream = response.getInputStream()) {
-                    json = stream.readAllBytes();
+                    serviceDescriptorsRepresentation = serviceDescriptorsCodec.fromJson(stream);
                 }
                 catch (IOException e) {
                     throw new DiscoveryException(format("Lookup of %s failed", type), e);
                 }
 
-                ServiceDescriptorsRepresentation serviceDescriptorsRepresentation = serviceDescriptorsCodec.fromJson(json);
                 if (!environment.equals(serviceDescriptorsRepresentation.getEnvironment())) {
                     throw new DiscoveryException(format("Expected environment to be %s, but was %s", environment, serviceDescriptorsRepresentation.getEnvironment()));
                 }

diff --git a/discovery/src/main/java/io/airlift/discovery/client/ServiceInventory.java b/discovery/src/main/java/io/airlift/discovery/client/ServiceInventory.java
@@ -42,7 +42,7 @@
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
 import static io.airlift.http.client.JsonResponseHandler.createJsonResponseHandler;
 import static io.airlift.http.client.Request.Builder.prepareGet;
-import static java.nio.file.Files.readAllBytes;
+import static java.nio.file.Files.newBufferedReader;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
 import static java.util.stream.Collectors.toList;
@@ -156,7 +156,7 @@ public final void updateServiceInventory()
             }
             else {
                 File file = new File(serviceInventoryUri);
-                serviceDescriptorsRepresentation = serviceDescriptorsCodec.fromJson(readAllBytes(file.toPath()));
+                serviceDescriptorsRepresentation = serviceDescriptorsCodec.fromJson(newBufferedReader(file.toPath()));
             }
 
             if (!environment.equals(serviceDescriptorsRepresentation.getEnvironment())) {

diff --git a/http-client/src/main/java/io/airlift/http/client/DefaultingJsonResponseHandler.java b/http-client/src/main/java/io/airlift/http/client/DefaultingJsonResponseHandler.java
@@ -73,7 +73,7 @@ public T handle(Request request, Response response)
             return defaultValue;
         }
         try (InputStream inputStream = response.getInputStream()) {
-            return jsonCodec.fromJson(inputStream.readAllBytes());
+            return jsonCodec.fromJson(inputStream);
         }
         catch (Exception e) {
             return defaultValue;

diff --git a/json/src/main/java/io/airlift/json/JsonCodec.java b/json/src/main/java/io/airlift/json/JsonCodec.java
@@ -24,6 +24,8 @@
 import io.airlift.json.LengthLimitedWriter.LengthLimitExceededException;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
 import java.io.StringWriter;
 import java.lang.reflect.Type;
 import java.util.List;
@@ -221,6 +223,46 @@ public byte[] toJsonBytes(T instance)
         }
     }
 
+    /**
+     * Coverts the specified {@link InputStream} (UTF-8) into an instance of type T.
+     *
+     * @param json the json stream (UTF-8) to parse
+     * @return parsed response; never null
+     * @throws IllegalArgumentException if the json bytes can not be converted to the type T
+     */
+    public T fromJson(InputStream json)
+            throws IllegalArgumentException
+    {
+        try (JsonParser parser = mapper.createParser(json)) {
+            T value = mapper.readerFor(javaType).readValue(parser);
+            checkArgument(parser.nextToken() == null, "Found characters after the expected end of input");
+            return value;
+        }
+        catch (IOException e) {
+            throw new IllegalArgumentException(format("Invalid JSON bytes for %s", javaType), e);
+        }
+    }
+
+    /**
+     * Coverts the specified {@link Reader} into an instance of type T.
+     *
+     * @param json the json character stream to parse
+     * @return parsed response; never null
+     * @throws IllegalArgumentException if the json characters can not be converted to the type T
+     */
+    public T fromJson(Reader json)
+            throws IllegalArgumentException
+    {
+        try (JsonParser parser = mapper.createParser(json)) {
+            T value = mapper.readerFor(javaType).readValue(parser);
+            checkArgument(parser.nextToken() == null, "Found characters after the expected end of input");
+            return value;
+        }
+        catch (IOException e) {
+            throw new IllegalArgumentException(format("Invalid JSON characters for %s", javaType), e);
+        }
+    }
+
     @SuppressWarnings("unchecked")
     TypeToken<T> getTypeToken()
     {

diff --git a/json/src/test/java/io/airlift/json/ImmutablePerson.java b/json/src/test/java/io/airlift/json/ImmutablePerson.java
@@ -20,11 +20,14 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
 import static com.google.common.base.MoreObjects.toStringHelper;
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class ImmutablePerson
@@ -42,6 +45,10 @@ public static void validatePersonJsonCodec(JsonCodec<ImmutablePerson> jsonCodec)
 
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     public static void validatePersonListJsonCodec(JsonCodec<List<ImmutablePerson>> jsonCodec)
@@ -56,6 +63,10 @@ public static void validatePersonListJsonCodec(JsonCodec<List<ImmutablePerson>>
 
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     public static void validatePersonMapJsonCodec(JsonCodec<Map<String, ImmutablePerson>> jsonCodec)
@@ -71,6 +82,10 @@ public static void validatePersonMapJsonCodec(JsonCodec<Map<String, ImmutablePer
 
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     @JsonCreator

diff --git a/json/src/test/java/io/airlift/json/Person.java b/json/src/test/java/io/airlift/json/Person.java
@@ -19,12 +19,15 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 
 import static com.google.common.base.MoreObjects.toStringHelper;
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class Person
@@ -45,6 +48,10 @@ public static void validatePersonJsonCodec(JsonCodec<Person> jsonCodec)
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
 
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
+
         // create object with missing lastName
         expected.setLastName(Optional.empty());
 
@@ -55,6 +62,10 @@ public static void validatePersonJsonCodec(JsonCodec<Person> jsonCodec)
         bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
 
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
+
         // create object with present lastName
         expected.setLastName(Optional.of("Awesome"));
 
@@ -64,6 +75,10 @@ public static void validatePersonJsonCodec(JsonCodec<Person> jsonCodec)
 
         bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     public static void validatePersonListJsonCodec(JsonCodec<List<Person>> jsonCodec)
@@ -78,6 +93,10 @@ public static void validatePersonListJsonCodec(JsonCodec<List<Person>> jsonCodec
 
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     public static void validatePersonMapJsonCodec(JsonCodec<Map<String, Person>> jsonCodec)
@@ -93,6 +112,10 @@ public static void validatePersonMapJsonCodec(JsonCodec<Map<String, Person>> jso
 
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     @JsonProperty

diff --git a/json/src/test/java/io/airlift/json/TestJsonCodec.java b/json/src/test/java/io/airlift/json/TestJsonCodec.java
@@ -21,6 +21,9 @@
 import com.google.common.reflect.TypeToken;
 import org.junit.jupiter.api.Test;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
+import java.io.StringReader;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -232,6 +235,10 @@ public void testToJsonExceedingDefaultStringLimit()
 
         byte[] bytes = jsonCodec.toJsonBytes(person);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(person);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(person);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(person);
     }
 
     @Test
@@ -278,6 +285,16 @@ public void testTrailingContent()
                 .hasMessage("Invalid JSON bytes for [simple type, class io.airlift.json.ImmutablePerson]")
                 .hasStackTraceContaining("Unrecognized token 'trailer': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')");
 
+        assertThatThrownBy(() -> codec.fromJson(new ByteArrayInputStream(jsonWithTrailingContent.getBytes(UTF_8))))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessage("Invalid JSON bytes for [simple type, class io.airlift.json.ImmutablePerson]")
+                .hasStackTraceContaining("Unrecognized token 'trailer': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')");
+
+        assertThatThrownBy(() -> codec.fromJson(new StringReader(jsonWithTrailingContent)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessage("Invalid JSON characters for [simple type, class io.airlift.json.ImmutablePerson]")
+                .hasStackTraceContaining("Unrecognized token 'trailer': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')");
+
         String jsonWithTrailingJsonContent = json + " \"valid json value\"";
         assertThatThrownBy(() -> codec.fromJson(jsonWithTrailingJsonContent))
                 .isInstanceOf(IllegalArgumentException.class)
@@ -286,6 +303,14 @@ public void testTrailingContent()
         assertThatThrownBy(() -> codec.fromJson(jsonWithTrailingJsonContent.getBytes(UTF_8)))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("Found characters after the expected end of input");
+
+        assertThatThrownBy(() -> codec.fromJson(new ByteArrayInputStream(jsonWithTrailingJsonContent.getBytes(UTF_8))))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessage("Found characters after the expected end of input");
+
+        assertThatThrownBy(() -> codec.fromJson(new StringReader(jsonWithTrailingJsonContent)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessage("Found characters after the expected end of input");
     }
 
     @Test

diff --git a/json/src/test/java/io/airlift/json/Vehicle.java b/json/src/test/java/io/airlift/json/Vehicle.java
@@ -5,9 +5,12 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
 import java.util.List;
 import java.util.Map;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
 
 @JsonTypeInfo(
@@ -30,6 +33,10 @@ static void validateVehicleJsonCodec(JsonCodec<Vehicle> jsonCodec)
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
 
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
+
         expected = new Truck("volvo");
 
         json = jsonCodec.toJson(expected);
@@ -38,6 +45,10 @@ static void validateVehicleJsonCodec(JsonCodec<Vehicle> jsonCodec)
 
         bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     static void validateVehicleListJsonCodec(JsonCodec<List<Vehicle>> jsonCodec)
@@ -51,6 +62,10 @@ static void validateVehicleListJsonCodec(JsonCodec<List<Vehicle>> jsonCodec)
 
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 
     static void validateVehicleMapJsonCodec(JsonCodec<Map<String, Vehicle>> jsonCodec)
@@ -65,5 +80,9 @@ static void validateVehicleMapJsonCodec(JsonCodec<Map<String, Vehicle>> jsonCode
 
         byte[] bytes = jsonCodec.toJsonBytes(expected);
         assertThat(jsonCodec.fromJson(bytes)).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new ByteArrayInputStream(bytes))).isEqualTo(expected);
+
+        assertThat(jsonCodec.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes), UTF_8))).isEqualTo(expected);
     }
 }