Added @ActivityInterface implementation to POJOActivityTaskHandler

Author: mfateev

src/main/java/com/uber/cadence/internal/common/InternalUtils.java

@@ -1,8 +1,8 @@
 /*
+ *  Modifications Copyright (c) 2017-2020 Uber Technologies Inc.
+ *  Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
  *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
- *  Modifications copyright (C) 2017 Uber Technologies, Inc.
- *
  *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
  *  use this file except in compliance with the License. A copy of the License is
  *  located at
@@ -51,7 +51,11 @@ public final class InternalUtils {
    * @return "Simple class name"::"methodName"
    */
   public static String getSimpleName(Method method) {
-    return method.getDeclaringClass().getSimpleName() + "::" + method.getName();
+    return getSimpleName(method.getDeclaringClass(), method);
+  }
+
+  public static String getSimpleName(Class<?> type, Method method) {
+    return type.getSimpleName() + "::" + method.getName();
   }
 
   public static String getWorkflowType(Method method, WorkflowMethod workflowMethod) {

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

@@ -1,8 +1,8 @@
 /*
+ *  Modifications Copyright (c) 2017-2020 Uber Technologies Inc.
+ *  Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
  *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
- *  Modifications copyright (C) 2017 Uber Technologies, Inc.
- *
  *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
  *  use this file except in compliance with the License. A copy of the License is
  *  located at
@@ -18,10 +18,11 @@
 package com.uber.cadence.internal.sync;
 
 import com.google.common.base.Joiner;
-import com.google.common.reflect.TypeToken;
+import com.google.common.base.Objects;
 import com.uber.cadence.PollForActivityTaskResponse;
 import com.uber.cadence.RespondActivityTaskCompletedRequest;
 import com.uber.cadence.RespondActivityTaskFailedRequest;
+import com.uber.cadence.activity.ActivityInterface;
 import com.uber.cadence.activity.ActivityMethod;
 import com.uber.cadence.client.ActivityCancelledException;
 import com.uber.cadence.common.MethodRetry;
@@ -33,11 +34,14 @@
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.testing.SimulatedTimeoutException;
 import com.uber.m3.tally.Scope;
+import java.lang.annotation.Annotation;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.BiFunction;
@@ -82,30 +86,29 @@ private void addActivityImplementation(
                 + "\" This annotation can be used only on the interface method it implements.");
       }
     }
-    TypeToken<?>.TypeSet interfaces = TypeToken.of(cls).getTypes().interfaces();
-    if (interfaces.isEmpty()) {
-      throw new IllegalArgumentException("Activity must implement at least one interface");
+    Set<MethodInterfacePair> activityMethods =
+        getAnnotatedInterfaceMethods(cls, ActivityInterface.class);
+    if (activityMethods.isEmpty()) {
+      throw new IllegalArgumentException(
+          "Class doesn't implement any non empty interface annotated with @ActivityInterface: "
+              + cls.getName());
     }
-    for (TypeToken<?> i : interfaces) {
-      if (i.getType().getTypeName().startsWith("org.mockito")) {
-        continue;
+    for (MethodInterfacePair pair : activityMethods) {
+      Method method = pair.getMethod();
+      ActivityMethod annotation = method.getAnnotation(ActivityMethod.class);
+      String activityType;
+      if (annotation != null && !annotation.name().isEmpty()) {
+        activityType = annotation.name();
+      } else {
+        activityType = InternalUtils.getSimpleName(pair.getType(), method);
       }
-      for (Method method : i.getRawType().getMethods()) {
-        ActivityMethod annotation = method.getAnnotation(ActivityMethod.class);
-        String activityType;
-        if (annotation != null && !annotation.name().isEmpty()) {
-          activityType = annotation.name();
-        } else {
-          activityType = InternalUtils.getSimpleName(method);
-        }
-        if (activities.containsKey(activityType)) {
-          throw new IllegalStateException(
-              activityType + " activity type is already registered with the worker");
-        }
-
-        ActivityTaskExecutor implementation = newTaskExecutor.apply(method, activity);
-        activities.put(activityType, implementation);
+      if (activities.containsKey(activityType)) {
+        throw new IllegalStateException(
+            activityType + " activity type is already registered with the worker");
       }
+
+      ActivityTaskExecutor implementation = newTaskExecutor.apply(method, activity);
+      activities.put(activityType, implementation);
     }
   }
 
@@ -266,4 +269,108 @@ public ActivityTaskHandler.Result execute(ActivityTaskImpl task, Scope metricsSc
   void setWorkflowService(IWorkflowService service) {
     this.service = service;
   }
+
+  static class MethodInterfacePair {
+    private final Method method;
+    private final Class<?> type;
+
+    MethodInterfacePair(Method method, Class<?> type) {
+      this.method = method;
+      this.type = type;
+    }
+
+    public Method getMethod() {
+      return method;
+    }
+
+    public Class<?> getType() {
+      return type;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      MethodInterfacePair that = (MethodInterfacePair) o;
+      return Objects.equal(method, that.method) && Objects.equal(type, that.type);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(method, type);
+    }
+
+    @Override
+    public String toString() {
+      return "MethodInterfacePair{" + "method=" + method + ", type=" + type + '}';
+    }
+  }
+
+  /** Used to override equals and hashCode of Method to ensure deduping by method name in a set. */
+  static class MethodWrapper {
+    private final Method method;
+
+    MethodWrapper(Method method) {
+      this.method = method;
+    }
+
+    public Method getMethod() {
+      return method;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      MethodWrapper that = (MethodWrapper) o;
+      return Objects.equal(method.getName(), that.method.getName());
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(method.getName());
+    }
+  }
+
+  Set<MethodInterfacePair> getAnnotatedInterfaceMethods(
+      Class<?> implementationClass, Class<? extends Annotation> annotationClass) {
+    if (implementationClass.isInterface()) {
+      throw new IllegalArgumentException(
+          "Concrete class expected. Found interface: " + implementationClass.getSimpleName());
+    }
+    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);
+    return pairs;
+  }
+
+  private void getAnnotatedInterfaceMethods(
+      Class<?> current,
+      Class<? extends Annotation> annotationClass,
+      Set<MethodWrapper> methods,
+      Set<MethodInterfacePair> result) {
+    // Using set to dedupe methods which are defined in both non activity parent and current
+    Set<MethodWrapper> ourMethods = new HashSet<>();
+    if (current.isInterface()) {
+      Method[] declaredMethods = current.getDeclaredMethods();
+      for (int i = 0; i < declaredMethods.length; i++) {
+        Method declaredMethod = declaredMethods[i];
+        ourMethods.add(new MethodWrapper(declaredMethod));
+      }
+    }
+    Class<?>[] interfaces = current.getInterfaces();
+    for (int i = 0; i < interfaces.length; i++) {
+      Class<?> anInterface = interfaces[i];
+      getAnnotatedInterfaceMethods(anInterface, annotationClass, ourMethods, result);
+    }
+    Annotation annotation = current.getAnnotation(annotationClass);
+    if (annotation == null) {
+      methods.addAll(ourMethods);
+      return;
+    }
+    for (MethodWrapper method : ourMethods) {
+      result.add(new MethodInterfacePair(method.getMethod(), current));
+    }
+  }
 }

