changes

build.gradle.kts

@@ -39,6 +39,7 @@ dependencies {
 dependencies {
 	//web
 	implementation("org.springframework.boot:spring-boot-starter-web")
+	implementation("org.springframework.boot:spring-boot-starter-aop")
 
 	//monitoring
 	implementation("org.springframework.boot:spring-boot-starter-actuator")

gradle.properties

@@ -1,2 +1,2 @@
 group=org.goafabric
-version=3.3.1-SNAPSHOT
+version=3.3.1-aspect-SNAPSHOT

src/main/java/org/goafabric/eventdispatcher/consumer/nats/NatsListener.java

@@ -0,0 +1,14 @@
+package org.goafabric.eventdispatcher.consumer.nats;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface NatsListener {
+    String consumerName();
+    String subject();
+}
+

src/main/java/org/goafabric/eventdispatcher/consumer/nats/NatsListenerAspect.java

@@ -0,0 +1,42 @@
+package org.goafabric.eventdispatcher.consumer.nats;
+
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.annotation.Pointcut;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Aspect
+@Component
+public class NatsListenerAspect {
+
+    private final ConsumerUtil consumerUtil;
+
+    @Autowired
+    public NatsListenerAspect(ConsumerUtil consumerUtil) {
+        this.consumerUtil = consumerUtil;
+    }
+
+    @Pointcut("@annotation(org.goafabric.eventdispatcher.consumer.nats.NatsListener)")
+    public void natsListenerAnnotation() {}
+
+    @Around("natsListenerAnnotation()")
+    public Object invokeNatsListener(ProceedingJoinPoint joinPoint, NatsListener natsListener) throws Throwable {
+        String consumerName = natsListener.consumerName();
+        String subject = natsListener.subject();
+
+        // Subscribe using ConsumerUtil
+        consumerUtil.subscribe(consumerName, subject, (msg, eventData) -> {
+            try {
+                // Invoke the annotated method
+                joinPoint.proceed(new Object[]{msg, eventData});
+            } catch (Throwable throwable) {
+                throw new RuntimeException("Error invoking annotated method", throwable);
+            }
+        });
+
+        // Return null as we are handling the method execution manually
+        return null;
+    }
+}

src/main/java/org/goafabric/eventdispatcher/consumer/nats/NatsLoggerConsumer.java

@@ -12,14 +12,16 @@ public class NatsLoggerConsumer {
     private final Logger log = LoggerFactory.getLogger(this.getClass());
     private static final String CONSUMER_NAME = "Logger";
 
+    /*
     public NatsLoggerConsumer(ConsumerUtil consumerUtil) {
         consumerUtil.subscribe(CONSUMER_NAME, "*.*",
                 (msg, eventData) -> process(msg.getSubject(), eventData));
     }
+     */
 
-    private void process(String key, EventData eventData) {
+    @NatsListener(consumerName = CONSUMER_NAME, subject = "*.*")
+    public void process(String key, EventData eventData) {
         log.info("logging event: {}; id = {}", key, eventData.referenceId());
-        //msg.ack();
     }
 
 }