feat: resolve rabbit template dynamically by routing key (#1516)

* feat: enable rabbit template selection by routing key

* feat: test fallback scenario and change order

Author: sge93

springwolf-plugins/springwolf-amqp-plugin/src/main/java/io/github/springwolf/plugins/amqp/producer/SpringwolfAmqpProducer.java

@@ -12,21 +12,28 @@
 import org.springframework.util.CollectionUtils;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 @Slf4j
 public class SpringwolfAmqpProducer {
 
     private final AsyncApiService asyncApiService;
-    private final Optional<RabbitTemplate> rabbitTemplate;
+    private final Map<String, RabbitTemplate> rabbitTemplateRegistry;
 
     public boolean isEnabled() {
-        return rabbitTemplate.isPresent();
+        return !rabbitTemplateRegistry.isEmpty();
     }
 
     public SpringwolfAmqpProducer(AsyncApiService asyncApiService, List<RabbitTemplate> rabbitTemplates) {
         this.asyncApiService = asyncApiService;
-        this.rabbitTemplate = rabbitTemplates.isEmpty() ? Optional.empty() : Optional.of(rabbitTemplates.get(0));
+        this.rabbitTemplateRegistry = buildRabbitTemplateRegistry(rabbitTemplates);
+    }
+
+    private Map<String, RabbitTemplate> buildRabbitTemplateRegistry(List<RabbitTemplate> templates) {
+        return templates.stream()
+                .collect(Collectors.toMap(RabbitTemplate::getRoutingKey, rabbitTemplate -> rabbitTemplate));
     }
 
     public void send(String channelName, Object payload) {
@@ -44,6 +51,7 @@ public void send(String channelName, Object payload) {
             routingKey = channelName;
         }
 
+        Optional<RabbitTemplate> rabbitTemplate = getRabbitTemplate(routingKey);
         if (rabbitTemplate.isPresent()) {
             rabbitTemplate.get().convertAndSend(exchange, routingKey, payload);
         } else {
@@ -79,4 +87,9 @@ private String getRoutingKey(Operation operation) {
 
         return routingKey;
     }
+
+    private Optional<RabbitTemplate> getRabbitTemplate(String routingKey) {
+        return Optional.ofNullable(rabbitTemplateRegistry.get(routingKey))
+                .or(() -> rabbitTemplateRegistry.values().stream().findFirst());
+    }
 }

springwolf-plugins/springwolf-amqp-plugin/src/test/java/io/github/springwolf/plugins/amqp/producer/SpringwolfAmqpProducerTest.java

@@ -19,9 +19,11 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -114,4 +116,84 @@ void send_exchangeAndRoutingKeyFromBindings() {
 
         verify(rabbitTemplate).convertAndSend(eq("exchange-name"), eq("routing-key"), same(payload));
     }
+
+    @Test
+    void send_usesRabbitTemplateMatchingRoutingKey() {
+        RabbitTemplate template1 = mock(RabbitTemplate.class);
+        RabbitTemplate template2 = mock(RabbitTemplate.class);
+        when(template1.getRoutingKey()).thenReturn("any-key");
+        when(template2.getRoutingKey()).thenReturn("routing-key");
+
+        springwolfAmqpProducer = new SpringwolfAmqpProducer(asyncApiService, List.of(template1, template2));
+
+        AMQPChannelExchangeProperties properties = new AMQPChannelExchangeProperties();
+        properties.setName("exchange-name");
+        ChannelObject channelItem = ChannelObject.builder()
+                .bindings(Map.of(
+                        "amqp",
+                        AMQPChannelBinding.builder().exchange(properties).build()))
+                .build();
+        Map<String, ChannelObject> channels = Map.of("channel-name", channelItem);
+        Operation operation = Operation.builder()
+                .bindings(Map.of(
+                        "amqp",
+                        AMQPOperationBinding.builder()
+                                .cc(List.of("routing-key"))
+                                .build()))
+                .build();
+        Map<String, Operation> operations = Map.of("amqp", operation);
+
+        AsyncAPI asyncAPI = AsyncAPI.builder()
+                .info(new Info())
+                .channels(channels)
+                .operations(operations)
+                .build();
+        when(asyncApiService.getAsyncAPI()).thenReturn(asyncAPI);
+
+        Map<String, Object> payload = new HashMap<>();
+        springwolfAmqpProducer.send("channel-name", payload);
+
+        verify(template1, never()).convertAndSend(anyString(), anyString(), same(payload));
+        verify(template2).convertAndSend(eq("exchange-name"), eq("routing-key"), same(payload));
+    }
+
+    @Test
+    void send_fallsBackToFirstTemplateWhenNoMatch() {
+        RabbitTemplate template1 = mock(RabbitTemplate.class);
+        RabbitTemplate template2 = mock(RabbitTemplate.class);
+        when(template1.getRoutingKey()).thenReturn("key-1");
+        when(template2.getRoutingKey()).thenReturn("key-2");
+
+        springwolfAmqpProducer = new SpringwolfAmqpProducer(asyncApiService, List.of(template1, template2));
+
+        AMQPChannelExchangeProperties properties = new AMQPChannelExchangeProperties();
+        properties.setName("exchange-name");
+        ChannelObject channelItem = ChannelObject.builder()
+                .bindings(Map.of(
+                        "amqp",
+                        AMQPChannelBinding.builder().exchange(properties).build()))
+                .build();
+        Map<String, ChannelObject> channels = Map.of("channel-name", channelItem);
+        Operation operation = Operation.builder()
+                .bindings(Map.of(
+                        "amqp",
+                        AMQPOperationBinding.builder()
+                                .cc(List.of("routing-key"))
+                                .build()))
+                .build();
+        Map<String, Operation> operations = Map.of("amqp", operation);
+
+        AsyncAPI asyncAPI = AsyncAPI.builder()
+                .info(new Info())
+                .channels(channels)
+                .operations(operations)
+                .build();
+        when(asyncApiService.getAsyncAPI()).thenReturn(asyncAPI);
+
+        Map<String, Object> payload = new HashMap<>();
+        springwolfAmqpProducer.send("channel-name", payload);
+
+        verify(template1).convertAndSend(eq("exchange-name"), eq("routing-key"), same(payload));
+        verify(template2, never()).convertAndSend(anyString(), anyString(), same(payload));
+    }
 }