support invokeXXX return Response

sdk/src/main/java/io/dapr/client/DaprClientGrpc.java

@@ -37,6 +37,8 @@
 import io.dapr.client.domain.StateOptions;
 import io.dapr.client.domain.SubscribeConfigurationRequest;
 import io.dapr.client.domain.TransactionalStateOperation;
+import io.dapr.client.domain.response.DaprResponse;
+import io.dapr.client.domain.response.GrpcDaprResponse;
 import io.dapr.config.Properties;
 import io.dapr.exceptions.DaprException;
 import io.dapr.internal.opencensus.GrpcWrapper;
@@ -62,9 +64,13 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -206,6 +212,11 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
           ).flatMap(
               it -> {
                 try {
+                  if (type.getType().getTypeName().startsWith(DaprResponse.class.getName())) {
+                    Map<String, String> headers = new HashMap<>();
+                    headers.put(io.dapr.client.domain.Metadata.CONTENT_TYPE,it.getContentType());
+                    return getMono(type, it.getData().getValue().toByteArray(),headers);
+                  }
                   return Mono.justOrEmpty(objectSerializer.deserialize(it.getData().getValue().toByteArray(), type));
                 } catch (IOException e)  {
                   throw DaprException.propagate(e);
@@ -217,6 +228,11 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
     }
   }
 
+  private <T> Mono<T> getMono(TypeRef<T> type, byte[] data, Map<String,String> headers) {
+    type = DaprResponse.getSubType(type);
+    return (Mono<T>) Mono.just(new GrpcDaprResponse<T>(data, headers,objectSerializer, type));
+  }
+
   /**
    * {@inheritDoc}
    */
@@ -253,6 +269,9 @@ public <T> Mono<T> invokeBinding(InvokeBindingRequest request, TypeRef<T> type)
           ).flatMap(
               it -> {
                 try {
+                  if (type.getType().getTypeName().startsWith(DaprResponse.class.getName())) {
+                    return getMono(type, it.getData().toByteArray(),it.getMetadataMap());
+                  }
                   return Mono.justOrEmpty(objectSerializer.deserialize(it.getData().toByteArray(), type));
                 } catch (IOException e) {
                   throw DaprException.propagate(e);

sdk/src/main/java/io/dapr/client/DaprClientHttp.java

@@ -36,6 +36,8 @@
 import io.dapr.client.domain.SubscribeConfigurationRequest;
 import io.dapr.client.domain.TransactionalStateOperation;
 import io.dapr.client.domain.TransactionalStateRequest;
+import io.dapr.client.domain.response.DaprResponse;
+import io.dapr.client.domain.response.HttpDaprResponse;
 import io.dapr.config.Properties;
 import io.dapr.exceptions.DaprException;
 import io.dapr.serializer.DaprObjectSerializer;
@@ -46,13 +48,16 @@
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
@@ -221,6 +226,10 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
 
   private <T> Mono<T> getMono(TypeRef<T> type, DaprHttp.Response r) {
     try {
+      if (type.getType().getTypeName().startsWith(DaprResponse.class.getName())) {
+        type = DaprResponse.getSubType(type);
+        return (Mono<T>) Mono.just(new HttpDaprResponse<T>(r, objectSerializer, type));
+      }
       T object = objectSerializer.deserialize(r.getBody(), type);
       if (object == null) {
         return Mono.empty();

sdk/src/main/java/io/dapr/client/domain/response/DaprResponse.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2021 The Dapr Authors
+ * Licensed 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
+ *     http://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 io.dapr.client.domain.response;
+
+import io.dapr.utils.TypeRef;
+
+import java.io.IOException;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Response.
+ */
+public interface DaprResponse<T> {
+
+  /**
+   * get response code.
+   * @return response code
+   */
+  int getCode();
+
+  /**
+   * get response data.
+   * @return response data
+   */
+  T getData() throws IOException;
+
+  /**
+   * get response header.
+   * @return response header
+   */
+  Map<String,String> getHeaders();
+
+  /**
+   * get sub type from type.
+   * @param type response type
+   * @param <T> type
+   * @return sub type
+   */
+  static <T> TypeRef getSubType(TypeRef<T> type) {
+    TypeRef resultType = type;
+    if (type.getType() instanceof ParameterizedType) {
+      Type[] actualTypeArguments = ((ParameterizedType) type.getType()).getActualTypeArguments();
+      if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
+        resultType = TypeRef.get(actualTypeArguments[0]);
+      }
+    } else {
+      resultType = TypeRef.get(byte[].class);
+    }
+    return resultType;
+  }
+
+  /**
+   * get response bytes.
+   * @return bytes
+   */
+  byte[] getSourceBytesData();
+
+  /**
+   * get repsonse bytes without `34`.
+   * string that serialized by objectmapper will surrounded by `"`.
+   * it will be removed when return bytes.
+   * @return bytes
+   */
+  default byte[] getBytes() {
+    byte[] data = getSourceBytesData();
+    if (data.length > 1 && data[0] == 34) {
+      data = Arrays.copyOfRange(data, 1, data.length - 1);
+    }
+    return data;
+  }
+}

sdk/src/main/java/io/dapr/client/domain/response/GrpcDaprResponse.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2021 The Dapr Authors
+ * Licensed 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
+ *     http://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 io.dapr.client.domain.response;
+
+import io.dapr.serializer.DaprObjectSerializer;
+import io.dapr.utils.TypeRef;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * GrpcDaprResponse.
+ */
+public class GrpcDaprResponse<T> implements DaprResponse<T> {
+
+  private final DaprObjectSerializer serializer;
+
+  private final TypeRef<T> type;
+
+  private final byte[] data;
+
+  private final Map<String,String> headers;
+
+  /**
+   * build grpc dapr response.
+   * @param data grpc invoke response data
+   * @param headers grpc invoke headers
+   * @param serializer objectSerializer
+   * @param type type
+   */
+  public GrpcDaprResponse(byte[] data, Map<String,String> headers, DaprObjectSerializer serializer, TypeRef<T> type) {
+    this.data = data;
+    this.headers = headers;
+    this.serializer = serializer;
+    this.type = type;
+  }
+
+  @Override
+  public int getCode() {
+    // InvokeResponse didn't have it.
+    return 200;
+  }
+
+  @Override
+  public T getData() throws IOException {
+    if (type.getType() == byte[].class) {
+      return (T) getBytes();
+    }
+    return serializer.deserialize(data, type);
+  }
+
+  @Override
+  public Map<String, String> getHeaders() {
+    return this.headers;
+  }
+
+  @Override
+  public byte[] getSourceBytesData() {
+    return data;
+  }
+
+
+}

sdk/src/main/java/io/dapr/client/domain/response/HttpDaprResponse.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2021 The Dapr Authors
+ * Licensed 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
+ *     http://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 io.dapr.client.domain.response;
+
+import io.dapr.client.DaprHttp;
+import io.dapr.serializer.DaprObjectSerializer;
+import io.dapr.utils.TypeRef;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * HttpDaprResponse.
+ */
+public class HttpDaprResponse<T> implements DaprResponse<T> {
+
+  private final DaprHttp.Response response;
+
+  private final DaprObjectSerializer serializer;
+
+  private final TypeRef<T> type;
+
+  /**
+   * build http dapr response.
+   * @param response http invoke response
+   * @param serializer serializer
+   * @param type type
+   */
+  public HttpDaprResponse(DaprHttp.Response response, DaprObjectSerializer serializer, TypeRef<T> type) {
+    this.response = response;
+    this.serializer = serializer;
+    this.type = type;
+  }
+
+  @Override
+  public int getCode() {
+    return response.getStatusCode();
+  }
+
+  @Override
+  public T getData() throws IOException {
+    byte[] data = response.getBody();
+    if (type.getType() == String.class) {
+      return (T) new String(data);
+    }
+    if (type.getType() == byte[].class) {
+      return (T) getBytes();
+    }
+    return serializer.deserialize(data, type);
+  }
+
+  @Override
+  public Map<String, String> getHeaders() {
+    return response.getHeaders();
+  }
+
+  @Override
+  public byte[] getSourceBytesData() {
+    return response.getBody();
+  }
+}

sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java

@@ -25,6 +25,7 @@
 import io.dapr.client.domain.State;
 import io.dapr.client.domain.StateOptions;
 import io.dapr.client.domain.TransactionalStateOperation;
+import io.dapr.client.domain.response.DaprResponse;
 import io.dapr.config.Properties;
 import io.dapr.serializer.DaprObjectSerializer;
 import io.dapr.serializer.DefaultObjectSerializer;
@@ -633,6 +634,38 @@ public void invokeServiceTest() {
     assertEquals(expected, strOutput);
   }
 
+  @Test
+  public void invokeServiceTestReturnResponse() throws IOException {
+    String expected = "Value";
+    doAnswer((Answer<Void>) invocation -> {
+      StreamObserver<CommonProtos.InvokeResponse> observer = (StreamObserver<CommonProtos.InvokeResponse>) invocation.getArguments()[1];
+      observer.onNext(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build());
+      observer.onCompleted();
+      return null;
+    }).when(daprStub).invokeService(any(DaprProtos.InvokeServiceRequest.class), any());
+
+    Mono<DaprResponse<String>> result = client.invokeMethod("appId", "method", "request", HttpExtension.NONE, null, new TypeRef<DaprResponse<String>>() {});
+    DaprResponse<String> res = result.block();
+
+    assertEquals(expected, res.getData());
+  }
+
+  @Test
+  public void invokeServiceTestReturnResponseWithBytes() throws IOException {
+    String expected = "Value";
+    doAnswer((Answer<Void>) invocation -> {
+      StreamObserver<CommonProtos.InvokeResponse> observer = (StreamObserver<CommonProtos.InvokeResponse>) invocation.getArguments()[1];
+      observer.onNext(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build());
+      observer.onCompleted();
+      return null;
+    }).when(daprStub).invokeService(any(DaprProtos.InvokeServiceRequest.class), any());
+
+    Mono<DaprResponse> result = client.invokeMethod("appId", "method", "request", HttpExtension.NONE, null, new TypeRef<DaprResponse>() {});
+    DaprResponse res = result.block();
+
+    assertEquals(expected, new String((byte[]) res.getData()));
+  }
+
   @Test
   public void invokeServiceObjectTest() throws Exception {
     MyObject object = new MyObject(1, "Value");