Added support for calling polymorphic activities

temporalio · mfateev · Apr 4, 2020 · Apr 1, 2020 · Apr 2, 2020 · Apr 2, 2020
commit bac95464dd5efe7230bd69c9b0441c033439d006
@@ -32,14 +32,19 @@ class ActivityInvocationHandler extends ActivityInvocationHandlerBase {
   private final WorkflowCallsInterceptor activityExecutor;
 
   static InvocationHandler newInstance(
-      ActivityOptions options, WorkflowCallsInterceptor activityExecutor) {
-    return new ActivityInvocationHandler(options, activityExecutor);
+      Class<?> activityInterface,
+      ActivityOptions options,
+      WorkflowCallsInterceptor activityExecutor) {
+    return new ActivityInvocationHandler(activityInterface, activityExecutor, options);
   }
 
   private ActivityInvocationHandler(
-      ActivityOptions options, WorkflowCallsInterceptor activityExecutor) {
+      Class<?> activityInterface,
+      WorkflowCallsInterceptor activityExecutor,
+      ActivityOptions options) {
     this.options = options;
     this.activityExecutor = activityExecutor;
+    init(activityInterface);
   }
 
   @Override

@@ -20,17 +20,19 @@
 package io.temporal.internal.sync;
 
 import static io.temporal.internal.common.InternalUtils.getValueOrDefault;
+import static io.temporal.internal.sync.POJOActivityTaskHandler.getAnnotatedInterfaceMethodsFromInterface;
 
+import io.temporal.activity.ActivityInterface;
 import io.temporal.activity.ActivityMethod;
 import io.temporal.common.MethodRetry;
 import io.temporal.internal.common.InternalUtils;
 import io.temporal.internal.sync.AsyncInternal.AsyncMarker;
-import io.temporal.workflow.Workflow;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Function;
 
 /** Dynamic implementation of a strongly typed activity interface. */
@@ -47,33 +49,35 @@ static <T> T newProxy(Class<T> activityInterface, InvocationHandler invocationHa
             invocationHandler);
   }
 
+  protected void init(Class<?> activityInterface) {
+    Set<POJOActivityTaskHandler.MethodInterfacePair> activityMethods =
+        getAnnotatedInterfaceMethodsFromInterface(activityInterface, ActivityInterface.class);
+    if (activityMethods.isEmpty()) {
+      throw new IllegalArgumentException(
+          "Class doesn't implement any non empty interface annotated with @ActivityInterface: "
+              + activityInterface.getName());
+    }
+    for (POJOActivityTaskHandler.MethodInterfacePair pair : activityMethods) {
+      Method method = pair.getMethod();
+      ActivityMethod activityMethod = method.getAnnotation(ActivityMethod.class);
+      String activityType;
+      if (activityMethod != null && !activityMethod.name().isEmpty()) {
+        activityType = activityMethod.name();
+      } else {
+        activityType = InternalUtils.getSimpleName(pair.getType(), method);
+      }
+
+      MethodRetry methodRetry = method.getAnnotation(MethodRetry.class);
+      Function<Object[], Object> function = getActivityFunc(method, methodRetry, activityType);
+      methodFunctions.put(method, function);
+    }
+  }
+
   @Override
   public Object invoke(Object proxy, Method method, Object[] args) {
     Function<Object[], Object> function = methodFunctions.get(method);
     if (function == null) {
-      try {
-        if (method.equals(Object.class.getMethod("toString"))) {
-          // TODO: activity info
-          return "ActivityInvocationHandlerBase";
-        }
-        if (!method.getDeclaringClass().isInterface()) {
-          throw new IllegalArgumentException(
-              "Interface type is expected: " + method.getDeclaringClass());
-        }
-        MethodRetry methodRetry = method.getAnnotation(MethodRetry.class);
-        ActivityMethod activityMethod = method.getAnnotation(ActivityMethod.class);
-        String activityName;
-        if (activityMethod == null || activityMethod.name().isEmpty()) {
-          activityName = InternalUtils.getSimpleName(method);
-        } else {
-          activityName = activityMethod.name();
-        }
-
-        function = getActivityFunc(method, methodRetry, activityName);
-        methodFunctions.put(method, function);
-      } catch (NoSuchMethodException e) {
-        throw Workflow.wrap(e);
-      }
+      throw new IllegalArgumentException("Unexpected method: " + method);
     }
     return getValueOrDefault(function.apply(args), method.getReturnType());
   }

