Add support for Component classes for spring cloud stream plugin (#684)

springwolf · Apr 6, 2024 · 0eb0023 · 0eb0023
1 parent a5e772f
commit 0eb0023
Show file tree

Hide file tree

Showing 13 changed files with 410 additions and 123 deletions.
diff --git a/...ringwolf/bindings/googlepubsub/scanners/channels/GooglePubSubChannelBindingProcessor.java b/...ringwolf/bindings/googlepubsub/scanners/channels/GooglePubSubChannelBindingProcessor.java
@@ -11,7 +11,7 @@
 import org.springframework.context.EmbeddedValueResolverAware;
 import org.springframework.util.StringValueResolver;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Arrays;
 import java.util.Optional;
 
@@ -24,8 +24,8 @@ public void setEmbeddedValueResolver(StringValueResolver resolver) {
     }
 
     @Override
-    public Optional<ProcessedChannelBinding> process(Method method) {
-        return Arrays.stream(method.getAnnotations())
+    public Optional<ProcessedChannelBinding> process(AnnotatedElement annotatedElement) {
+        return Arrays.stream(annotatedElement.getAnnotations())
                 .filter(GooglePubSubAsyncChannelBinding.class::isInstance)
                 .map(GooglePubSubAsyncChannelBinding.class::cast)
                 .findAny()

diff --git a/...o/github/springwolf/core/asyncapi/scanners/bindings/channels/ChannelBindingProcessor.java b/...o/github/springwolf/core/asyncapi/scanners/bindings/channels/ChannelBindingProcessor.java
@@ -1,17 +1,17 @@
 // SPDX-License-Identifier: Apache-2.0
 package io.github.springwolf.core.asyncapi.scanners.bindings.channels;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Optional;
 
 public interface ChannelBindingProcessor {
 
     /**
-     * Process the methods annotated with Channel Binding Annotation
+     * Process the elements annotated with Channel Binding Annotation
      * for protocol specific channelBinding annotations, method parameters, etc
      *
-     * @param method The method being annotated
+     * @param annotatedElement The element being annotated
      * @return A message binding, if found
      */
-    Optional<ProcessedChannelBinding> process(Method method);
+    Optional<ProcessedChannelBinding> process(AnnotatedElement annotatedElement);
 }
diff --git a/...in/java/io/github/springwolf/core/asyncapi/scanners/common/utils/AsyncAnnotationUtil.java b/...in/java/io/github/springwolf/core/asyncapi/scanners/common/utils/AsyncAnnotationUtil.java
@@ -20,6 +20,7 @@
 import org.springframework.util.StringUtils;
 import org.springframework.util.StringValueResolver;
 
+import java.lang.reflect.AnnotatedElement;
 import java.lang.reflect.Method;
 import java.util.Arrays;
 import java.util.List;
@@ -149,9 +150,9 @@ public static List<String> getServers(AsyncOperation op, StringValueResolver res
     }
 
     public static Map<String, ChannelBinding> processChannelBindingFromAnnotation(
-            Method method, List<ChannelBindingProcessor> channelBindingProcessors) {
+            AnnotatedElement annotatedElement, List<ChannelBindingProcessor> channelBindingProcessors) {
         return channelBindingProcessors.stream()
-                .map(channelBindingProcessor -> channelBindingProcessor.process(method))
+                .map(channelBindingProcessor -> channelBindingProcessor.process(annotatedElement))
                 .filter(Optional::isPresent)
                 .map(Optional::get)
                 .collect(Collectors.toMap(

diff --git a/...hub/springwolf/core/asyncapi/scanners/bindings/processor/TestChannelBindingProcessor.java b/...hub/springwolf/core/asyncapi/scanners/bindings/processor/TestChannelBindingProcessor.java
@@ -13,7 +13,7 @@
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Arrays;
 import java.util.Optional;
 
@@ -24,8 +24,8 @@ public class TestChannelBindingProcessor implements ChannelBindingProcessor {
     public static final ChannelBinding BINDING = new EmptyChannelBinding();
 
     @Override
-    public Optional<ProcessedChannelBinding> process(Method method) {
-        return Arrays.stream(method.getAnnotations())
+    public Optional<ProcessedChannelBinding> process(AnnotatedElement annotatedElement) {
+        return Arrays.stream(annotatedElement.getAnnotations())
                 .filter(annotation -> annotation instanceof TestChannelBindingProcessor.TestChannelBinding)
                 .map(annotation -> (TestChannelBindingProcessor.TestChannelBinding) annotation)
                 .findAny()

diff --git a/.../src/main/java/io/github/springwolf/examples/cloudstream/configuration/ConsumerClass.java b/.../src/main/java/io/github/springwolf/examples/cloudstream/configuration/ConsumerClass.java
@@ -0,0 +1,18 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.springwolf.examples.cloudstream.configuration;
+
+import io.github.springwolf.examples.cloudstream.dtos.ExamplePayloadDto;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.function.Consumer;
+
+@Slf4j
+@Component
+public class ConsumerClass implements Consumer<ExamplePayloadDto> {
+
+    @Override
+    public void accept(ExamplePayloadDto payload) {
+        log.info("Called with payload: {}", payload);
+    }
+}
diff --git a/...ngwolf-examples/springwolf-cloud-stream-example/src/main/resources/application.properties b/...ngwolf-examples/springwolf-cloud-stream-example/src/main/resources/application.properties
@@ -10,6 +10,7 @@ spring.kafka.bootstrap-servers=${BOOTSTRAP_SERVER:localhost:29092}
 spring.cloud.stream.bindings.process-in-0.destination=example-topic
 spring.cloud.stream.bindings.process-out-0.destination=another-topic
 spring.cloud.stream.bindings.consumerMethod-in-0.destination=another-topic
+spring.cloud.stream.bindings.consumerClass-in-0.destination=consumer-class-topic
 spring.cloud.stream.bindings.googlePubSubConsumerMethod-in-0.destination=google-pubsub-topic
 
 

diff --git a/springwolf-examples/springwolf-cloud-stream-example/src/test/resources/asyncapi.json b/springwolf-examples/springwolf-cloud-stream-example/src/test/resources/asyncapi.json
@@ -32,6 +32,16 @@
         "kafka": { }
       }
     },
+    "consumer-class-topic": {
+      "messages": {
+        "io.github.springwolf.examples.cloudstream.dtos.ExamplePayloadDto": {
+          "$ref": "#/components/messages/io.github.springwolf.examples.cloudstream.dtos.ExamplePayloadDto"
+        }
+      },
+      "bindings": {
+        "kafka": { }
+      }
+    },
     "example-topic": {
       "messages": {
         "io.github.springwolf.examples.cloudstream.dtos.ExamplePayloadDto": {
@@ -256,6 +266,21 @@
         }
       ]
     },
+    "consumer-class-topic_publish_ConsumerClass": {
+      "action": "receive",
+      "channel": {
+        "$ref": "#/channels/consumer-class-topic"
+      },
+      "description": "Auto-generated description",
+      "bindings": {
+        "kafka": { }
+      },
+      "messages": [
+        {
+          "$ref": "#/channels/consumer-class-topic/messages/io.github.springwolf.examples.cloudstream.dtos.ExamplePayloadDto"
+        }
+      ]
+    },
     "example-topic_publish_process": {
       "action": "receive",
       "channel": {

diff --git a/...lf/plugins/cloudstream/asyncapi/scanners/channels/CloudStreamFunctionChannelsScanner.java b/...lf/plugins/cloudstream/asyncapi/scanners/channels/CloudStreamFunctionChannelsScanner.java
@@ -19,6 +19,7 @@
 import io.github.springwolf.core.asyncapi.scanners.beans.BeanMethodsScanner;
 import io.github.springwolf.core.asyncapi.scanners.bindings.channels.ChannelBindingProcessor;
 import io.github.springwolf.core.asyncapi.scanners.channels.ChannelMerger;
+import io.github.springwolf.core.asyncapi.scanners.classes.spring.ComponentClassScanner;
 import io.github.springwolf.core.asyncapi.scanners.common.utils.AsyncAnnotationUtil;
 import io.github.springwolf.core.configuration.docket.AsyncApiDocket;
 import io.github.springwolf.core.configuration.docket.AsyncApiDocketService;
@@ -28,7 +29,8 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -39,20 +41,26 @@ public class CloudStreamFunctionChannelsScanner implements ChannelsScanner {
 
     private final AsyncApiDocketService asyncApiDocketService;
     private final BeanMethodsScanner beanMethodsScanner;
+    private final ComponentClassScanner componentClassScanner;
     private final ComponentsService componentsService;
     private final BindingServiceProperties cloudStreamBindingsProperties;
     private final FunctionalChannelBeanBuilder functionalChannelBeanBuilder;
     protected final List<ChannelBindingProcessor> channelBindingProcessors;
 
     @Override
     public Map<String, ChannelObject> scan() {
-        Set<Method> beanMethods = beanMethodsScanner.getBeanMethods();
-        return ChannelMerger.mergeChannels(beanMethods.stream()
-                .map(functionalChannelBeanBuilder::fromMethodBean)
+        Set<AnnotatedElement> elements = new HashSet<>();
+        elements.addAll(componentClassScanner.scan());
+        elements.addAll(beanMethodsScanner.getBeanMethods());
+
+        List<Map.Entry<String, ChannelObject>> channels = elements.stream()
+                .map(functionalChannelBeanBuilder::build)
                 .flatMap(Set::stream)
                 .filter(this::isChannelBean)
                 .map(this::toChannelEntry)
-                .toList());
+                .toList();
+
+        return ChannelMerger.mergeChannels(channels);
     }
 
     private boolean isChannelBean(FunctionalChannelBeanData beanData) {
@@ -88,7 +96,7 @@ private ChannelObject buildChannel(FunctionalChannelBeanData beanData) {
                 .build();
         this.componentsService.registerMessage(message);
 
-        Map<String, ChannelBinding> channelBinding = buildChannelBinding(beanData.method());
+        Map<String, ChannelBinding> channelBinding = buildChannelBinding(beanData.annotatedElement());
         return ChannelObject.builder()
                 .bindings(channelBinding)
                 .messages(Map.of(message.getName(), MessageReference.toComponentMessage(message)))
@@ -101,9 +109,9 @@ private Map<String, MessageBinding> buildMessageBinding() {
         return Map.of(protocolName, new EmptyMessageBinding());
     }
 
-    private Map<String, ChannelBinding> buildChannelBinding(Method method) {
+    private Map<String, ChannelBinding> buildChannelBinding(AnnotatedElement annotatedElement) {
         Map<String, ChannelBinding> channelBindingMap =
-                AsyncAnnotationUtil.processChannelBindingFromAnnotation(method, channelBindingProcessors);
+                AsyncAnnotationUtil.processChannelBindingFromAnnotation(annotatedElement, channelBindingProcessors);
         if (!channelBindingMap.isEmpty()) {
             return channelBindingMap;
         }

diff --git a/...springwolf/plugins/cloudstream/asyncapi/scanners/common/FunctionalChannelBeanBuilder.java b/...springwolf/plugins/cloudstream/asyncapi/scanners/common/FunctionalChannelBeanBuilder.java
@@ -4,6 +4,7 @@
 import io.github.springwolf.core.asyncapi.scanners.common.payload.PayloadClassExtractor;
 import lombok.RequiredArgsConstructor;
 
+import java.lang.reflect.AnnotatedElement;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.util.Arrays;
@@ -12,6 +13,7 @@
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 import static java.util.stream.Collectors.toList;
@@ -20,41 +22,103 @@
 public class FunctionalChannelBeanBuilder {
     private final PayloadClassExtractor extractor;
 
-    public Set<FunctionalChannelBeanData> fromMethodBean(Method methodBean) {
-        Class<?> returnType = methodBean.getReturnType();
-        if (Consumer.class.isAssignableFrom(returnType)) {
-            Class<?> payloadType = getReturnTypeGenerics(methodBean).get(0);
-            return Set.of(ofConsumer(methodBean, payloadType));
+    public Set<FunctionalChannelBeanData> build(AnnotatedElement element) {
+        Class<?> type = getRawType(element);
+
+        if (Consumer.class.isAssignableFrom(type)) {
+            Class<?> payloadType = getTypeGenerics(element).get(0);
+            return Set.of(ofConsumer(element, payloadType));
         }
 
-        if (Supplier.class.isAssignableFrom(returnType)) {
-            Class<?> payloadType = getReturnTypeGenerics(methodBean).get(0);
-            return Set.of(ofSupplier(methodBean, payloadType));
+        if (Supplier.class.isAssignableFrom(type)) {
+            Class<?> payloadType = getTypeGenerics(element).get(0);
+            return Set.of(ofSupplier(element, payloadType));
         }
 
-        if (Function.class.isAssignableFrom(returnType)) {
-            Class<?> inputType = getReturnTypeGenerics(methodBean).get(0);
-            Class<?> outputType = getReturnTypeGenerics(methodBean).get(1);
+        if (Function.class.isAssignableFrom(type)) {
+            Class<?> inputType = getTypeGenerics(element).get(0);
+            Class<?> outputType = getTypeGenerics(element).get(1);
 
-            return Set.of(ofConsumer(methodBean, inputType), ofSupplier(methodBean, outputType));
+            return Set.of(ofConsumer(element, inputType), ofSupplier(element, outputType));
         }
 
         return Collections.emptySet();
     }
 
-    private static FunctionalChannelBeanData ofConsumer(Method method, Class<?> payloadType) {
+    private static Class<?> getRawType(AnnotatedElement element) {
+        if (element instanceof Method m) {
+            return m.getReturnType();
+        }
+
+        if (element instanceof Class<?> c) {
+            return c;
+        }
+
+        throw new IllegalArgumentException("Must be a Method or Class");
+    }
+
+    private static FunctionalChannelBeanData ofConsumer(AnnotatedElement element, Class<?> payloadType) {
+        String name = getElementName(element);
+        String cloudStreamBinding = firstCharToLowerCase(name) + "-in-0";
         return new FunctionalChannelBeanData(
-                method, payloadType, FunctionalChannelBeanData.BeanType.CONSUMER, method.getName() + "-in-0");
+                name, element, payloadType, FunctionalChannelBeanData.BeanType.CONSUMER, cloudStreamBinding);
     }
 
-    private static FunctionalChannelBeanData ofSupplier(Method method, Class<?> payloadType) {
+    private static FunctionalChannelBeanData ofSupplier(AnnotatedElement element, Class<?> payloadType) {
+        String name = getElementName(element);
+        String cloudStreamBinding = firstCharToLowerCase(name) + "-out-0";
         return new FunctionalChannelBeanData(
-                method, payloadType, FunctionalChannelBeanData.BeanType.SUPPLIER, method.getName() + "-out-0");
+                name, element, payloadType, FunctionalChannelBeanData.BeanType.SUPPLIER, cloudStreamBinding);
+    }
+
+    private static String firstCharToLowerCase(String name) {
+        return name.substring(0, 1).toLowerCase() + name.substring(1);
+    }
+
+    private static String getElementName(AnnotatedElement element) {
+        if (element instanceof Method m) {
+            return m.getName();
+        }
+
+        if (element instanceof Class<?> c) {
+            return c.getSimpleName();
+        }
+
+        throw new IllegalArgumentException("Must be a Method or Class");
+    }
+
+    private List<Class<?>> getTypeGenerics(AnnotatedElement element) {
+        if (element instanceof Method m) {
+            ParameterizedType genericReturnType = (ParameterizedType) m.getGenericReturnType();
+            return getTypeGenerics(genericReturnType);
+        }
+
+        if (element instanceof Class<?> c) {
+            return getTypeGenerics(c);
+        }
+
+        throw new IllegalArgumentException("Must be a Method or Class");
+    }
+
+    private List<Class<?>> getTypeGenerics(Class<?> c) {
+        Predicate<Class<?>> isConsumerPredicate = Consumer.class::isAssignableFrom;
+        Predicate<Class<?>> isSupplierPredicate = Supplier.class::isAssignableFrom;
+        Predicate<Class<?>> isFunctionPredicate = Function.class::isAssignableFrom;
+        Predicate<Class<?>> hasFunctionalInterfacePredicate =
+                isConsumerPredicate.or(isSupplierPredicate).or(isFunctionPredicate);
+
+        return Arrays.stream(c.getGenericInterfaces())
+                .filter(type -> type instanceof ParameterizedType)
+                .map(type -> (ParameterizedType) type)
+                .filter(type -> type.getRawType() instanceof Class<?>)
+                .filter(type -> hasFunctionalInterfacePredicate.test((Class<?>) type.getRawType()))
+                .map(this::getTypeGenerics)
+                .findFirst()
+                .orElse(Collections.emptyList());
     }
 
-    private List<Class<?>> getReturnTypeGenerics(Method methodBean) {
-        ParameterizedType genericReturnType = (ParameterizedType) methodBean.getGenericReturnType();
-        return Arrays.stream(genericReturnType.getActualTypeArguments())
+    private List<Class<?>> getTypeGenerics(ParameterizedType parameterizedType) {
+        return Arrays.stream(parameterizedType.getActualTypeArguments())
                 .map(extractor::typeToClass)
                 .collect(toList());
     }

diff --git a/...ub/springwolf/plugins/cloudstream/asyncapi/scanners/common/FunctionalChannelBeanData.java b/...ub/springwolf/plugins/cloudstream/asyncapi/scanners/common/FunctionalChannelBeanData.java
@@ -1,10 +1,21 @@
 // SPDX-License-Identifier: Apache-2.0
 package io.github.springwolf.plugins.cloudstream.asyncapi.scanners.common;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 
+/**
+ * @param elementName        The simple name of the element (Method or Class).
+ * @param annotatedElement   The element (Method or Class) from which this instance has been processed.
+ * @param payloadType        The payload type of the Channel this bean is bound to.
+ * @param beanType           Consumer or Supplier.
+ * @param cloudStreamBinding The expected binding string of this bean.
+ */
 public record FunctionalChannelBeanData(
-        Method method, Class<?> payloadType, BeanType beanType, String cloudStreamBinding) {
+        String elementName,
+        AnnotatedElement annotatedElement,
+        Class<?> payloadType,
+        BeanType beanType,
+        String cloudStreamBinding) {
 
     public enum BeanType {
         CONSUMER,