src/test/java/com/uber/cadence/activity/LocalActivityContextPropagationTest.java

@@ -155,6 +155,7 @@ public String foo() {
     }
   }
 
+  @ActivityInterface
   public interface LocalActivityContextPropagation {
 
     @ActivityMethod

src/test/java/com/uber/cadence/internal/testing/ActivityTestingTest.java

@@ -26,6 +26,7 @@
 
 import com.uber.cadence.RecordActivityTaskHeartbeatResponse;
 import com.uber.cadence.activity.Activity;
+import com.uber.cadence.activity.ActivityInterface;
 import com.uber.cadence.client.ActivityCancelledException;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.testing.TestActivityEnvironment;
@@ -47,6 +48,7 @@ public void setUp() {
     testEnvironment = TestActivityEnvironment.newInstance();
   }
 
+  @ActivityInterface
   public interface TestActivity {
 
     String activity1(String input);
@@ -110,6 +112,7 @@ public void testHeartbeat() {
     assertEquals("details1", details.get());
   }
 
+  @ActivityInterface
   public interface InterruptibleTestActivity {
     void activity1() throws InterruptedException;
   }

src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java

@@ -36,6 +36,7 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowExecutionInfo;
 import com.uber.cadence.activity.Activity;
+import com.uber.cadence.activity.ActivityInterface;
 import com.uber.cadence.activity.ActivityOptions;
 import com.uber.cadence.client.*;
 import com.uber.cadence.context.ContextPropagator;
@@ -161,6 +162,7 @@ public void testFailure() {
     }
   }
 
+  @ActivityInterface
   public interface TestActivity {
     String activity1(String input);
   }
@@ -486,6 +488,7 @@ public void testConcurrentDecision() throws ExecutionException, InterruptedExcep
     log.info(testEnvironment.getDiagnostics());
   }
 
+  @ActivityInterface
   public interface TestCancellationActivity {
     String activity1(String input);
   }