@@ -32,14 +32,19 @@ class LocalActivityInvocationHandler extends ActivityInvocationHandlerBase {
   private final WorkflowCallsInterceptor activityExecutor;
 
   static InvocationHandler newInstance(
-      LocalActivityOptions options, WorkflowCallsInterceptor activityExecutor) {
-    return new LocalActivityInvocationHandler(options, activityExecutor);
+      Class<?> activityInterface,
+      LocalActivityOptions options,
+      WorkflowCallsInterceptor activityExecutor) {
+    return new LocalActivityInvocationHandler(activityInterface, activityExecutor, options);
   }
 
   private LocalActivityInvocationHandler(
-      LocalActivityOptions options, WorkflowCallsInterceptor activityExecutor) {
+      Class<?> activityInterface,
+      WorkflowCallsInterceptor activityExecutor,
+      LocalActivityOptions options) {
     this.options = options;
     this.activityExecutor = activityExecutor;
+    init(activityInterface);
   }
 
   @Override

@@ -91,7 +91,7 @@ private void addActivityImplementation(
       }
     }
     Set<MethodInterfacePair> activityMethods =
-        getAnnotatedInterfaceMethods(cls, ActivityInterface.class);
+        getAnnotatedInterfaceMethodsFromImplementation(cls, ActivityInterface.class);
     if (activityMethods.isEmpty()) {
       throw new IllegalArgumentException(
           "Class doesn't implement any non empty interface annotated with @ActivityInterface: "
@@ -342,7 +342,7 @@ public int hashCode() {
     }
   }
 
-  Set<MethodInterfacePair> getAnnotatedInterfaceMethods(
+  static Set<MethodInterfacePair> getAnnotatedInterfaceMethodsFromImplementation(
       Class<?> implementationClass, Class<? extends Annotation> annotationClass) {
     if (implementationClass.isInterface()) {
       throw new IllegalArgumentException(
@@ -351,11 +351,33 @@ Set<MethodInterfacePair> getAnnotatedInterfaceMethods(
     Set<MethodInterfacePair> pairs = new HashSet<>();
     // Methods inherited from interfaces that are not annotated with @ActivityInterface
     Set<MethodWrapper> ignored = new HashSet<>();
-    getAnnotatedInterfaceMethods(implementationClass, annotationClass, ignored, pairs);
+    getAnnotatedInterfaceMethodsFromImplementation(
+        implementationClass, annotationClass, ignored, pairs);
     return pairs;
   }
 
-  private void getAnnotatedInterfaceMethods(
+  static Set<MethodInterfacePair> getAnnotatedInterfaceMethodsFromInterface(
+      Class<?> iClass, Class<? extends Annotation> annotationClass) {
+    if (!iClass.isInterface()) {
+      throw new IllegalArgumentException("Interface expected. Found: " + iClass.getSimpleName());
+    }
+    Annotation annotation = iClass.getAnnotation(annotationClass);
+    if (annotation == null) {
+      throw new IllegalArgumentException(
67ED          "@ActivityInterface annotation is required on the stub interface: "
+              + iClass.getSimpleName());
+    }
+    Set<MethodInterfacePair> pairs = new HashSet<>();
+    // Methods inherited from interfaces that are not annotated with @ActivityInterface
+    Set<MethodWrapper> ignored = new HashSet<>();
+    getAnnotatedInterfaceMethodsFromImplementation(iClass, annotationClass, ignored, pairs);
+    if (!ignored.isEmpty()) {
+      throw new IllegalStateException("Not empty ignored: " + ignored);
+    }
+    return pairs;
+  }
+
+  private static void getAnnotatedInterfaceMethodsFromImplementation(
       Class<?> current,
       Class<? extends Annotation> annotationClass,
       Set<MethodWrapper> methods,
@@ -372,7 +394,8 @@ private void getAnnotatedInterfaceMethods(
     Class<?>[] interfaces = current.getInterfaces();
     for (int i = 0; i < interfaces.length; i++) {
       Class<?> anInterface = interfaces[i];
-      getAnnotatedInterfaceMethods(anInterface, annotationClass, ourMethods, result);
+      getAnnotatedInterfaceMethodsFromImplementation(
+          anInterface, annotationClass, ourMethods, result);
     }
     Annotation annotation = current.getAnnotation(annotationClass);
     if (annotation == null) {

@@ -173,7 +173,8 @@ public <T> T newActivityStub(Class<T> activityInterface) {
             .setHeartbeatTimeout(Duration.ofSeconds(1))
             .build();
     InvocationHandler invocationHandler =
-        ActivityInvocationHandler.newInstance(options, new TestActivityExecutor());
+        ActivityInvocationHandler.newInstance(
+            activityInterface, options, new TestActivityExecutor());
     invocationHandler = new DeterministicRunnerWrapper(invocationHandler);
     return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
   }

@@ -151,7 +151,7 @@ public static long currentTimeMillis() {
   public static <T> T newActivityStub(Class<T> activityInterface, ActivityOptions options) {
     WorkflowCallsInterceptor decisionContext = WorkflowInternal.getWorkflowInterceptor();
     InvocationHandler invocationHandler =
-        ActivityInvocationHandler.newInstance(options, decisionContext);
+        ActivityInvocationHandler.newInstance(activityInterface, options, decisionContext);
     return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
   }
 
@@ -164,7 +164,7 @@ public static <T> T newLocalActivityStub(
       Class<T> activityInterface, LocalActivityOptions options) {
     WorkflowCallsInterceptor decisionContext = WorkflowInternal.getWorkflowInterceptor();
     InvocationHandler invocationHandler =
-        LocalActivityInvocationHandler.newInstance(options, decisionContext);
+        LocalActivityInvocationHandler.newInstance(activityInterface, options, decisionContext);
     return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
   }
 

@@ -28,6 +28,8 @@
 import io.temporal.testing.TestActivityEnvironment;
 import io.temporal.workflow.ActivityFailureException;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -249,4 +251,63 @@ public void testHeartbeatIntermittentError() throws InterruptedException {
     activity.activity1();
     assertEquals(3, count.get());
   }
+
+  public interface A {
+    void a();
+  }
+
+  @ActivityInterface
+  public interface B extends A {
+    void b();
+  }
+
+  @ActivityInterface
+  public interface C extends B, A {
+    void c();
+  }
+
+  public class CImpl implements C {
+    private List<String> invocations = new ArrayList<>();
+
+    @Override
+    public void a() {
+      invocations.add("a");
+    }
+
+    @Override
+    public void b() {
+      invocations.add("b");
+    }
+
+    @Override
+    public void c() {
+      invocations.add("c");
+    }
+  }
+
+  @Test
+  public void testInvokingActivityByBaseInterface() {
+    CImpl impl = new CImpl();
+    testEnvironment.registerActivitiesImplementations(impl);
+    try {
+      testEnvironment.newActivityStub(A.class);
+      fail("A doesn't implement activity");
+    } catch (IllegalArgumentException e) {
+      // expected as A doesn't implement any activity
+    }
+    B b = testEnvironment.newActivityStub(B.class);
+    b.a();
+    b.b();
+    C c = testEnvironment.newActivityStub(C.class);
+    c.a();
+    c.b();
+    c.c();
+    List<String> expected = new ArrayList<>();
+    expected.add("a");
+    expected.add("b");
+    expected.add("a");
+    expected.add("b");
+    expected.add("c");
+    assertEquals(expected, impl.invocations);
+  }
 }