(fix): added annotation for google pubsub message binding (#677)

* (fix): added annotation for google pubsub message binding * (fix): CloudStreamFunctionChannelsScanner constructor issue * (fix): updated cloudstream googlepubsub example and added missing binding from auto-configuration * (fix): update annotation package, fix typo and removed unwanted target and retention annotation * refactor(core): use AnnotatedElement in MessageBindingProcessor Continuation of 86960c6 --------- Co-authored-by: sheheryar aamir <[email protected]>
springwolf · Apr 6, 2024 · 56f4090 · 56f4090
1 parent 86960c6
commit 56f4090
Show file tree

Hide file tree

Showing 24 changed files with 215 additions and 61 deletions.
diff --git a/.../github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncChannelBinding.java b/.../github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncChannelBinding.java
@@ -20,7 +20,7 @@
 
     String messageRetentionDuration() default "";
 
-    GooglePubsubAsyncMessageStoragePolicy messageStoragePolicy() default @GooglePubsubAsyncMessageStoragePolicy;
+    GooglePubSubAsyncMessageStoragePolicy messageStoragePolicy() default @GooglePubSubAsyncMessageStoragePolicy;
 
     GooglePubSubAsyncSchemaSetting schemaSettings();
 

diff --git a/.../github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncMessageBinding.java b/.../github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncMessageBinding.java
@@ -0,0 +1,26 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.springwolf.bindings.googlepubsub.annotations;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * {@code @GooglePubSubAsyncMessageBinding} is a method-level annotation.
+ * It configures the message binding for the Google pubsub protocol.
+ * @see io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubMessageBinding
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(value = {ElementType.METHOD, ElementType.ANNOTATION_TYPE})
+@Inherited
+public @interface GooglePubSubAsyncMessageBinding {
+    String type() default "googlepubsub";
+
+    String orderingKey() default "";
+
+    GooglePubSubAsyncMessageSchema schema() default @GooglePubSubAsyncMessageSchema;
+
+    String bindingVersion() default "0.2.0";
+}
diff --git a/...o/github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncMessageSchema.java b/...o/github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncMessageSchema.java
@@ -0,0 +1,14 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.springwolf.bindings.googlepubsub.annotations;
+
+import java.lang.annotation.Inherited;
+
+/**
+ * {@code @GooglePubSubAsyncMessageSchema} is a method-level annotation.
+ * It configures the message schema for the Google pubsub protocol.
+ * @see io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubSchema
+ */
+@Inherited
+public @interface GooglePubSubAsyncMessageSchema {
+    String name() default "";
+}
diff --git a/...b/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncMessageStoragePolicy.java b/...b/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncMessageStoragePolicy.java
@@ -0,0 +1,9 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.springwolf.bindings.googlepubsub.annotations;
+
+import java.lang.annotation.Inherited;
+
+@Inherited
+public @interface GooglePubSubAsyncMessageStoragePolicy {
+    String[] allowedPersistenceRegions() default {};
+}
diff --git a/...o/github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncSchemaSetting.java b/...o/github/springwolf/bindings/googlepubsub/annotations/GooglePubSubAsyncSchemaSetting.java
@@ -3,17 +3,11 @@
 
 import io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubSchemaSettings;
 
-import java.lang.annotation.ElementType;
 import java.lang.annotation.Inherited;
-import java.lang.annotation.Retention;
-import java.lang.annotation.RetentionPolicy;
-import java.lang.annotation.Target;
 
 /**
  * @see GooglePubSubSchemaSettings
  */
-@Retention(RetentionPolicy.RUNTIME)
-@Target(value = {ElementType.METHOD})
 @Inherited
 public @interface GooglePubSubAsyncSchemaSetting {
     /**

diff --git a/...b/springwolf/bindings/googlepubsub/annotations/GooglePubsubAsyncMessageStoragePolicy.java b/...b/springwolf/bindings/googlepubsub/annotations/GooglePubsubAsyncMessageStoragePolicy.java
diff --git a/...f/bindings/googlepubsub/configuration/SpringwolfGooglePubSubBindingAutoConfiguration.java b/...f/bindings/googlepubsub/configuration/SpringwolfGooglePubSubBindingAutoConfiguration.java
@@ -2,6 +2,7 @@
 package io.github.springwolf.bindings.googlepubsub.configuration;
 
 import io.github.springwolf.bindings.googlepubsub.scanners.channels.GooglePubSubChannelBindingProcessor;
+import io.github.springwolf.bindings.googlepubsub.scanners.messages.GooglePubSubMessageBindingProcessor;
 import io.github.springwolf.core.asyncapi.scanners.bindings.BindingProcessorPriority;
 import io.github.springwolf.core.configuration.properties.SpringwolfConfigConstants;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
@@ -23,4 +24,11 @@ public class SpringwolfGooglePubSubBindingAutoConfiguration {
     public GooglePubSubChannelBindingProcessor googlePubSubChannelBindingProcessor() {
         return new GooglePubSubChannelBindingProcessor();
     }
+
+    @Bean
+    @Order(value = BindingProcessorPriority.PROTOCOL_BINDING)
+    @ConditionalOnMissingBean
+    public GooglePubSubMessageBindingProcessor googlePubSubMessageBindingProcessor() {
+        return new GooglePubSubMessageBindingProcessor();
+    }
 }
diff --git a/...ringwolf/bindings/googlepubsub/scanners/messages/GooglePubSubMessageBindingProcessor.java b/...ringwolf/bindings/googlepubsub/scanners/messages/GooglePubSubMessageBindingProcessor.java
@@ -0,0 +1,49 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.springwolf.bindings.googlepubsub.scanners.messages;
+
+import io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubMessageBinding;
+import io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubSchema;
+import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncMessageBinding;
+import io.github.springwolf.core.asyncapi.scanners.bindings.messages.MessageBindingProcessor;
+import io.github.springwolf.core.asyncapi.scanners.bindings.messages.ProcessedMessageBinding;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.context.EmbeddedValueResolverAware;
+import org.springframework.util.StringValueResolver;
+
+import java.lang.reflect.AnnotatedElement;
+import java.util.Arrays;
+import java.util.Optional;
+
+public class GooglePubSubMessageBindingProcessor implements MessageBindingProcessor, EmbeddedValueResolverAware {
+    private StringValueResolver resolver;
+
+    @Override
+    public void setEmbeddedValueResolver(StringValueResolver resolver) {
+        this.resolver = resolver;
+    }
+
+    @Override
+    public Optional<ProcessedMessageBinding> process(AnnotatedElement annotatedElement) {
+        return Arrays.stream(annotatedElement.getAnnotations())
+                .filter(GooglePubSubAsyncMessageBinding.class::isInstance)
+                .map(GooglePubSubAsyncMessageBinding.class::cast)
+                .findAny()
+                .map(this::mapToMessageBinding);
+    }
+
+    private ProcessedMessageBinding mapToMessageBinding(GooglePubSubAsyncMessageBinding bindingAnnotation) {
+        GooglePubSubSchema.GooglePubSubSchemaBuilder googlePubSubSchemaBuilder = GooglePubSubSchema.builder();
+        if (StringUtils.isNotBlank(bindingAnnotation.schema().name())) {
+            googlePubSubSchemaBuilder.name(bindingAnnotation.schema().name());
+        }
+        GooglePubSubMessageBinding.GooglePubSubMessageBindingBuilder bindingBuilder =
+                GooglePubSubMessageBinding.builder().schema(googlePubSubSchemaBuilder.build());
+        if (StringUtils.isNotBlank(bindingAnnotation.orderingKey())) {
+            bindingBuilder.orderingKey(bindingAnnotation.orderingKey());
+        }
+        if (StringUtils.isNotBlank(bindingAnnotation.bindingVersion())) {
+            bindingBuilder.bindingVersion(bindingAnnotation.bindingVersion());
+        }
+        return new ProcessedMessageBinding(bindingAnnotation.type(), bindingBuilder.build());
+    }
+}
diff --git a/...pringwolf-googlepubsub-binding/src/test/java/GooglePubSubChannelBindingProcessorTest.java b/...pringwolf-googlepubsub-binding/src/test/java/GooglePubSubChannelBindingProcessorTest.java
@@ -3,8 +3,8 @@
 import io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubMessageStoragePolicy;
 import io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubSchemaSettings;
 import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncChannelBinding;
+import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncMessageStoragePolicy;
 import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncSchemaSetting;
-import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubsubAsyncMessageStoragePolicy;
 import io.github.springwolf.bindings.googlepubsub.scanners.channels.GooglePubSubChannelBindingProcessor;
 import io.github.springwolf.core.asyncapi.scanners.bindings.channels.ProcessedChannelBinding;
 import org.junit.jupiter.api.Test;
@@ -52,7 +52,7 @@ void processWithoutAnnotationTest() throws NoSuchMethodException {
     @GooglePubSubAsyncChannelBinding(
             messageRetentionDuration = "messageRetentionDuration",
             messageStoragePolicy =
-                    @GooglePubsubAsyncMessageStoragePolicy(allowedPersistenceRegions = {"region1", "region2"}),
+                    @GooglePubSubAsyncMessageStoragePolicy(allowedPersistenceRegions = {"region1", "region2"}),
             schemaSettings =
                     @GooglePubSubAsyncSchemaSetting(
                             encoding = "BINARY",

diff --git a/...pringwolf-googlepubsub-binding/src/test/java/GooglePubSubMessageBindingProcessorTest.java b/...pringwolf-googlepubsub-binding/src/test/java/GooglePubSubMessageBindingProcessorTest.java
@@ -0,0 +1,51 @@
+// SPDX-License-Identifier: Apache-2.0
+import io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubMessageBinding;
+import io.github.springwolf.asyncapi.v3.bindings.googlepubsub.GooglePubSubSchema;
+import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncMessageBinding;
+import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncMessageSchema;
+import io.github.springwolf.bindings.googlepubsub.scanners.messages.GooglePubSubMessageBindingProcessor;
+import io.github.springwolf.core.asyncapi.scanners.bindings.messages.ProcessedMessageBinding;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class GooglePubSubMessageBindingProcessorTest {
+    private final GooglePubSubMessageBindingProcessor processor = new GooglePubSubMessageBindingProcessor();
+
+    @Test
+    void processTest() throws NoSuchMethodException {
+        // given
+        Method method = GooglePubSubMessageBindingProcessorTest.class.getMethod("methodWithAnnotation");
+
+        // when
+        ProcessedMessageBinding binding = processor.process(method).get();
+
+        // then
+        assertThat(binding.getType()).isEqualTo("googlepubsub");
+        assertThat(binding.getBinding())
+                .isEqualTo(
+                        new GooglePubSubMessageBinding(null, "key", new GooglePubSubSchema("project/test"), "0.2.0"));
+    }
+
+    @Test
+    void processWithoutAnnotationTest() throws NoSuchMethodException {
+        // given
+        Method method = GooglePubSubChannelBindingProcessorTest.class.getMethod("methodWithoutAnnotation");
+
+        // when
+        Optional<ProcessedMessageBinding> binding = processor.process(method);
+
+        // then
+        assertThat(binding).isNotPresent();
+    }
+
+    @GooglePubSubAsyncMessageBinding(
+            orderingKey = "key",
+            schema = @GooglePubSubAsyncMessageSchema(name = "project/test"))
+    public void methodWithAnnotation() {}
+
+    public void methodWithoutAnnotation() {}
+}
diff --git a/.../java/io/github/springwolf/bindings/sns/scanners/messages/SnsMessageBindingProcessor.java b/.../java/io/github/springwolf/bindings/sns/scanners/messages/SnsMessageBindingProcessor.java
@@ -8,7 +8,7 @@
 import org.springframework.context.EmbeddedValueResolverAware;
 import org.springframework.util.StringValueResolver;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Arrays;
 import java.util.Optional;
 
@@ -21,8 +21,8 @@ public void setEmbeddedValueResolver(StringValueResolver resolver) {
     }
 
     @Override
-    public Optional<ProcessedMessageBinding> process(Method method) {
-        return Arrays.stream(method.getAnnotations())
+    public Optional<ProcessedMessageBinding> process(AnnotatedElement annotatedElement) {
+        return Arrays.stream(annotatedElement.getAnnotations())
                 .filter(SnsAsyncOperationBinding.class::isInstance)
                 .map(SnsAsyncOperationBinding.class::cast)
                 .findAny()

diff --git a/.../java/io/github/springwolf/bindings/sqs/scanners/messages/SqsMessageBindingProcessor.java b/.../java/io/github/springwolf/bindings/sqs/scanners/messages/SqsMessageBindingProcessor.java
@@ -8,7 +8,7 @@
 import org.springframework.context.EmbeddedValueResolverAware;
 import org.springframework.util.StringValueResolver;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Arrays;
 import java.util.Optional;
 
@@ -21,8 +21,8 @@ public void setEmbeddedValueResolver(StringValueResolver resolver) {
     }
 
     @Override
-    public Optional<ProcessedMessageBinding> process(Method method) {
-        return Arrays.stream(method.getAnnotations())
+    public Optional<ProcessedMessageBinding> process(AnnotatedElement annotatedElement) {
+        return Arrays.stream(annotatedElement.getAnnotations())
                 .filter(SqsAsyncOperationBinding.class::isInstance)
                 .map(SqsAsyncOperationBinding.class::cast)
                 .findAny()

diff --git a/...o/github/springwolf/core/asyncapi/scanners/bindings/messages/MessageBindingProcessor.java b/...o/github/springwolf/core/asyncapi/scanners/bindings/messages/MessageBindingProcessor.java
@@ -4,17 +4,17 @@
 import io.github.springwolf.core.asyncapi.annotations.AsyncListener;
 import io.github.springwolf.core.asyncapi.annotations.AsyncPublisher;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Optional;
 
 public interface MessageBindingProcessor {
 
     /**
      * Process the methods annotated with {@link AsyncPublisher} and {@link AsyncListener}
-     * for protocol specific messageBinding annotations, method parameters, etc
+     * for protocol specific messageBinding annotations, annotatedElement parameters, etc
      *
-     * @param method The method being annotated
+     * @param annotatedElement The element being annotated
      * @return A message binding, if found
      */
-    Optional<ProcessedMessageBinding> process(Method method);
+    Optional<ProcessedMessageBinding> process(AnnotatedElement annotatedElement);
 }
diff --git a/...in/java/io/github/springwolf/core/asyncapi/scanners/common/utils/AsyncAnnotationUtil.java b/...in/java/io/github/springwolf/core/asyncapi/scanners/common/utils/AsyncAnnotationUtil.java
@@ -99,9 +99,9 @@ public static Map<String, OperationBinding> processOperationBindingFromAnnotatio
     }
 
     public static Map<String, MessageBinding> processMessageBindingFromAnnotation(
-            Method method, List<MessageBindingProcessor> messageBindingProcessors) {
+            AnnotatedElement annotatedElement, List<MessageBindingProcessor> messageBindingProcessors) {
         return messageBindingProcessors.stream()
-                .map(messageBindingProcessor -> messageBindingProcessor.process(method))
+                .map(messageBindingProcessor -> messageBindingProcessor.process(annotatedElement))
                 .filter(Optional::isPresent)
                 .map(Optional::get)
                 .collect(Collectors.toMap(

diff --git a/...hub/springwolf/core/asyncapi/scanners/bindings/processor/TestMessageBindingProcessor.java b/...hub/springwolf/core/asyncapi/scanners/bindings/processor/TestMessageBindingProcessor.java
@@ -13,7 +13,7 @@
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Arrays;
 import java.util.Optional;
 
@@ -24,8 +24,8 @@ public class TestMessageBindingProcessor implements MessageBindingProcessor {
     public static final MessageBinding BINDING = new EmptyMessageBinding();
 
     @Override
-    public Optional<ProcessedMessageBinding> process(Method method) {
-        return Arrays.stream(method.getAnnotations())
+    public Optional<ProcessedMessageBinding> process(AnnotatedElement annotatedElement) {
+        return Arrays.stream(annotatedElement.getAnnotations())
                 .filter(annotation -> annotation instanceof TestOperationBindingProcessor.TestOperationBinding)
                 .map(annotation -> (TestOperationBindingProcessor.TestOperationBinding) annotation)
                 .findAny()

diff --git a/...ava/io/github/springwolf/examples/cloudstream/configuration/CloudstreamConfiguration.java b/...ava/io/github/springwolf/examples/cloudstream/configuration/CloudstreamConfiguration.java
@@ -2,6 +2,8 @@
 package io.github.springwolf.examples.cloudstream.configuration;
 
 import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncChannelBinding;
+import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncMessageBinding;
+import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncMessageSchema;
 import io.github.springwolf.bindings.googlepubsub.annotations.GooglePubSubAsyncSchemaSetting;
 import io.github.springwolf.examples.cloudstream.dtos.AnotherPayloadDto;
 import io.github.springwolf.examples.cloudstream.dtos.ExamplePayloadDto;
@@ -31,6 +33,7 @@ public Consumer<AnotherPayloadDto> consumerMethod() {
 
     @GooglePubSubAsyncChannelBinding(
             schemaSettings = @GooglePubSubAsyncSchemaSetting(encoding = "BINARY", name = "project/test"))
+    @GooglePubSubAsyncMessageBinding(schema = @GooglePubSubAsyncMessageSchema(name = "project/test"))
     @Bean
     public Consumer<GooglePubSubPayloadDto> googlePubSubConsumerMethod() {
         return input -> log.info("Received new message in google-pubsub-topic: {}", input.toString());

diff --git a/springwolf-examples/springwolf-cloud-stream-example/src/test/resources/asyncapi.json b/springwolf-examples/springwolf-cloud-stream-example/src/test/resources/asyncapi.json
@@ -230,7 +230,12 @@
         "name": "io.github.springwolf.examples.cloudstream.dtos.GooglePubSubPayloadDto",
         "title": "GooglePubSubPayloadDto",
         "bindings": {
-          "kafka": { }
+          "googlepubsub": {
+            "schema": {
+              "name": "project/test"
+            },
+            "bindingVersion": "0.2.0"
+          }
         }
       }
     }
@@ -312,4 +317,4 @@
       ]
     }
   }
-}
+}
diff --git a/...ingwolf/plugins/amqp/asyncapi/scanners/bindings/messages/AmqpMessageBindingProcessor.java b/...ingwolf/plugins/amqp/asyncapi/scanners/bindings/messages/AmqpMessageBindingProcessor.java
@@ -8,7 +8,7 @@
 import org.springframework.context.EmbeddedValueResolverAware;
 import org.springframework.util.StringValueResolver;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.AnnotatedElement;
 import java.util.Arrays;
 import java.util.Optional;
 
@@ -21,8 +21,8 @@ public void setEmbeddedValueResolver(StringValueResolver resolver) {
     }
 
     @Override
-    public Optional<ProcessedMessageBinding> process(Method method) {
-        return Arrays.stream(method.getAnnotations())
+    public Optional<ProcessedMessageBinding> process(AnnotatedElement annotatedElement) {
+        return Arrays.stream(annotatedElement.getAnnotations())
                 .filter(AmqpAsyncOperationBinding.class::isInstance)
                 .map(AmqpAsyncOperationBinding.class::cast)
                 .findAny()