Event loop, signalix objects

Author: satiracode

README.md

@@ -10,13 +10,13 @@ Add the following dependency to your project:
 <dependency>
     <groupId>org.exploit</groupId>
     <artifactId>signalix</artifactId>
-    <version>0.1.1</version>
+    <version>0.1.3</version>
 </dependency>
 ```
 
 ### Gradle
 ```groovy
-implementation("org.exploit:signalix:0.1.1")
+implementation("org.exploit:signalix:0.1.3")
 ```
 
 ### Usage
@@ -58,7 +58,7 @@ public class MyListener implements Listener {
 You can also define event listener in a functional way:
 ```java
 public static void main(String[] args) {
-    EventScope scope = new EventScope();
+    var scope = new EventScope();
 
     scope.registerEvent(MyEvent.class, (e) -> {
         System.out.println("Handling event: " + e.getSampleData());
@@ -67,7 +67,8 @@ public static void main(String[] args) {
 ```
 
 #### Calling events
-Finally, we need to create an instance of `EventScope`, register the listener, and call our event.
+Finally, we need to create an instance of `EventScope`, register the listener, and call our event. All events are called in a non-blocking manner in a single event loop thread.
+
 ```java
 import org.exploit.signalix.manager.EventScope;
 
@@ -81,12 +82,6 @@ public class Main {
 }
 ```
 
-To call event in a async way, use `callAsync`
-```java
-CompletableFuture<Void> call = scope.callAsync(new MyEvent(""));
-```
-___
-
 You can handle the same event in multiple methods.
 Event Handler's order can be controlled by priority, an event can extend `Cancellable` to stop further handlers execution.
 
@@ -145,5 +140,18 @@ public class MyListener implements Listener {
 }
 ```
 
+## Error handling
+In case of event execution error, `EventExecutionException` will be called as event:
+To handle the exception, simply define event handlers for `EventExecutionException`.
+```java
+@EventHandler
+public void onException(EventExecutionException exceptionEvent) {
+    var event = exceptionEvent.getEvent();
+    var exception = exceptionEvent.getException();
+    
+    // Handle exception
+}
+```
+
 ## License
 Signalix is distributed under the [BSD 2-Clause License](LICENSE.md)

build.gradle

@@ -9,6 +9,10 @@ repositories {
     mavenCentral()
 }
 
+tasks.withType(JavaCompile) {
+    options.compilerArgs << "-parameters"
+}
+
 dependencies {
     testImplementation platform('org.junit:junit-bom:5.9.1')
     testImplementation 'org.junit.jupiter:junit-jupiter'
@@ -39,7 +43,7 @@ allprojects {
     }
 
     group = 'org.exploit'
-    version = '0.1.1'
+    version = '0.1.3'
 
     afterEvaluate { p ->
         publishing {

settings.gradle

@@ -1,3 +1,4 @@
 rootProject.name = 'signalix'
 include 'signalix-kafka'
+include 'signalix-objects'
 

signalix-kafka/README.md

@@ -8,13 +8,13 @@ First add following dependency to your project:
 <dependency>
     <groupId>org.exploit</groupId>
     <artifactId>signalix-kafka</artifactId>
-    <version>0.1</version>
+    <version>0.1.3</version>
 </dependency>
 ```
 
 ### Gradle
 ```groovy
-implementation("org.exploit:signalix-kafka:0.1")
+implementation("org.exploit:signalix-kafka:0.1.3")
 ```
 
 ## Usage
@@ -60,7 +60,7 @@ Now simply register it as default listener:
 ```java
 public static void main(String[] args) {
     Map<String, String> kafkaProperties = new HashMap<String, String>();
-    KafkaEventScope kafkaEventScope = new KafkaEventScope(kafkaProperties);
+    var kafkaEventScope = new KafkaEventScope(kafkaProperties);
 
     kafkaEventScope.registerListener(new MyKafkaListener());
 }
@@ -101,7 +101,7 @@ Now you can call your event from KafkaEventScope:
 ```java
 public static void main(String[] args) {
     Map<String, String> kafkaProperties = new HashMap<String, String>();
-    KafkaEventScope kafkaEventScope = new KafkaEventScope(kafkaProperties);
+    var kafkaEventScope = new KafkaEventScope(kafkaProperties);
 
     kafkaEventScope.call(new MyEvent("Hello World!"));
 }

signalix-kafka/build.gradle

@@ -19,7 +19,7 @@ dependencies {
 
     compileOnly project(":")
 
-    implementation 'org.apache.kafka:kafka-clients:3.5.0'
+    implementation 'org.apache.kafka:kafka-clients:3.7.1'
     implementation 'org.slf4j:slf4j-api:2.0.12'
     implementation 'com.fasterxml.jackson.core:jackson-databind:2.15.2'
 }

signalix-kafka/src/main/java/org/exploit/signalix/KafkaEventScope.java

@@ -2,16 +2,18 @@
 
 import org.exploit.signalix.annotation.KafkaEvent;
 import org.exploit.signalix.annotation.KafkaEventListener;
+import org.exploit.signalix.core.KafkaEventManager;
 import org.exploit.signalix.event.listener.ConsumerEventListener;
 import org.exploit.signalix.event.listener.EventListener;
 import org.exploit.signalix.event.listener.ReflectedEventListener;
-import org.exploit.signalix.core.KafkaEventManager;
-import org.exploit.signalix.model.KafkaEventMeta;
-import org.exploit.signalix.utils.EventObjectMapper;
+import org.exploit.signalix.exception.EventExecutionException;
 import org.exploit.signalix.manager.EventScope;
 import org.exploit.signalix.marker.Event;
 import org.exploit.signalix.marker.Listener;
+import org.exploit.signalix.model.KafkaEventMeta;
+import org.exploit.signalix.model.KafkaWiredEvent;
 import org.exploit.signalix.utils.EventConsumer;
+import org.exploit.signalix.utils.EventObjectMapper;
 import org.jetbrains.annotations.NotNull;
 
 import java.util.List;
@@ -32,16 +34,6 @@ public KafkaEventScope(Map<String, String> properties) {
         this(convertToProperties(properties));
     }
 
-    @Override
-    public <T extends Event> void call(T event) {
-        var eventClass = event.getClass();
-
-        extractAnnotation(eventClass, KafkaEvent.class).ifPresentOrElse(
-                meta -> kafkaEventManager.call(event, new KafkaEventMeta(meta)),
-                () -> innerCall(event)
-        );
-    }
-
     public <T extends Event> void call(String topic, T event) {
         extractAnnotation(event.getClass(), KafkaEvent.class).ifPresentOrElse(
                 annotation -> {
@@ -52,8 +44,19 @@ public <T extends Event> void call(String topic, T event) {
         );
     }
 
+    @Override
+    public void call(Event event) {
+        extractAnnotation(event.getClass(), KafkaEvent.class).ifPresentOrElse(
+            annotation -> {
+                var meta = new KafkaEventMeta(annotation.topic(), annotation.partition(), List.of(annotation.headers()));
+                call(meta, event);
+            },
+            () -> super.call(event)
+        );
+    }
+
     public <T extends Event> void call(KafkaEventMeta meta, T event) {
-        kafkaEventManager.call(event, meta);
+        eventLoop.callEvent(new KafkaWiredEvent(event, meta));
     }
 
     public <T extends Event> void innerCall(T event) {
@@ -104,6 +107,19 @@ public void registerEventListener(EventListener listener) {
         kafkaEventManager.getEventObjectMapper().registerEvents(listener);
     }
 
+    @Override
+    protected void dispatchEvent(Event event) {
+        if (event instanceof KafkaWiredEvent kafkaEvent) {
+            try {
+                kafkaEventManager.call(kafkaEvent);
+            } catch (Exception e) {
+                call(new EventExecutionException(e, kafkaEvent.event()));
+            }
+        } else {
+            super.dispatchEvent(event);
+        }
+    }
+
     @Override
     public void close() {
         super.close();

signalix-kafka/src/main/java/org/exploit/signalix/core/KafkaEventConsumer.java

@@ -1,13 +1,13 @@
 package org.exploit.signalix.core;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import org.exploit.signalix.annotation.KafkaEventListener;
-import org.exploit.signalix.event.DeserializationFailureEvent;
-import org.exploit.signalix.utils.UtilityMethods;
-import org.exploit.signalix.marker.Event;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.errors.WakeupException;
+import org.exploit.signalix.annotation.KafkaEventListener;
+import org.exploit.signalix.event.DeserializationFailureEvent;
+import org.exploit.signalix.marker.Event;
+import org.exploit.signalix.utils.UtilityMethods;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

signalix-kafka/src/main/java/org/exploit/signalix/core/KafkaEventManager.java

@@ -2,9 +2,8 @@
 
 import org.exploit.signalix.KafkaEventScope;
 import org.exploit.signalix.annotation.KafkaEventListener;
-import org.exploit.signalix.model.KafkaEventMeta;
+import org.exploit.signalix.model.KafkaWiredEvent;
 import org.exploit.signalix.utils.EventObjectMapper;
-import org.exploit.signalix.marker.Event;
 
 import java.io.Closeable;
 import java.util.Properties;
@@ -28,8 +27,8 @@ public KafkaEventManager(Properties properties, KafkaEventScope eventScope) {
         this.eventProducer = new KafkaEventProducer(this);
     }
 
-    public <T extends Event> void call(T event, KafkaEventMeta meta) {
-        eventProducer.sendEvent(event, meta);
+    public void call(KafkaWiredEvent e) {
+        eventProducer.sendEvent(e);
     }
 
     public void registerKafkaListener(Class<?> clazz, KafkaEventListener meta) {

signalix-kafka/src/main/java/org/exploit/signalix/core/KafkaEventProducer.java

@@ -1,11 +1,9 @@
 package org.exploit.signalix.core;
 
-import org.exploit.signalix.exception.EventExecutionException;
-import org.exploit.signalix.model.KafkaEventMeta;
-import org.exploit.signalix.utils.UtilityMethods;
-import org.exploit.signalix.marker.Event;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.exploit.signalix.model.KafkaWiredEvent;
+import org.exploit.signalix.utils.UtilityMethods;
 
 public class KafkaEventProducer extends KafkaProducer<String, String> {
     private final KafkaEventManager eventManager;
@@ -15,8 +13,11 @@ public KafkaEventProducer(KafkaEventManager eventManager) {
         this.eventManager = eventManager;
     }
 
-    public <T extends Event> void sendEvent(T event, KafkaEventMeta meta) {
+    public void sendEvent(KafkaWiredEvent e) {
         try {
+            var event = e.event();
+            var meta = e.meta();
+
             var topic = meta.topic();
             var name = event.getClass().getName();
             var value = eventManager.getEventObjectMapper().writeValueAsString(event);
@@ -25,8 +26,8 @@ public <T extends Event> void sendEvent(T event, KafkaEventMeta meta) {
             var headers = UtilityMethods.extractHeaders(meta);
 
             send(new ProducerRecord<>(topic, partition, name, value, headers));
-        } catch (Exception e) {
-            throw new EventExecutionException(e.getMessage());
+        } catch (Exception exception) {
+            throw new IllegalStateException("Couldn't produce kafka event", exception);
         }
     }
 }

signalix-kafka/src/main/java/org/exploit/signalix/model/KafkaWiredEvent.java

@@ -0,0 +1,6 @@
+package org.exploit.signalix.model;
+
+import org.exploit.signalix.marker.Event;
+
+public record KafkaWiredEvent(Event event, KafkaEventMeta meta) implements Event {
+}

signalix-kafka/src/main/java/org/exploit/signalix/utils/UtilityMethods.java

@@ -1,7 +1,7 @@
 package org.exploit.signalix.utils;
 
-import org.exploit.signalix.model.KafkaEventMeta;
 import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.exploit.signalix.model.KafkaEventMeta;
 
 import java.lang.annotation.Annotation;
 import java.util.*;

signalix-kafka/src/test/java/org/exploit/signalix/kafka/test/WireSerializationTest.java

@@ -21,16 +21,16 @@ public void init() {
 
     @Test
     public void testSerialize() throws JsonProcessingException {
-        String expected = "{\"_name\":\"SampleEvent\",\"foo\":\"Hello world\"}";
-        String result = objectMapper.writeValueAsString(new SampleEvent("Hello world"));
+        var expected = "{\"_name\":\"SampleEvent\",\"foo\":\"Hello world\"}";
+        var result = objectMapper.writeValueAsString(new SampleEvent("Hello world"));
 
         assertEquals(expected, result);
     }
 
     @Test
     public void testDeserialize() throws JsonProcessingException {
-        String serialized = "{\"_name\":\"SampleEvent\",\"foo\":\"Hello world\"}";
-        Event event = objectMapper.readValue(serialized, Event.class);
+        var serialized = "{\"_name\":\"SampleEvent\",\"foo\":\"Hello world\"}";
+        var event = objectMapper.readValue(serialized, Event.class);
 
         assertEquals(event.getClass(), SampleEvent.class);
     }