Annotations for publisher and subscriber documentation (#118)

Author: timonback

.gitignore

@@ -1,8 +1,10 @@
 springwolf-core/build/
 springwolf-ui/build/
 springwolf-examples/springwolf-amqp-example/build/
+springwolf-examples/springwolf-cloud-stream-example/build/
 springwolf-examples/springwolf-kafka-example/build/
 springwolf-plugins/springwolf-amqp-plugin/build/
+springwolf-plugins/springwolf-cloud-stream-plugin/build/
 springwolf-plugins/springwolf-kafka-plugin/build/
 springwolf-add-ons/springwolf-common-model-converters/build/
 

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

@@ -1,5 +1,6 @@
 package io.github.stavshamir.springwolf.asyncapi;
 
+import com.asyncapi.v2.binding.amqp.AMQPOperationBinding;
 import com.asyncapi.v2.binding.kafka.KafkaChannelBinding;
 import com.asyncapi.v2.binding.kafka.KafkaOperationBinding;
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -11,6 +12,7 @@
 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.AmqpOperationBindingSerializer;
 import io.github.stavshamir.springwolf.asyncapi.serializers.KafkaChannelBindingSerializer;
 import io.github.stavshamir.springwolf.asyncapi.serializers.KafkaOperationBindingSerializer;
 import io.github.stavshamir.springwolf.asyncapi.types.AsyncAPI;
@@ -36,6 +38,7 @@ private void registerKafkaOperationBindingSerializer() {
         SimpleModule module = new SimpleModule();
         module.addSerializer(EmptyChannelBinding.class, new EmptyChannelBindingSerializer());
         module.addSerializer(EmptyOperationBinding.class, new EmptyOperationBindingSerializer());
+        module.addSerializer(AMQPOperationBinding.class, new AmqpOperationBindingSerializer());
         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/DefaultChannelsService.java

@@ -1,57 +1,40 @@
 package io.github.stavshamir.springwolf.asyncapi;
 
 import com.asyncapi.v2.model.channel.ChannelItem;
-import com.asyncapi.v2.model.channel.operation.Operation;
+import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelMerger;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelsScanner;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.PostConstruct;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
 
 @Slf4j
 @Service
 @RequiredArgsConstructor
 public class DefaultChannelsService implements ChannelsService {
 
-    private final Set<? extends ChannelsScanner> channelsScanners;
+    private final List<? extends ChannelsScanner> channelsScanners;
     private final Map<String, ChannelItem> channels = new TreeMap<>();
 
     @PostConstruct
     void findChannels() {
+        List<Map.Entry<String, ChannelItem>> foundChannelItems = new ArrayList<>();
 
         for (ChannelsScanner scanner : channelsScanners) {
             try {
                 Map<String, ChannelItem> channels = scanner.scan();
-                processFoundChannels(channels);
+                foundChannelItems.addAll(channels.entrySet());
             } catch (Exception e) {
                 log.error("An error was encountered during channel scanning with {}: {}", scanner, e.getMessage());
             }
         }
-    }
 
-    private void processFoundChannels(Map<String, ChannelItem> foundChannels) {
-        for (Map.Entry<String, ChannelItem> foundChannel: foundChannels.entrySet()) {
-            if (!this.channels.containsKey(foundChannel.getKey())) {
-                this.channels.put(foundChannel.getKey(), foundChannel.getValue());
-            } else {
-                ChannelItem existingChannel = this.channels.get(foundChannel.getKey());
-
-                Operation subscribeOperation = foundChannel.getValue().getSubscribe();
-                if(subscribeOperation != null) {
-                    existingChannel.setSubscribe(subscribeOperation);
-                }
-
-                Operation publishOperation = foundChannel.getValue().getPublish();
-                if(publishOperation != null) {
-                    existingChannel.setPublish(publishOperation);
-                }
-            }
-        }
+        this.channels.putAll(ChannelMerger.merge(foundChannelItems));
     }
 
     @Override

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

@@ -0,0 +1,63 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.channels;
+
+import com.asyncapi.v2.model.channel.ChannelItem;
+import com.asyncapi.v2.model.channel.operation.Operation;
+import io.github.stavshamir.springwolf.asyncapi.MessageHelper;
+import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.Message;
+
+import java.util.*;
+
+import static io.github.stavshamir.springwolf.asyncapi.MessageHelper.toMessageObjectOrComposition;
+
+
+/**
+ * Util to merge multiple {@link ChannelItem}s
+ */
+public class ChannelMerger {
+
+    /**
+     * Merges multiple channelItems by channel name
+     * <p>
+     * Given two channelItems for the same channel name, the first seen ChannelItem is used
+     * If an operation is null, the next non-null operation is used
+     * Messages within operations are merged
+     *
+     * @param channelEntries Ordered pairs of channel name to ChannelItem
+     * @return A map of channelName to a single ChannelItem
+     */
+    public static Map<String, ChannelItem> merge(List<Map.Entry<String, ChannelItem>> channelEntries) {
+        Map<String, ChannelItem> mergedChannels = new TreeMap<>();
+
+        for (Map.Entry<String, ChannelItem> entry : channelEntries) {
+            if (!mergedChannels.containsKey(entry.getKey())) {
+                mergedChannels.put(entry.getKey(), entry.getValue());
+            } else {
+                ChannelItem channelItem = mergedChannels.get(entry.getKey());
+                channelItem.setPublish(mergeOperation(channelItem.getPublish(), entry.getValue().getPublish()));
+                channelItem.setSubscribe(mergeOperation(channelItem.getSubscribe(), entry.getValue().getSubscribe()));
+            }
+        }
+
+        return mergedChannels;
+    }
+
+    private static Operation mergeOperation(Operation operation, Operation otherOperation) {
+        Set<Message> mergedMessages = getMessages(operation);
+        Set<Message> currentEntryMessages = getMessages(otherOperation);
+        mergedMessages.addAll(currentEntryMessages);
+
+        Operation mergedOperation = operation != null ? operation : otherOperation;
+        if (!mergedMessages.isEmpty()) {
+            mergedOperation.setMessage(toMessageObjectOrComposition(mergedMessages));
+        }
+        return mergedOperation;
+    }
+
+    private static Set<Message> getMessages(Operation operation) {
+        return Optional
+                .ofNullable(operation)
+                .map(Operation::getMessage)
+                .map(MessageHelper::messageObjectToSet)
+                .orElseGet(TreeSet::new);
+    }
+}

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

@@ -0,0 +1,28 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.channels;
+
+import io.github.stavshamir.springwolf.asyncapi.scanners.channels.operationdata.annotation.AsyncPublisher;
+import io.github.stavshamir.springwolf.asyncapi.scanners.channels.operationdata.annotation.AsyncSubscriber;
+import io.github.stavshamir.springwolf.configuration.AsyncApiDocket;
+
+public class ChannelPriority {
+    /**
+     * Manual defined channels have the highest priority
+     * <p>
+     * Example: Definition via {@link AsyncApiDocket}
+     */
+    public static final int MANUAL_DEFINED = 1;
+
+    /**
+     * Definition via custom annotations
+     * <p>
+     * Example: {@link AsyncPublisher}, {@link AsyncSubscriber}
+     */
+    public static final int ASYNC_ANNOTATION = 2;
+
+    /**
+     * Definitions found via spring listener annotations are used last.
+     * <p>
+     * Examples: Plugins like KafkaListener, etc
+     */
+    public static final int AUTO_DISCOVERED = 3;
+}

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

@@ -5,7 +5,7 @@
 import com.asyncapi.v2.model.channel.ChannelItem;
 import com.asyncapi.v2.model.channel.operation.Operation;
 import com.google.common.collect.Maps;
-import io.github.stavshamir.springwolf.asyncapi.MessageHelper;
+import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelMerger;
 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;
@@ -84,7 +84,7 @@ protected AsyncHeaders buildHeaders(Method method) {
     public Map<String, ChannelItem> scan() {
         Set<Class<?>> components = componentClassScanner.scan();
         Set<Map.Entry<String, ChannelItem>> channels = mapToChannels(components);
-        return mergeChannels(channels);
+        return ChannelMerger.merge(new ArrayList<>(channels));
     }
 
     private Set<Map.Entry<String, ChannelItem>> mapToChannels(Set<Class<?>> components) {
@@ -99,32 +99,6 @@ private boolean isClassAnnotated(Class<?> component) {
         return component.isAnnotationPresent(getListenerAnnotationClass());
     }
 
-    private Map<String, ChannelItem> mergeChannels(Set<Map.Entry<String, ChannelItem>> channelEntries) {
-        Map<String, ChannelItem> mergedChannels = new TreeMap<>();
-
-        for (Map.Entry<String, ChannelItem> entry : channelEntries) {
-            if (!mergedChannels.containsKey(entry.getKey())) {
-                mergedChannels.put(entry.getKey(), entry.getValue());
-            } else {
-                ChannelItem channelItem = mergedChannels.get(entry.getKey());
-                Set<Message> mergedMessages = getChannelMessages(channelItem);
-                Set<Message> currentEntryMessages = getChannelMessages(entry.getValue());
-                mergedMessages.addAll(currentEntryMessages);
-                channelItem.getPublish().setMessage(toMessageObjectOrComposition(mergedMessages));
-            }
-        }
-
-        return mergedChannels;
-    }
-
-    private Set<Message> getChannelMessages(ChannelItem channelItem) {
-        return Optional
-                .ofNullable(channelItem.getPublish())
-                .map(Operation::getMessage)
-                .map(MessageHelper::messageObjectToSet)
-                .orElseGet(TreeSet::new);
-    }
-
     private Optional<Map.Entry<String, ChannelItem>> mapClassToChannel(Class<?> component) {
         log.debug("Mapping class \"{}\" to channel", component.getName());
 

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/operationdata/AbstractOperationDataScanner.java

@@ -4,13 +4,15 @@
 import com.asyncapi.v2.binding.OperationBinding;
 import com.asyncapi.v2.model.channel.ChannelItem;
 import com.asyncapi.v2.model.channel.operation.Operation;
+import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelPriority;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelsScanner;
 import io.github.stavshamir.springwolf.asyncapi.types.OperationData;
 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.HeaderReference;
 import io.github.stavshamir.springwolf.schemas.SchemasService;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.annotation.Order;
 
 import java.util.List;
 import java.util.Map;
@@ -20,10 +22,13 @@
 import static java.util.stream.Collectors.*;
 
 @Slf4j
+@Order(value = ChannelPriority.MANUAL_DEFINED)
 public abstract class AbstractOperationDataScanner implements ChannelsScanner {
 
     protected abstract SchemasService getSchemaService();
+
     protected abstract List<OperationData> getOperationData();
+
     protected abstract OperationData.OperationType getOperationType();
 
     @Override
@@ -64,9 +69,9 @@ private ChannelItem buildChannel(List<OperationData> operationDataList) {
 
         ChannelItem.ChannelItemBuilder channelBuilder = ChannelItem.builder()
                 .bindings(channelBinding);
-        switch(getOperationType()) {
+        switch (getOperationType()) {
             case PUBLISH:
-                channelBuilder =  channelBuilder.publish(operation);
+                channelBuilder = channelBuilder.publish(operation);
                 break;
             case SUBSCRIBE:
                 channelBuilder = channelBuilder.subscribe(operation);

springwolf-core/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/operationdata/ProcessedOperationBinding.java

@@ -0,0 +1,10 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.channels.operationdata;
+
+import com.asyncapi.v2.binding.OperationBinding;
+import lombok.Data;
+
+@Data
+public class ProcessedOperationBinding {
+     private final String type;
+     private final OperationBinding binding;
+}

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

@@ -0,0 +1,69 @@
+package io.github.stavshamir.springwolf.asyncapi.scanners.channels.operationdata.annotation;
+
+import com.asyncapi.v2.binding.OperationBinding;
+import io.github.stavshamir.springwolf.asyncapi.scanners.channels.operationdata.ProcessedOperationBinding;
+import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.header.AsyncHeaderSchema;
+import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.header.AsyncHeaders;
+import org.apache.commons.lang3.StringUtils;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static java.util.stream.Collectors.groupingBy;
+
+class AsyncAnnotationScannerUtil {
+    public static AsyncHeaders getAsyncHeaders(AsyncOperation op) {
+        if (op.headers().values().length == 0) {
+            return AsyncHeaders.NOT_DOCUMENTED;
+        }
+
+        AsyncHeaders asyncHeaders = new AsyncHeaders(op.headers().schemaName());
+        Arrays.stream(op.headers().values())
+                .collect(groupingBy(AsyncOperation.Headers.Header::name))
+                .forEach((headerName, headers) -> {
+                    List<String> values = getHeaderValues(headers);
+                    String exampleValue = values.stream().findFirst().orElse(null);
+                    asyncHeaders.addHeader(
+                            AsyncHeaderSchema
+                                    .headerBuilder()
+                                    .headerName(headerName)
+                                    .description(getDescription(headers))
+                                    .enumValue(values)
+                                    .example(exampleValue)
+                                    .build()
+                    );
+                });
+
+        return asyncHeaders;
+    }
+
+    private static List<String> getHeaderValues(List<AsyncOperation.Headers.Header> value) {
+        return value
+                .stream()
+                .map(AsyncOperation.Headers.Header::value)
+                .sorted()
+                .collect(Collectors.toList());
+    }
+
+    private static String getDescription(List<AsyncOperation.Headers.Header> value) {
+        return value
+                .stream()
+                .map(AsyncOperation.Headers.Header::description)
+                .filter(StringUtils::isNotBlank)
+                .sorted()
+                .findFirst()
+                .orElse(null);
+    }
+
+    public static Map<String, OperationBinding> processBindingFromAnnotation(Method method, List<OperationBindingProcessor> operationBindingProcessors) {
+        return operationBindingProcessors.stream()
+                .map(operationBindingProcessor -> operationBindingProcessor.process(method))
+                .filter(Optional::isPresent)
+                .map(Optional::get)
+                .collect(Collectors.toMap(ProcessedOperationBinding::getType, ProcessedOperationBinding::getBinding));
+    }
+}