feat(amqp): differentiate between exchange and queue (#1014)

* feat(amqp): differentiate between exchange and queue * feat(jms): handle routing key for exchange * feat(amqp): update deps * test(amqp): update e2e channel name
springwolf · Oct 18, 2024 · 6810a81 · 6810a81
1 parent baf8f70
commit 6810a81
Show file tree

Hide file tree

Showing 28 changed files with 704 additions and 409 deletions.
diff --git a/...re/src/main/java/io/github/springwolf/core/asyncapi/scanners/bindings/BindingFactory.java b/...re/src/main/java/io/github/springwolf/core/asyncapi/scanners/bindings/BindingFactory.java
@@ -4,11 +4,16 @@
 import io.github.springwolf.asyncapi.v3.bindings.ChannelBinding;
 import io.github.springwolf.asyncapi.v3.bindings.MessageBinding;
 import io.github.springwolf.asyncapi.v3.bindings.OperationBinding;
+import io.github.springwolf.asyncapi.v3.model.ReferenceUtil;
 import io.github.springwolf.asyncapi.v3.model.schema.SchemaObject;
 
 import java.util.Map;
 
 public interface BindingFactory<T> {
+    default String getChannelId(T annotation) {
+        return ReferenceUtil.toValidId(getChannelName(annotation));
+    }
+
     String getChannelName(T annotation);
 
     Map<String, ChannelBinding> buildChannelBinding(T annotation);

diff --git a/...-core/src/main/java/io/github/springwolf/core/asyncapi/scanners/common/MessageHelper.java b/...-core/src/main/java/io/github/springwolf/core/asyncapi/scanners/common/MessageHelper.java
@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 package io.github.springwolf.core.asyncapi.scanners.common;
 
-import io.github.springwolf.asyncapi.v3.model.ReferenceUtil;
 import io.github.springwolf.asyncapi.v3.model.channel.message.MessageObject;
 import io.github.springwolf.asyncapi.v3.model.channel.message.MessageReference;
 import lombok.extern.slf4j.Slf4j;
@@ -28,14 +27,13 @@ public static Map<String, MessageReference> toMessagesMap(Set<MessageObject> mes
         return toMessageReferences(messages, aggregator);
     }
 
-    public static Map<String, MessageReference> toOperationsMessagesMap(
-            String channelName, Set<MessageObject> messages) {
-        if (channelName == null || channelName.isBlank()) {
-            throw new IllegalArgumentException("channelName must not be empty");
+    public static Map<String, MessageReference> toOperationsMessagesMap(String channelId, Set<MessageObject> messages) {
+        if (channelId == null || channelId.isBlank()) {
+            throw new IllegalArgumentException("channelId must not be empty");
         }
 
-        Function<MessageObject, MessageReference> aggregator = (message) ->
-                MessageReference.toChannelMessage(ReferenceUtil.toValidId(channelName), message.getMessageId());
+        Function<MessageObject, MessageReference> aggregator =
+                (message) -> MessageReference.toChannelMessage(channelId, message.getMessageId());
         return toMessageReferences(messages, aggregator);
     }
 

diff --git a/...hub/springwolf/core/asyncapi/scanners/common/message/SpringAnnotationMessagesService.java b/...hub/springwolf/core/asyncapi/scanners/common/message/SpringAnnotationMessagesService.java
@@ -49,8 +49,8 @@ public Map<String, MessageReference> buildMessages(
                 .collect(toSet());
 
         if (messageType == MessageType.OPERATION) {
-            String channelName = bindingFactory.getChannelName(classAnnotation);
-            return toOperationsMessagesMap(channelName, messages);
+            String channelId = bindingFactory.getChannelName(classAnnotation);
+            return toOperationsMessagesMap(channelId, messages);
         }
         return toMessagesMap(messages);
     }

diff --git a/.../springwolf/core/asyncapi/scanners/common/operation/SpringAnnotationOperationService.java b/.../springwolf/core/asyncapi/scanners/common/operation/SpringAnnotationOperationService.java
@@ -2,7 +2,6 @@
 package io.github.springwolf.core.asyncapi.scanners.common.operation;
 
 import io.github.springwolf.asyncapi.v3.bindings.OperationBinding;
-import io.github.springwolf.asyncapi.v3.model.ReferenceUtil;
 import io.github.springwolf.asyncapi.v3.model.channel.ChannelReference;
 import io.github.springwolf.asyncapi.v3.model.channel.message.MessageObject;
 import io.github.springwolf.asyncapi.v3.model.channel.message.MessageReference;
@@ -30,7 +29,7 @@ public Operation buildOperation(
         MessageObject message = springAnnotationMessageService.buildMessage(annotation, payloadType, headerSchema);
         Map<String, OperationBinding> operationBinding = bindingFactory.buildOperationBinding(annotation);
         Map<String, OperationBinding> opBinding = operationBinding != null ? new HashMap<>(operationBinding) : null;
-        String channelId = ReferenceUtil.toValidId(bindingFactory.getChannelName(annotation));
+        String channelId = bindingFactory.getChannelId(annotation);
 
         return Operation.builder()
                 .action(OperationAction.RECEIVE)

diff --git a/...asyncapi/scanners/operations/annotations/SpringAnnotationClassLevelOperationsScanner.java b/...asyncapi/scanners/operations/annotations/SpringAnnotationClassLevelOperationsScanner.java
@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 package io.github.springwolf.core.asyncapi.scanners.operations.annotations;
 
-import io.github.springwolf.asyncapi.v3.model.ReferenceUtil;
 import io.github.springwolf.asyncapi.v3.model.operation.Operation;
 import io.github.springwolf.asyncapi.v3.model.operation.OperationAction;
 import io.github.springwolf.core.asyncapi.scanners.bindings.BindingFactory;
@@ -44,8 +43,7 @@ private Stream<Map.Entry<String, Operation>> mapClassToOperation(
             Class<?> component, Set<MethodAndAnnotation<MethodAnnotation>> annotatedMethods) {
         ClassAnnotation classAnnotation = AnnotationUtil.findFirstAnnotationOrThrow(classAnnotationClass, component);
 
-        String channelName = bindingFactory.getChannelName(classAnnotation);
-        String channelId = ReferenceUtil.toValidId(channelName);
+        String channelId = bindingFactory.getChannelId(classAnnotation);
         String operationId =
                 StringUtils.joinWith("_", channelId, OperationAction.RECEIVE.type, component.getSimpleName());
 

diff --git a/...syncapi/scanners/operations/annotations/SpringAnnotationMethodLevelOperationsScanner.java b/...syncapi/scanners/operations/annotations/SpringAnnotationMethodLevelOperationsScanner.java
@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 package io.github.springwolf.core.asyncapi.scanners.operations.annotations;
 
-import io.github.springwolf.asyncapi.v3.model.ReferenceUtil;
 import io.github.springwolf.asyncapi.v3.model.operation.Operation;
 import io.github.springwolf.asyncapi.v3.model.operation.OperationAction;
 import io.github.springwolf.asyncapi.v3.model.schema.SchemaObject;
@@ -44,8 +43,7 @@ public Stream<Map.Entry<String, Operation>> scan(Class<?> clazz) {
     private Map.Entry<String, Operation> mapMethodToOperation(MethodAndAnnotation<MethodAnnotation> method) {
         MethodAnnotation annotation = AnnotationUtil.findFirstAnnotationOrThrow(methodAnnotationClass, method.method());
 
-        String channelName = bindingFactory.getChannelName(annotation);
-        String channelId = ReferenceUtil.toValidId(channelName);
+        String channelId = bindingFactory.getChannelId(annotation);
         String operationId = StringUtils.joinWith(
                 "_", channelId, OperationAction.RECEIVE.type, method.method().getName());
 

diff --git a/...ingwolf/core/asyncapi/scanners/common/operation/SpringAnnotationOperationServiceTest.java b/...ingwolf/core/asyncapi/scanners/common/operation/SpringAnnotationOperationServiceTest.java
@@ -40,7 +40,7 @@ class SpringAnnotationOperationServiceTest {
     @BeforeEach
     void setUp() {
         // when
-        when(bindingFactory.getChannelName(any())).thenReturn(CHANNEL_ID);
+        when(bindingFactory.getChannelId(any())).thenReturn(CHANNEL_ID);
         doReturn(defaultOperationBinding).when(bindingFactory).buildOperationBinding(any());
     }
 

diff --git a/...capi/scanners/operations/annotations/SpringAnnotationClassLevelOperationsScannerTest.java b/...capi/scanners/operations/annotations/SpringAnnotationClassLevelOperationsScannerTest.java
@@ -35,14 +35,14 @@ class SpringAnnotationClassLevelOperationsScannerTest {
                     springAnnotationOperationsService,
                     List.of(operationCustomizer));
 
-    private static final String CHANNEL_NAME = "test-channel";
+    private static final String CHANNEL_NAME_ID = "test-channel";
 
     private static final Map<String, MessageBinding> defaultMessageBinding =
             Map.of("protocol", new AMQPMessageBinding());
 
     @BeforeEach
     void setUp() {
-        when(bindingFactory.getChannelName(any())).thenReturn(CHANNEL_NAME);
+        when(bindingFactory.getChannelId(any())).thenReturn(CHANNEL_NAME_ID);
     }
 
     @Test
@@ -56,7 +56,7 @@ void scan() {
                 scanner.scan(ClassWithTestListenerAnnotation.class).toList();
 
         // then
-        String operationName = CHANNEL_NAME + "_receive_ClassWithTestListenerAnnotation";
+        String operationName = CHANNEL_NAME_ID + "_receive_ClassWithTestListenerAnnotation";
         assertThat(operations).containsExactly(Map.entry(operationName, operation));
     }
 

diff --git a/...api/scanners/operations/annotations/SpringAnnotationMethodLevelOperationsScannerTest.java b/...api/scanners/operations/annotations/SpringAnnotationMethodLevelOperationsScannerTest.java
@@ -37,11 +37,11 @@ class SpringAnnotationMethodLevelOperationsScannerTest {
                     springAnnotationOperationService,
                     List.of(operationCustomizer));
 
-    private static final String CHANNEL_NAME = "test-channel";
+    private static final String CHANNEL_ID = "test-channel";
 
     @BeforeEach
     void setUp() {
-        when(bindingFactory.getChannelName(any())).thenReturn(CHANNEL_NAME);
+        when(bindingFactory.getChannelId(any())).thenReturn(CHANNEL_ID);
     }
 
     @Test
@@ -56,7 +56,7 @@ void scan_componentHasTestListenerMethods() {
                 scanner.scan(ClassWithTestListenerAnnotation.class).toList();
 
         // then
-        String operationName = CHANNEL_NAME + "_receive_methodWithAnnotation";
+        String operationName = CHANNEL_ID + "_receive_methodWithAnnotation";
         assertThat(operations).containsExactly(Map.entry(operationName, operation));
     }
 

diff --git a/springwolf-examples/e2e/tests/publishing.spec.ts b/springwolf-examples/e2e/tests/publishing.spec.ts
@@ -72,8 +72,8 @@ function testPublishingEveryChannelItem() {
         messageTitle === "Message" || // Unable to instantiate ExamplePayloadProtobufDto$Message class
         messageTitle === "VehicleBase" || // Unable to publish abstract class for discriminator demo
         messageTitle.startsWith("GenericPayload") || // Unable to publish generic payload (amqp)
-        channelName === "#" || // Publishing through amqp exchange is not supported, see GH-366
-        channelName === "example-topic-routing-key" // Publishing through amqp exchange is not supported, see GH-366
+        channelName === "CRUD-topic-exchange-2" || // Publishing through amqp exchange is not supported, see GH-366
+        channelName === "example-topic-exchange_example-topic-routing-key" // Publishing through amqp exchange is not supported, see GH-366
       ) {
         return; // skip
       }

diff --git a/...e/src/main/java/io/github/springwolf/examples/amqp/configuration/RabbitConfiguration.java b/...e/src/main/java/io/github/springwolf/examples/amqp/configuration/RabbitConfiguration.java
@@ -2,6 +2,7 @@
 package io.github.springwolf.examples.amqp.configuration;
 
 import io.github.springwolf.examples.amqp.AmqpConstants;
+import io.github.springwolf.examples.amqp.dtos.AnotherPayloadDto;
 import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Exchange;
@@ -32,11 +33,6 @@ public Queue anotherQueue() {
         return new Queue(AmqpConstants.QUEUE_ANOTHER_QUEUE, false);
     }
 
-    @Bean
-    public Queue exampleBindingsQueue() {
-        return new Queue(AmqpConstants.QUEUE_EXAMPLE_BINDINGS_QUEUE, false, false, true);
-    }
-
     @Bean
     public Queue queueRead() {
         return new Queue(AmqpConstants.QUEUE_READ, false);
@@ -52,6 +48,17 @@ public Queue multiPayloadQueue() {
         return new Queue(AmqpConstants.QUEUE_MULTI_PAYLOAD_QUEUE);
     }
 
+    /**
+     * Defined by @RabbitListener annotation in {@link io.github.springwolf.examples.amqp.consumers.ExampleConsumer#bindingsExample(AnotherPayloadDto)}
+     */
+    @Bean
+    public Queue exampleBindingsQueue() {
+        return new Queue(AmqpConstants.QUEUE_EXAMPLE_BINDINGS_QUEUE, false, false, true);
+    }
+
+    /**
+     * Defined by @RabbitListener annotation in {@link io.github.springwolf.examples.amqp.consumers.ExampleConsumer#bindingsExample(AnotherPayloadDto)}
+     */
     @Bean
     public Binding exampleTopicBinding(Queue exampleBindingsQueue, Exchange exampleTopicExchange) {
         return BindingBuilder.bind(exampleBindingsQueue)

diff --git a/...p-example/src/main/java/io/github/springwolf/examples/amqp/consumers/ExampleConsumer.java b/...p-example/src/main/java/io/github/springwolf/examples/amqp/consumers/ExampleConsumer.java
@@ -55,7 +55,7 @@ public void receiveAnotherPayload(AnotherPayloadDto payload) {
                                         autoDelete = "true"),
                         key = AmqpConstants.ROUTING_KEY_EXAMPLE_TOPIC_ROUTING_KEY)
             })
-    public void bindingsExample(AnotherPayloadDto payload) {
+    public void bindingsExample(ExamplePayloadDto payload) {
         log.info(
                 "Received new message in {}" + " through exchange {}" + " using routing key {}: {}",
                 AmqpConstants.QUEUE_EXAMPLE_BINDINGS_QUEUE,
@@ -112,7 +112,7 @@ public void bindingsUpdate(Message message, @Payload GenericPayloadDto<ExamplePa
         log.info(
                 "Received new message {} in {} (GenericPayloadDto<ExamplePayloadDto>): {}",
                 message,
-                AmqpConstants.QUEUE_UPDATE,
+                AmqpConstants.EXCHANGE_CRUD_TOPIC_EXCHANGE_1,
                 payload.toString());
     }
 
@@ -130,7 +130,7 @@ public void bindingsRead(Message message, @Payload ExamplePayloadDto payload) {
         log.info(
                 "Received new message {} in {} (ExamplePayloadDto): {}",
                 message,
-                AmqpConstants.QUEUE_UPDATE,
+                AmqpConstants.EXCHANGE_CRUD_TOPIC_EXCHANGE_2,
                 payload.toString());
     }
 }
diff --git a/...xample/src/test/java/io/github/springwolf/examples/amqp/SpringContextIntegrationTest.java b/...xample/src/test/java/io/github/springwolf/examples/amqp/SpringContextIntegrationTest.java
@@ -46,12 +46,13 @@ void testContextWithApplicationProperties() {
         void testAllChannelsAreFound() {
             assertThat(asyncApiService.getAsyncAPI().getChannels().keySet())
                     .containsExactlyInAnyOrder(
-                            "#",
+                            "CRUD-topic-exchange-1",
+                            "CRUD-topic-exchange-2",
                             "another-queue",
                             "example-bindings-queue",
                             "example-queue",
                             "example-topic-exchange",
-                            "example-topic-routing-key",
+                            "example-topic-exchange_example-topic-routing-key",
                             "multi-payload-queue",
                             "queue-create",
                             "queue-delete",