Extend springwolf-kafka-plugin to implement the new methods in abstract class and method level listener scanners to include an empty message binding so that there is always a message binding

Author: sam0r040

springwolf-plugins/springwolf-kafka-plugin/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/ClassLevelKafkaListenerScanner.java

@@ -1,6 +1,7 @@
 package io.github.stavshamir.springwolf.asyncapi.scanners.channels.annotation;
 
 import com.asyncapi.v2.binding.ChannelBinding;
+import com.asyncapi.v2.binding.MessageBinding;
 import com.asyncapi.v2.binding.OperationBinding;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelPriority;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelsScanner;
@@ -59,6 +60,13 @@ protected String getChannelName(KafkaListener annotation) {
         return KafkaListenerUtil.buildChannelBinding();
     }
 
+    @Override
+    protected Map<String, ? extends MessageBinding> buildMessageBinding(Method method) {
+        // Currently there is no interesting data in the KafkaListener annotation, but we keep it for the sake of
+        // consistency in the code and in the serialized specification (always have at least an empty binding for kafka)
+        return KafkaListenerUtil.buildMessageBinding();
+    }
+
     @Override
     protected AsyncHeaders buildHeaders(Method method) {
         Class<?> payloadType = getPayloadType(method);

springwolf-plugins/springwolf-kafka-plugin/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/KafkaListenerUtil.java

@@ -1,8 +1,10 @@
 package io.github.stavshamir.springwolf.asyncapi.scanners.channels.annotation;
 
 import com.asyncapi.v2.binding.ChannelBinding;
+import com.asyncapi.v2.binding.MessageBinding;
 import com.asyncapi.v2.binding.OperationBinding;
 import com.asyncapi.v2.binding.kafka.KafkaChannelBinding;
+import com.asyncapi.v2.binding.kafka.KafkaMessageBinding;
 import com.asyncapi.v2.binding.kafka.KafkaOperationBinding;
 import com.google.common.collect.ImmutableMap;
 import lombok.extern.slf4j.Slf4j;
@@ -44,4 +46,8 @@ public static String getChannelName(KafkaListener annotation, StringValueResolve
         binding.setGroupId(groupId);
         return ImmutableMap.of("kafka", binding);
     }
+
+    public static Map<String, ? extends MessageBinding> buildMessageBinding() {
+        return ImmutableMap.of("kafka", new KafkaMessageBinding());
+    }
 }

springwolf-plugins/springwolf-kafka-plugin/src/main/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/MethodLevelKafkaListenerScanner.java

@@ -1,6 +1,7 @@
 package io.github.stavshamir.springwolf.asyncapi.scanners.channels.annotation;
 
 import com.asyncapi.v2.binding.ChannelBinding;
+import com.asyncapi.v2.binding.MessageBinding;
 import com.asyncapi.v2.binding.OperationBinding;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelPriority;
 import io.github.stavshamir.springwolf.asyncapi.scanners.channels.ChannelsScanner;
@@ -49,6 +50,13 @@ protected String getChannelName(KafkaListener annotation) {
         return KafkaListenerUtil.buildOperationBinding(annotation, resolver);
     }
 
+    @Override
+    protected Map<String, ? extends MessageBinding> buildMessageBinding(KafkaListener annotation) {
+        // Currently there is no interesting data in the KafkaListener annotation, but we keep it for the sake of
+        // consistency in the code and in the serialized specification (always have at least an empty binding for kafka)
+        return KafkaListenerUtil.buildMessageBinding();
+    }
+
     @Override
     protected Class<?> getPayloadType(Method method) {
         return SpringPayloadAnnotationTypeExtractor.getPayloadType(method);

springwolf-plugins/springwolf-kafka-plugin/src/test/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/ClassLevelKafkaListenerScannerTest.java

@@ -1,6 +1,7 @@
 package io.github.stavshamir.springwolf.asyncapi.scanners.channels.annotation;
 
 import com.asyncapi.v2.binding.kafka.KafkaChannelBinding;
+import com.asyncapi.v2.binding.kafka.KafkaMessageBinding;
 import com.asyncapi.v2.binding.kafka.KafkaOperationBinding;
 import com.asyncapi.v2.model.channel.ChannelItem;
 import com.asyncapi.v2.model.channel.operation.Operation;
@@ -78,13 +79,15 @@ public void scan_componentWithMultipleKafkaListenersAndHandlers() {
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName("SpringKafkaDefaultHeaders-" + SimpleFoo.class.getSimpleName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Message barMessage = Message.builder()
                 .name(SimpleBar.class.getName())
                 .title(SimpleBar.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleBar.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName("SpringKafkaDefaultHeaders-" + SimpleBar.class.getSimpleName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()
@@ -140,6 +143,7 @@ public void scan_componentWithSingleKafkaHandlerMethod() {
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName("SpringKafkaDefaultHeaders-" + SimpleFoo.class.getSimpleName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()
@@ -172,13 +176,15 @@ public void scan_componentWithMultipleKafkaHandlerMethods() {
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName("SpringKafkaDefaultHeaders-" + SimpleFoo.class.getSimpleName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Message barMessage = Message.builder()
                 .name(SimpleBar.class.getName())
                 .title(SimpleBar.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleBar.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName("SpringKafkaDefaultHeaders-" + SimpleBar.class.getSimpleName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()
@@ -212,6 +218,7 @@ public void scan_componentWithSingleKafkaHandlerMethod_batchPayload() {
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName("SpringKafkaDefaultHeaders-" + SimpleFoo.class.getSimpleName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()

springwolf-plugins/springwolf-kafka-plugin/src/test/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/MethodLevelKafkaListenerScannerTest.java

@@ -2,6 +2,7 @@
 
 import com.asyncapi.v2.binding.OperationBinding;
 import com.asyncapi.v2.binding.kafka.KafkaChannelBinding;
+import com.asyncapi.v2.binding.kafka.KafkaMessageBinding;
 import com.asyncapi.v2.binding.kafka.KafkaOperationBinding;
 import com.asyncapi.v2.model.channel.ChannelItem;
 import com.asyncapi.v2.model.channel.operation.Operation;
@@ -80,6 +81,7 @@ public void scan_componentHasKafkaListenerMethods_hardCodedTopic() {
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName(AsyncHeaders.NOT_DOCUMENTED.getSchemaName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()
@@ -112,6 +114,7 @@ public void scan_componentHasKafkaListenerMethods_embeddedValueTopic() {
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName(AsyncHeaders.NOT_DOCUMENTED.getSchemaName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()
@@ -179,6 +182,7 @@ public void scan_componentHasKafkaListenerMethods_multipleParamsWithPayloadAnnot
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName(AsyncHeaders.NOT_DOCUMENTED.getSchemaName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()
@@ -211,6 +215,7 @@ public void scan_componentHasKafkaListenerMethods_batchPayload() {
                 .title(SimpleFoo.class.getSimpleName())
                 .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
                 .headers(HeaderReference.fromModelName(AsyncHeaders.NOT_DOCUMENTED.getSchemaName()))
+                .bindings(ImmutableMap.of("kafka", new KafkaMessageBinding()))
                 .build();
 
         Operation operation = Operation.builder()