Added support for conversion of generic types (#179)

Author: mfateev

src/main/java/com/uber/cadence/client/WorkflowStub.java

@@ -18,6 +18,7 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.lang.reflect.Type;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -43,30 +44,64 @@ public interface WorkflowStub {
    * Returns workflow result potentially waiting for workflow to complete. Behind the scene this
    * call performs long poll on Cadence service waiting for workflow completion notification.
    *
-   * @param returnType class of the workflow return value
+   * @param resultClass class of the workflow return value
+   * @param resultType type of the workflow return value. Differs from resultClass for generic
+   *     types.
    * @param <R> type of the workflow return value
    * @return workflow return value
    */
-  <R> R getResult(Class<R> returnType);
+  <R> R getResult(Class<R> resultClass, Type resultType);
 
-  <R> CompletableFuture<R> getResultAsync(Class<R> returnType);
+  <R> CompletableFuture<R> getResultAsync(Class<R> resultClass, Type resultType);
+
+  /**
+   * Returns workflow result potentially waiting for workflow to complete. Behind the scene this
+   * call performs long poll on Cadence service waiting for workflow completion notification.
+   *
+   * @param resultClass class of the workflow return value
+   * @param <R> type of the workflow return value
+   * @return workflow return value
+   */
+  <R> R getResult(Class<R> resultClass);
+
+  <R> CompletableFuture<R> getResultAsync(Class<R> resultClass);
+
+  /**
+   * Returns workflow result potentially waiting for workflow to complete. Behind the scene this
+   * call performs long poll on Cadence service waiting for workflow completion notification.
+   *
+   * @param timeout maximum time to wait
+   * @param unit unit of timeout
+   * @param resultClass class of the workflow return value
+   * @param resultType type of the workflow return value. Differs from resultClass for generic
+   * @param <R> type of the workflow return value
+   * @return workflow return value
+   * @throws TimeoutException if workflow is not completed after the timeout time.
+   */
+  <R> R getResult(long timeout, TimeUnit unit, Class<R> resultClass, Type resultType)
+      throws TimeoutException;
 
   /**
    * Returns workflow result potentially waiting for workflow to complete. Behind the scene this
    * call performs long poll on Cadence service waiting for workflow completion notification.
    *
    * @param timeout maximum time to wait
    * @param unit unit of timeout
-   * @param returnType class of the workflow return value
+   * @param resultClass class of the workflow return value
    * @param <R> type of the workflow return value
    * @return workflow return value
    * @throws TimeoutException if workflow is not completed after the timeout time.
    */
-  <R> R getResult(long timeout, TimeUnit unit, Class<R> returnType) throws TimeoutException;
+  <R> R getResult(long timeout, TimeUnit unit, Class<R> resultClass) throws TimeoutException;
+
+  <R> CompletableFuture<R> getResultAsync(
+      long timeout, TimeUnit unit, Class<R> resultClass, Type resultType);
+
+  <R> CompletableFuture<R> getResultAsync(long timeout, TimeUnit unit, Class<R> resultClass);
 
-  <R> CompletableFuture<R> getResultAsync(long timeout, TimeUnit unit, Class<R> returnType);
+  <R> R query(String queryType, Class<R> resultClass, Object... args);
 
-  <R> R query(String queryType, Class<R> returnType, Object... args);
+  <R> R query(String queryType, Class<R> resultClass, Type resultType, Object... args);
 
   /** Request cancellation. */
   void cancel();

src/main/java/com/uber/cadence/converter/DataConverter.java

@@ -17,6 +17,8 @@
 
 package com.uber.cadence.converter;
 
+import java.lang.reflect.Type;
+
 /**
  * Used by the framework to serialize/deserialize method parameters that need to be sent over the
  * wire.
@@ -39,11 +41,13 @@ public interface DataConverter {
    * Implements conversion of a single value.
    *
    * @param content Serialized value to convert to a Java object.
+   * @param valueClass
+   * @param valueType
    * @return converted Java object
    * @throws DataConverterException if conversion of the data passed as parameter failed for any
    *     reason.
    */
-  <T> T fromData(byte[] content, Class<T> valueType) throws DataConverterException;
+  <T> T fromData(byte[] content, Class<T> valueClass, Type valueType) throws DataConverterException;
 
   /**
    * Implements conversion of an array of values of different types. Useful for deserializing
@@ -54,5 +58,5 @@ public interface DataConverter {
    * @throws DataConverterException if conversion of the data passed as parameter failed for any
    *     reason.
    */
-  Object[] fromDataArray(byte[] content, Class<?>... valueType) throws DataConverterException;
+  Object[] fromDataArray(byte[] content, Type... valueType) throws DataConverterException;
 }

src/main/java/com/uber/cadence/converter/DataConverterException.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.converter;
 
+import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 
@@ -27,11 +28,11 @@
 @SuppressWarnings("serial")
 public class DataConverterException extends RuntimeException {
 
-  private Class<?>[] valueTypes;
+  private Type[] valueTypes;
 
   private String content;
 
-  public DataConverterException(byte[] content, Class<?>[] valueTypes, Throwable cause) {
+  public DataConverterException(byte[] content, Type[] valueTypes, Throwable cause) {
     super(cause);
     this.valueTypes = valueTypes;
     setContent(content);

src/main/java/com/uber/cadence/converter/JsonDataConverter.java

@@ -33,6 +33,7 @@
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
 import java.util.function.Function;
 import java.util.regex.Matcher;
@@ -128,14 +129,15 @@ public byte[] toData(Object... values) throws DataConverterException {
   }
 
   @Override
-  public <T> T fromData(byte[] content, Class<T> valueType) throws DataConverterException {
+  public <T> T fromData(byte[] content, Class<T> valueClass, Type valueType)
+      throws DataConverterException {
     if (content == null) {
       return null;
     }
     try {
       // Deserialize thrift values.
-      if (TBase.class.isAssignableFrom(valueType)) {
-        T instance = valueType.getConstructor().newInstance();
+      if (TBase.class.isAssignableFrom(valueClass)) {
+        T instance = valueClass.getConstructor().newInstance();
         newThriftDeserializer().deserialize((TBase) instance, content);
         return instance;
       }
@@ -146,15 +148,15 @@ public <T> T fromData(byte[] content, Class<T> valueType) throws DataConverterEx
   }
 
   @Override
-  public Object[] fromDataArray(byte[] content, Class<?>... valueType)
-      throws DataConverterException {
+  public Object[] fromDataArray(byte[] content, Type... valueType) throws DataConverterException {
     try {
       if ((content == null || content.length == 0)
           && (valueType == null || valueType.length == 0)) {
         return EMPTY_OBJECT_ARRAY;
       }
       if (valueType.length == 1) {
-        return new Object[] {fromData(content, valueType[0])};
+        Object result = gson.fromJson(new String(content, StandardCharsets.UTF_8), valueType[0]);
+        return new Object[] {result};
       }
       JsonArray array = parser.parse(new String(content, StandardCharsets.UTF_8)).getAsJsonArray();
       Object[] result = new Object[valueType.length];

src/main/java/com/uber/cadence/internal/replay/ClockDecisionContext.java

@@ -195,7 +195,8 @@ void handleMarkerRecorded(HistoryEvent event) {
   int getVersion(String changeId, DataConverter converter, int minSupported, int maxSupported) {
     Predicate<byte[]> changeIdEquals =
         (bytesInEvent) -> {
-          MarkerData markerData = converter.fromData(bytesInEvent, MarkerData.class);
+          MarkerData markerData =
+              converter.fromData(bytesInEvent, MarkerData.class, MarkerData.class);
           return markerData.getId().equals(changeId);
         };
     decisions.addAllMissingVersionMarker(true, Optional.of(changeIdEquals));
@@ -214,7 +215,7 @@ int getVersion(String changeId, DataConverter converter, int minSupported, int m
     if (!result.isPresent()) {
       return WorkflowInternal.DEFAULT_VERSION;
     }
-    int version = converter.fromData(result.get(), Integer.class);
+    int version = converter.fromData(result.get(), Integer.class, Integer.class);
     validateVersion(changeId, version, minSupported, maxSupported);
     return version;
   }

src/main/java/com/uber/cadence/internal/replay/MarkerHandler.java

@@ -143,7 +143,8 @@ private Optional<byte[]> getMarkerDataFromHistory(
     if (!markerName.equals(name)) {
       return Optional.empty();
     }
-    MarkerData markerData = converter.fromData(attributes.getDetails(), MarkerData.class);
+    MarkerData markerData =
+        converter.fromData(attributes.getDetails(), MarkerData.class, MarkerData.class);
     // access count is used to not return data from the marker before the recorded number of calls
     if (!markerId.equals(markerData.getId())
         || markerData.getAccessCount() > expectedAcccessCount) {

src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandler.java

@@ -82,7 +82,10 @@ public Object invoke(Object proxy, Method method, Object[] args) {
 
         ActivityOptions mergedOptions = ActivityOptions.merge(activityMethod, methodRetry, options);
         ActivityStub stub = ActivityStubImpl.newInstance(mergedOptions, activityExecutor);
-        function = (a) -> stub.execute(activityName, method.getReturnType(), a);
+        function =
+            (a) ->
+                stub.execute(
+                    activityName, method.getReturnType(), method.getGenericReturnType(), a);
         methodFunctions.put(method, function);
       } catch (NoSuchMethodException e) {
         throw Workflow.wrap(e);

src/main/java/com/uber/cadence/internal/sync/ActivityStubImpl.java

@@ -23,6 +23,7 @@
 import com.uber.cadence.workflow.ActivityStub;
 import com.uber.cadence.workflow.Promise;
 import com.uber.cadence.workflow.WorkflowInterceptor;
+import java.lang.reflect.Type;
 
 /** Supports calling activity by name and arguments without its strongly typed interface. */
 class ActivityStubImpl implements ActivityStub {
@@ -42,11 +43,16 @@ private ActivityStubImpl(ActivityOptions options, WorkflowInterceptor activityEx
   }
 
   @Override
-  public <T> T execute(String activityName, Class<T> returnType, Object... args) {
-    Promise<T> result = executeAsync(activityName, returnType, args);
+  public <T> T execute(String activityName, Class<T> resultClass, Object... args) {
+    return execute(activityName, resultClass, resultClass, args);
+  }
+
+  @Override
+  public <T> T execute(String activityName, Class<T> resultClass, Type resultType, Object... args) {
+    Promise<T> result = executeAsync(activityName, resultClass, resultType, args);
     if (AsyncInternal.isAsync()) {
       AsyncInternal.setAsyncResult(result);
-      return Defaults.defaultValue(returnType);
+      return Defaults.defaultValue(resultClass);
     }
     try {
       return result.get();
@@ -60,7 +66,13 @@ public <T> T execute(String activityName, Class<T> returnType, Object... args) {
   }
 
   @Override
-  public <R> Promise<R> executeAsync(String activityName, Class<R> returnType, Object... args) {
-    return activityExecutor.executeActivity(activityName, returnType, args, options);
+  public <R> Promise<R> executeAsync(String activityName, Class<R> resultClass, Object... args) {
+    return executeAsync(activityName, resultClass, resultClass, args);
+  }
+
+  @Override
+  public <R> Promise<R> executeAsync(
+      String activityName, Class<R> resultClass, Type resultType, Object... args) {
+    return activityExecutor.executeActivity(activityName, resultClass, resultType, args, options);
   }
 }

src/main/java/com/uber/cadence/internal/sync/ChildWorkflowInvocationHandler.java

@@ -70,7 +70,7 @@ public Object invoke(Object proxy, Method method, Object[] args) {
               + "from @WorkflowMethod, @QueryMethod or @SignalMethod");
     }
     if (workflowMethod != null) {
-      return stub.execute(method.getReturnType(), args);
+      return stub.execute(method.getReturnType(), method.getGenericReturnType(), args);
     }
     if (queryMethod != null) {
       throw new UnsupportedOperationException(

src/main/java/com/uber/cadence/internal/sync/ChildWorkflowStubImpl.java

@@ -28,6 +28,7 @@
 import com.uber.cadence.workflow.Workflow;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import com.uber.cadence.workflow.WorkflowInterceptor.WorkflowResult;
+import java.lang.reflect.Type;
 import java.util.Objects;
 
 class ChildWorkflowStubImpl implements ChildWorkflowStub {
@@ -61,11 +62,16 @@ public ChildWorkflowOptions getOptions() {
   }
 
   @Override
-  public <R> R execute(Class<R> returnType, Object... args) {
-    Promise<R> result = executeAsync(returnType, args);
+  public <R> R execute(Class<R> resultClass, Object... args) {
+    return execute(resultClass, resultClass, args);
+  }
+
+  @Override
+  public <R> R execute(Class<R> resultClass, Type resultType, Object... args) {
+    Promise<R> result = executeAsync(resultClass, resultType, args);
     if (AsyncInternal.isAsync()) {
       AsyncInternal.setAsyncResult(result);
-      return Defaults.defaultValue(returnType);
+      return Defaults.defaultValue(resultClass);
     }
     try {
       return result.get();
@@ -78,9 +84,14 @@ public <R> R execute(Class<R> returnType, Object... args) {
   }
 
   @Override
-  public <R> Promise<R> executeAsync(Class<R> returnType, Object... args) {
+  public <R> Promise<R> executeAsync(Class<R> resultClass, Object... args) {
+    return executeAsync(resultClass, resultClass, args);
+  }
+
+  @Override
+  public <R> Promise<R> executeAsync(Class<R> resultClass, Type resultType, Object... args) {
     WorkflowResult<R> result =
-        decisionContext.executeChildWorkflow(workflowType, returnType, args, options);
+        decisionContext.executeChildWorkflow(workflowType, resultClass, resultType, args, options);
     execution.completeFrom(result.getWorkflowExecution());
     return result.getResult();
   }

src/main/java/com/uber/cadence/internal/sync/POJOActivityTaskHandler.java

@@ -184,7 +184,7 @@ public ActivityTaskHandler.Result execute(
       ActivityExecutionContext context =
           new ActivityExecutionContextImpl(service, domain, task, dataConverter, heartbeatExecutor);
       byte[] input = task.getInput();
-      Object[] args = dataConverter.fromDataArray(input, method.getParameterTypes());
+      Object[] args = dataConverter.fromDataArray(input, method.getGenericParameterTypes());
       CurrentActivityExecutionContext.set(context);
       try {
         Object result = method.invoke(activity, args);

src/main/java/com/uber/cadence/internal/sync/POJOWorkflowImplementationFactory.java

@@ -202,7 +202,7 @@ private class POJOWorkflowImplementation implements SyncWorkflowDefinition {
 
     @Override
     public byte[] execute(byte[] input) throws CancellationException, WorkflowExecutionException {
-      Object[] args = dataConverter.fromDataArray(input, workflowMethod.getParameterTypes());
+      Object[] args = dataConverter.fromDataArray(input, workflowMethod.getGenericParameterTypes());
       try {
         newInstance();
         Object result = workflowMethod.invoke(workflow, args);
@@ -279,7 +279,7 @@ public void processSignal(String signalName, byte[] input, long eventId) {
                 + signalHandlers.keySet());
         return;
       }
-      Object[] args = dataConverter.fromDataArray(input, signalMethod.getParameterTypes());
+      Object[] args = dataConverter.fromDataArray(input, signalMethod.getGenericParameterTypes());
       try {
         newInstance();
         signalMethod.invoke(workflow, args);