Feature/spring cloud streams (#115)

* Simplify component scanning
* Add support for cloud stream functions

Author: stavshamir

.github/workflows/springwolf-cloud-stream.yml

@@ -0,0 +1,55 @@
+name: springwolf-cloud-stream
+
+on:
+  push:
+    branches:
+      - master
+    paths:
+      - '.github/workflows/springwolf-cloud-stream.yml'
+      - 'springwolf-core/**'
+      - 'springwolf-plugins/springwolf-cloud-stream-plugin/**'
+      - 'springwolf-examples/springwolf-cloud-stream-example/**'
+  pull_request:
+    types: [ opened, synchronize, ready_for_review ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Setup spectral
+        run: |
+          echo 'extends: ["spectral:asyncapi"]' > .spectral.yaml
+      - name: Lint asyncapi.json with spectral
+        uses: stoplightio/spectral-action@v0.8.8
+        with:
+          file_glob: 'springwolf-examples/springwolf-cloud-stream-example/src/test/resources/asyncapi.json'
+
+      - name: Set up JDK 1.8
+        uses: actions/setup-java@v1
+        with:
+          java-version: 1.8
+
+      - name: Setup Gradle
+        uses: gradle/gradle-build-action@v2
+
+      - name: Run unit tests
+        run: ./gradlew -p springwolf-plugins/springwolf-cloud-stream-plugin test
+
+      - name: Run integration tests
+        run: ./gradlew -p springwolf-examples/springwolf-cloud-stream-example test
+
+      - name: Publish package
+        if: github.ref == 'refs/heads/master'
+        run: ./gradlew publish -p springwolf-plugins/springwolf-cloud-stream-plugin
+        env:
+          ORG_GRADLE_PROJECT_SNAPSHOT: true
+
+          ORG_GRADLE_PROJECT_SIGNINGKEY: ${{secrets.ORG_GRADLE_PROJECT_SIGNINGKEY}}
+          ORG_GRADLE_PROJECT_SIGNINGPASSWORD: ${{secrets.ORG_GRADLE_PROJECT_SIGNINGPASSWORD}}
+
+          ORG_GRADLE_PROJECT_MAVEN_USERNAME: ${{ secrets.MAVEN_USERNAME }}
+          ORG_GRADLE_PROJECT_MAVEN_PASSWORD: ${{ secrets.MAVEN_PASSWORD }}

settings.gradle

@@ -4,8 +4,10 @@ include(
         'springwolf-core',
         'springwolf-plugins:springwolf-amqp-plugin',
         'springwolf-plugins:springwolf-kafka-plugin',
+        'springwolf-plugins:springwolf-cloud-stream-plugin',
         'springwolf-examples:springwolf-kafka-example',
         'springwolf-examples:springwolf-amqp-example',
+        'springwolf-examples:springwolf-cloud-stream-example',
         'springwolf-add-ons:springwolf-common-model-converters'
 )
 

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/DefaultAsyncApiSerializerService.java

@@ -9,9 +9,13 @@
 import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.module.SimpleModule;
+import io.github.stavshamir.springwolf.asyncapi.serializers.EmptyChannelBindingSerializer;
+import io.github.stavshamir.springwolf.asyncapi.serializers.EmptyOperationBindingSerializer;
 import io.github.stavshamir.springwolf.asyncapi.serializers.KafkaChannelBindingSerializer;
 import io.github.stavshamir.springwolf.asyncapi.serializers.KafkaOperationBindingSerializer;
 import io.github.stavshamir.springwolf.asyncapi.types.AsyncAPI;
+import io.github.stavshamir.springwolf.asyncapi.types.channel.bindings.EmptyChannelBinding;
+import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.bindings.EmptyOperationBinding;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.PostConstruct;
@@ -30,6 +34,8 @@ void postConstruct() {
 
     private void registerKafkaOperationBindingSerializer() {
         SimpleModule module = new SimpleModule();
+        module.addSerializer(EmptyChannelBinding.class, new EmptyChannelBindingSerializer());
+        module.addSerializer(EmptyOperationBinding.class, new EmptyOperationBindingSerializer());
         module.addSerializer(KafkaChannelBinding.class, new KafkaChannelBindingSerializer());
         module.addSerializer(KafkaOperationBinding.class, new KafkaOperationBindingSerializer());
         jsonMapper.registerModule(module);

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/beans/BeanMethodsScanner.java

@@ -0,0 +1,13 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.beans;
+
+import java.lang.reflect.Method;
+import java.util.Set;
+
+public interface BeanMethodsScanner {
+
+    /**
+     * @return Methods annotated with @Bean which are declared in @Configuration classes from the base package.
+     */
+    Set<Method> getBeanMethods();
+
+}

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/beans/DefaultBeanMethodsScanner.java

@@ -0,0 +1,36 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.beans;
+
+import io.github.stavshamir.springwolf.asyncapi.scanners.classes.ConfigurationClassScanner;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Service;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.toSet;
+
+@Service
+@RequiredArgsConstructor
+public class DefaultBeanMethodsScanner implements BeanMethodsScanner {
+
+    private final ConfigurationClassScanner configurationClassScanner;
+
+    @Override
+    public Set<Method> getBeanMethods() {
+        Set<Class<?>> configurationClasses = configurationClassScanner.scan();
+
+        Stream<Method> methods = configurationClasses.stream()
+                .map(Class::getDeclaredMethods)
+                .map(Arrays::asList)
+                .flatMap(List::stream);
+
+        return methods
+                .filter(method -> method.isAnnotationPresent(Bean.class))
+                .collect(toSet());
+    }
+
+}

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/AbstractClassLevelListenerScanner.java

@@ -7,11 +7,11 @@
 import com.google.common.collect.Maps;
 import io.github.stavshamir.springwolf.asyncapi.MessageHelper;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelsScanner;
+import io.github.stavshamir.springwolf.asyncapi.scanners.classes.ComponentClassScanner;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.Message;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.PayloadReference;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.header.AsyncHeaders;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.header.HeaderReference;
-import io.github.stavshamir.springwolf.configuration.AsyncApiDocket;
 import io.github.stavshamir.springwolf.schemas.SchemasService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -30,7 +30,7 @@
 public abstract class AbstractClassLevelListenerScanner<ClassAnnotation extends Annotation, MethodAnnotation extends Annotation> implements ChannelsScanner {
 
     @Autowired
-    private AsyncApiDocket docket;
+    private ComponentClassScanner componentClassScanner;
 
     @Autowired
     private SchemasService schemasService;
@@ -40,11 +40,14 @@ public abstract class AbstractClassLevelListenerScanner<ClassAnnotation extends
 
     /**
      * This annotation is used on class level
+     *
      * @return The class object of the listener annotation.
      */
     protected abstract Class<ClassAnnotation> getListenerAnnotationClass();
+
     /**
      * This annotation is used on the method level
+     *
      * @return The class object of the handler annotation.
      */
     protected abstract Class<MethodAnnotation> getHandlerAnnotationClass();
@@ -69,6 +72,7 @@ public abstract class AbstractClassLevelListenerScanner<ClassAnnotation extends
 
     /**
      * Can be overriden by implementations
+     *
      * @param method The specific method. Can be used to extract the payload type
      * @return The AsyncHeaders
      */
@@ -78,7 +82,7 @@ protected AsyncHeaders buildHeaders(Method method) {
 
     @Override
     public Map<String, ChannelItem> scan() {
-        Set<Class<?>> components = docket.getComponentsScanner().scanForComponents();
+        Set<Class<?>> components = componentClassScanner.scan();
         Set<Map.Entry<String, ChannelItem>> channels = mapToChannels(components);
         return mergeChannels(channels);
     }

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/AbstractMethodLevelListenerScanner.java

@@ -6,11 +6,11 @@
 import com.asyncapi.v2.model.channel.operation.Operation;
 import com.google.common.collect.Maps;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelsScanner;
+import io.github.stavshamir.springwolf.asyncapi.scanners.classes.ComponentClassScanner;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.Message;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.PayloadReference;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.header.AsyncHeaders;
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.header.HeaderReference;
-import io.github.stavshamir.springwolf.configuration.AsyncApiDocket;
 import io.github.stavshamir.springwolf.schemas.SchemasService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -26,14 +26,14 @@
 public abstract class AbstractMethodLevelListenerScanner<T extends Annotation> implements ChannelsScanner {
 
     @Autowired
-    private AsyncApiDocket docket;
+    private ComponentClassScanner componentClassScanner;
 
     @Autowired
     private SchemasService schemasService;
 
     @Override
     public Map<String, ChannelItem> scan() {
-        return docket.getComponentsScanner().scanForComponents().stream()
+        return componentClassScanner.scan().stream()
                 .map(this::getAnnotatedMethods).flatMap(Collection::stream)
                 .map(this::mapMethodToChannel)
                 .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/classes/AbstractAnnotatedClassScanner.java

@@ -0,0 +1,57 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.classes;
+
+import io.github.stavshamir.springwolf.configuration.AsyncApiDocket;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
+import org.springframework.core.type.filter.AnnotationTypeFilter;
+
+import java.lang.annotation.Annotation;
+import java.util.Optional;
+import java.util.Set;
+
+import static java.util.stream.Collectors.toSet;
+
+@Slf4j
+public abstract class AbstractAnnotatedClassScanner<T extends Annotation> implements ClassScanner {
+
+    @Autowired
+    private AsyncApiDocket docket;
+
+    /**
+     * @return The class object of the annotation to scan.
+     */
+    protected abstract Class<T> getAnnotationClass();
+
+    @Override
+    public Set<Class<?>> scan() {
+        String basePackage = docket.getBasePackage();
+        if (StringUtils.isBlank(basePackage)) {
+            throw new IllegalArgumentException("Base package must not be blank");
+        }
+
+        ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false);
+        provider.addIncludeFilter(new AnnotationTypeFilter(getAnnotationClass()));
+
+        log.debug("Scanning for {} classes in {}", getAnnotationClass().getSimpleName(), basePackage);
+        return provider.findCandidateComponents(basePackage).stream()
+                .map(BeanDefinition::getBeanClassName)
+                .map(this::getClass)
+                .filter(Optional::isPresent)
+                .map(Optional::get)
+                .collect(toSet());
+    }
+
+    private Optional<Class<?>> getClass(String className) {
+        try {
+            log.debug("Found candidate class: {}", className);
+            return Optional.of(Class.forName(className));
+        } catch (ClassNotFoundException e) {
+            log.warn("Failed to get class for name: {}", className);
+            return Optional.empty();
+        }
+    }
+
+}

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/components/ComponentsScanner.java renamed to springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/classes/ClassScanner.java

@@ -1,15 +1,15 @@
-package io.github.stavshamir.springwolf.asyncapi.scanners.components;
+package io.github.stavshamir.springwolf.asyncapi.scanners.classes;
 
 import java.util.Set;
 import io.github.stavshamir.springwolf.configuration.AsyncApiDocket;
 
-public interface ComponentsScanner {
+public interface ClassScanner {
 
     /**
      * Scan your project for components potentially containing asynchronous listeners,
      * based on the configuration of your registered {@link AsyncApiDocket}.
      *
      * @return A set of found classes.
      */
-    Set<Class<?>> scanForComponents();
+    Set<Class<?>> scan();
 }

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/classes/ComponentClassScanner.java

@@ -0,0 +1,14 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.classes;
+
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+
+@Service
+public class ComponentClassScanner extends AbstractAnnotatedClassScanner<Component> implements ClassScanner {
+
+    @Override
+    protected Class<Component> getAnnotationClass() {
+        return Component.class;
+    }
+
+}