Provide a way to automatically deserialize non-OK JSON response (#5002)

Motivation:

Please refer to #4382 

Modifications:

- Add some methods to specify what type of response is allowed 
- Add `HttpStatusPredicate` and `HttpStatusClassPredicate` class to
deserialize non-OK JSON response for more shortcut methods.

Result:

- Closes #4382 
- Users can deserialize non-OK JSON response like the following

```java
// WebClient
WebClient.of().prepare()
     .as(ResponseAs.blocking()
                   .andThen(res -> "Unexpected server error", res -> res.status().isServerError())
                   .andThen(res -> "missing header", res -> !res.headers().contains("x-header"))
                   .orElse(AggregatedHttpObject::contentUtf8))
     .execute();

// RestClient
final ResponseEntity<MyResponse> res =
        RestClient.of(server.httpUri()).get("/")
                  .execute(ResponseAs.blocking()
                                     .<MyResponse>andThenJson(MyError.class, res -> res.status().isClientError())
                                     .andThenJson(EmptyMessage.class, res -> res.status().isInformational())
                                     .orElseJson(MyMessage.class));
```

---------

Co-authored-by: jrhee17 <guins_j@guins.org>
Co-authored-by: Hannam Rhee <jrhee17@linecorp.com>
Co-authored-by: minu <minu@AL02442289.local>
Co-authored-by: minwoox <songmw725@gmail.com>

core/src/main/java/com/linecorp/armeria/client/AggregatedResponseAs.java

@@ -17,14 +17,14 @@
 package com.linecorp.armeria.client;
 
 import java.io.IOException;
+import java.util.function.Predicate;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import com.linecorp.armeria.common.AggregatedHttpResponse;
 import com.linecorp.armeria.common.ResponseEntity;
 import com.linecorp.armeria.common.util.Exceptions;
-import com.linecorp.armeria.internal.common.JacksonUtil;
 
 final class AggregatedResponseAs {
 
@@ -37,30 +37,40 @@ static ResponseAs<AggregatedHttpResponse, ResponseEntity<String>> string() {
         return response -> ResponseEntity.of(response.headers(), response.contentUtf8(), response.trailers());
     }
 
-    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(Class<? extends T> clazz) {
-        return response -> newJsonResponseEntity(response, bytes -> JacksonUtil.readValue(bytes, clazz));
+    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(
+            Class<? extends T> clazz, ObjectMapper mapper) {
+        return response -> newJsonResponseEntity(response, bytes -> mapper.readValue(bytes, clazz));
     }
 
-    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(Class<? extends T> clazz,
-                                                                          ObjectMapper mapper) {
-        return response -> newJsonResponseEntity(response, bytes -> mapper.readValue(bytes, clazz));
+    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(
+            TypeReference<? extends T> typeRef, ObjectMapper mapper) {
+        return response -> newJsonResponseEntity(
+                response, bytes -> mapper.readValue(bytes, typeRef));
     }
 
-    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(TypeReference<? extends T> typeRef) {
-        return response -> newJsonResponseEntity(response, bytes -> JacksonUtil.readValue(bytes, typeRef));
+    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(
+            Class<? extends T> clazz, ObjectMapper mapper, Predicate<AggregatedHttpResponse> predicate) {
+        return response -> newJsonResponseEntity(response, bytes -> mapper.readValue(bytes, clazz), predicate);
     }
 
-    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(TypeReference<? extends T> typeRef,
-                                                                          ObjectMapper mapper) {
-        return response -> newJsonResponseEntity(response, bytes -> mapper.readValue(bytes, typeRef));
+    static <T> ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> json(
+            TypeReference<? extends T> typeRef, ObjectMapper mapper,
+            Predicate<AggregatedHttpResponse> predicate) {
+        return response -> newJsonResponseEntity(
+                response, bytes -> mapper.readValue(bytes, typeRef), predicate);
     }
 
     private static <T> ResponseEntity<T> newJsonResponseEntity(AggregatedHttpResponse response,
-                                                               JsonDecoder<T> decoder) {
-        if (!response.status().isSuccess()) {
+                                                               JsonDecoder<T> decoder,
+                                                               Predicate<AggregatedHttpResponse> predicate) {
+        if (!predicate.test(response)) {
             throw newInvalidHttpResponseException(response);
         }
+        return newJsonResponseEntity(response, decoder);
+    }
 
+    private static <T> ResponseEntity<T> newJsonResponseEntity(AggregatedHttpResponse response,
+                                                               JsonDecoder<T> decoder) {
         try {
             return ResponseEntity.of(response.headers(), decoder.decode(response.content().array()),
                                      response.trailers());

core/src/main/java/com/linecorp/armeria/client/BlockingWebClientRequestPreparation.java

@@ -16,6 +16,8 @@
 
 package com.linecorp.armeria.client;
 
+import static com.linecorp.armeria.client.ResponseAsUtil.OBJECT_MAPPER;
+import static com.linecorp.armeria.client.ResponseAsUtil.SUCCESS_PREDICATE;
 import static java.util.Objects.requireNonNull;
 
 import java.time.Duration;
@@ -133,8 +135,7 @@ public TransformingRequestPreparation<AggregatedHttpResponse, ResponseEntity<Str
     @UnstableApi
     public <T> TransformingRequestPreparation<AggregatedHttpResponse, ResponseEntity<T>> asJson(
             Class<? extends T> clazz) {
-        requireNonNull(clazz, "clazz");
-        return as(AggregatedResponseAs.json(clazz));
+        return asJson(clazz, OBJECT_MAPPER);
     }
 
     /**
@@ -162,7 +163,7 @@ public <T> TransformingRequestPreparation<AggregatedHttpResponse, ResponseEntity
             Class<? extends T> clazz, ObjectMapper mapper) {
         requireNonNull(clazz, "clazz");
         requireNonNull(mapper, "mapper");
-        return as(AggregatedResponseAs.json(clazz, mapper));
+        return as(AggregatedResponseAs.json(clazz, mapper, SUCCESS_PREDICATE));
     }
 
     /**
@@ -187,8 +188,7 @@ public <T> TransformingRequestPreparation<AggregatedHttpResponse, ResponseEntity
     @UnstableApi
     public <T> TransformingRequestPreparation<AggregatedHttpResponse, ResponseEntity<T>> asJson(
             TypeReference<? extends T> typeRef) {
-        requireNonNull(typeRef, "typeRef");
-        return as(AggregatedResponseAs.json(typeRef));
+        return asJson(typeRef, OBJECT_MAPPER);
     }
 
     /**
@@ -214,7 +214,7 @@ public <T> TransformingRequestPreparation<AggregatedHttpResponse, ResponseEntity
             TypeReference<? extends T> typeRef, ObjectMapper mapper) {
         requireNonNull(typeRef, "typeRef");
         requireNonNull(mapper, "mapper");
-        return as(AggregatedResponseAs.json(typeRef, mapper));
+        return as(AggregatedResponseAs.json(typeRef, mapper, SUCCESS_PREDICATE));
     }
 
     @Override

core/src/main/java/com/linecorp/armeria/client/JsonConditionalResponseAs.java

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2023 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.client;
+
+import static com.linecorp.armeria.client.ResponseAsUtil.OBJECT_MAPPER;
+import static java.util.Objects.requireNonNull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+
+import com.linecorp.armeria.common.AggregatedHttpResponse;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.ResponseEntity;
+import com.linecorp.armeria.common.annotation.UnstableApi;
+
+/**
+ * Provides a way for users to add {@link ResponseAs} mappings to transform an aggregated response
+ * given that the corresponding {@link Predicate} is satisfied. Note that the conditionals are
+ * invoked in the order in which they are added.
+ *
+ * <pre>{@code
+ * RestClient.of(...)
+ *   .get("/")
+ *   .execute(
+ *     ResponseAs.<MyResponse>json(MyMessage.class, res -> res.status().isError())
+ *       .orElseJson(EmptyMessage.class, res -> res.status().isInformational())
+ *       .orElseJson(MyError.class)).join();
+ * }</pre>
+ */
+@UnstableApi
+public final class JsonConditionalResponseAs<T> {
+
+    private final List<Entry<Predicate<AggregatedHttpResponse>,
+            ResponseAs<AggregatedHttpResponse, ResponseEntity<T>>>> responseConverters = new ArrayList<>();
+
+    JsonConditionalResponseAs(Predicate<AggregatedHttpResponse> predicate,
+                              ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> responseAs) {
+        responseConverters.add(Maps.immutableEntry(
+                requireNonNull(predicate, "predicate"), requireNonNull(responseAs, "responseAs")));
+    }
+
+    /**
+     * Sets the {@link Predicate} and {@link Class} that the content is deserialized into the {@link Class}
+     * when the {@link AggregatedHttpResponse} passes the {@link Predicate}.
+     */
+    public JsonConditionalResponseAs<T> orElseJson(
+            Class<? extends T> clazz, Predicate<AggregatedHttpResponse> predicate) {
+        return orElseJson(clazz, OBJECT_MAPPER, predicate);
+    }
+
+    /**
+     * Sets the {@link Predicate} and {@link Class} that the content is deserialized into the {@link Class}
+     * using the {@link ObjectMapper} when the {@link AggregatedHttpResponse} passes the {@link Predicate}.
+     */
+    public JsonConditionalResponseAs<T> orElseJson(
+            Class<? extends T> clazz, ObjectMapper objectMapper, Predicate<AggregatedHttpResponse> predicate) {
+        requireNonNull(clazz, "clazz");
+        requireNonNull(objectMapper, "objectMapper");
+        requireNonNull(predicate, "predicate");
+        responseConverters.add(Maps.immutableEntry(predicate, AggregatedResponseAs.json(clazz, objectMapper)));
+        return this;
+    }
+
+    /**
+     * Sets the {@link Predicate} and {@link TypeReference} that the content is deserialized into the
+     * {@link TypeReference} when the {@link AggregatedHttpResponse} passes the {@link Predicate}.
+     */
+    public JsonConditionalResponseAs<T> orElseJson(
+            TypeReference<? extends T> typeRef, Predicate<AggregatedHttpResponse> predicate) {
+        return orElseJson(typeRef, OBJECT_MAPPER, predicate);
+    }
+
+    /**
+     * Sets the {@link Predicate} and {@link TypeReference} that the content is deserialized into the
+     * {@link TypeReference} using the {@link ObjectMapper} when the {@link AggregatedHttpResponse} passes
+     * the {@link Predicate}.
+     */
+    public JsonConditionalResponseAs<T> orElseJson(
+            TypeReference<? extends T> typeRef, ObjectMapper objectMapper,
+            Predicate<AggregatedHttpResponse> predicate) {
+        requireNonNull(typeRef, "typeRef");
+        requireNonNull(objectMapper, "objectMapper");
+        requireNonNull(predicate, "predicate");
+        responseConverters.add(Maps.immutableEntry(predicate,
+                                                   AggregatedResponseAs.json(typeRef, objectMapper)));
+        return this;
+    }
+
+    /**
+     * Returns {@link FutureResponseAs} that deserializes the {@link HttpResponse} based on the configured
+     * deserializers so far and deserializes to the {@link Class} lastly if none of the {@link Predicate} of
+     * configured deserializers pass.
+     */
+    public FutureResponseAs<ResponseEntity<T>> orElseJson(Class<? extends T> clazz) {
+        return orElseJson(clazz, OBJECT_MAPPER);
+    }
+
+    /**
+     * Returns {@link FutureResponseAs} that deserializes the {@link HttpResponse} based on the configured
+     * deserializers so far and deserializes to the {@link Class} lastly using the {@link ObjectMapper}
+     * if none of the {@link Predicate} of configured deserializers pass.
+     */
+    public FutureResponseAs<ResponseEntity<T>> orElseJson(
+            Class<? extends T> clazz, ObjectMapper objectMapper) {
+        return orElse(AggregatedResponseAs.json(clazz, objectMapper));
+    }
+
+    /**
+     * Returns {@link FutureResponseAs} that deserializes the {@link HttpResponse} based on the configured
+     * deserializers so far and deserializes to the {@link TypeReference} lastly if none of the
+     * {@link Predicate} of configured deserializers pass.
+     */
+    public FutureResponseAs<ResponseEntity<T>> orElseJson(TypeReference<? extends T> typeRef) {
+        return orElseJson(typeRef, OBJECT_MAPPER);
+    }
+
+    /**
+     * Returns {@link FutureResponseAs} that deserializes the {@link HttpResponse} based on the configured
+     * deserializers so far and deserializes to the {@link TypeReference} lastly using the {@link ObjectMapper}
+     * if none of the {@link Predicate} of configured deserializers pass.
+     */
+    public FutureResponseAs<ResponseEntity<T>> orElseJson(
+            TypeReference<? extends T> typeRef, ObjectMapper objectMapper) {
+        return orElse(AggregatedResponseAs.json(typeRef, objectMapper));
+    }
+
+    private FutureResponseAs<ResponseEntity<T>> orElse(
+            ResponseAs<AggregatedHttpResponse, ResponseEntity<T>> lastConverter) {
+        final List<Entry<Predicate<AggregatedHttpResponse>,
+                ResponseAs<AggregatedHttpResponse, ResponseEntity<T>>>> converters =
+                ImmutableList.copyOf(responseConverters);
+        return new FutureResponseAs<ResponseEntity<T>>() {
+            @Override
+            public CompletableFuture<ResponseEntity<T>> as(HttpResponse response) {
+                requireNonNull(response, "response");
+                return response.aggregate().thenApply(aggregated -> {
+                    for (Entry<Predicate<AggregatedHttpResponse>,
+                            ResponseAs<AggregatedHttpResponse, ResponseEntity<T>>>
+                            converter : converters) {
+                        if (converter.getKey().test(aggregated)) {
+                            return converter.getValue().as(aggregated);
+                        }
+                    }
+                    return lastConverter.as(aggregated);
+                });
+            }
+
+            @Override
+            public boolean requiresAggregation() {
+                return true;
+            }
+        };
+    }
+}