expose cron schedule entries in EventManager

build.gradle

@@ -13,7 +13,7 @@ ext {
     user = 'joerg'
     name = 'event'
     description = 'Event framework for Java (NIO paths, files, timers, journals)'
-    inceptionYear = '2021'
+    inceptionYear = '2024'
     url = 'https://xbib.org/' + user + '/' + name
     scmUrl = 'https://xbib.org/' + user + '/' + name
     scmConnection = 'scm:git:git://xbib.org/' + user + '/' + name + '.git'
@@ -28,20 +28,7 @@ subprojects {
     apply from: rootProject.file('gradle/compile/java.gradle')
     apply from: rootProject.file('gradle/test/junit5.gradle')
     apply from: rootProject.file('gradle/repositories/maven.gradle')
+    apply from: rootProject.file('gradle/publish/maven.gradle')
 }
 apply from: rootProject.file('gradle/publish/sonatype.gradle')
 apply from: rootProject.file('gradle/publish/forgejo.gradle')
-
-/*
-dependencies {
-    api libs.settings.api
-    implementation libs.net
-    implementation libs.time
-    implementation libs.datastructures.common
-    implementation libs.datastructures.json.tiny
-    implementation libs.netty.handler
-    implementation libs.reactivestreams
-    testImplementation libs.rxjava3
-    testImplementation libs.settings.datastructures.json
-}
-*/

event-api/src/main/java/org/xbib/event/Event.java

@@ -16,9 +16,7 @@ public interface Event {
 
     String getMessage();
 
-    Instant getCreated();
-
-    Instant getScheduledFor();
+    Instant getInstant();
 
     Payload getPayload();
 

event-api/src/main/java/org/xbib/event/GenericEvent.java

@@ -0,0 +1,8 @@
+package org.xbib.event;
+
+public interface GenericEvent extends Event {
+
+    GenericEvent setListener(Listener listener);
+
+    void received();
+}

event-api/src/main/java/org/xbib/event/Payload.java

@@ -1,16 +1,6 @@
 package org.xbib.event;
 
-import java.util.LinkedHashMap;
 import java.util.Map;
 
-@SuppressWarnings("serial")
-public class Payload extends LinkedHashMap<String, Object> {
-
-    public Payload() {
-        super();
-    }
-
-    public Payload(Map<String, Object> map) {
-        super(map);
-    }
+public interface Payload extends Map<String, Object> {
 }

event-common/build.gradle

@@ -1,7 +1,6 @@
 dependencies {
     api project(':event-api')
     api libs.settings.api
-    implementation libs.net
     implementation libs.time
     implementation libs.settings.datastructures.json
     implementation libs.datastructures.common

event-common/src/main/java/module-info.java

@@ -1,24 +1,26 @@
 import org.xbib.event.common.EventManagerService;
+import org.xbib.event.EventConsumer;
+import org.xbib.event.Event;
 
 module org.xbib.event.common {
+    requires java.logging;
+    requires org.xbib.event.api;
+    requires org.xbib.settings.api;
+    requires org.xbib.settings.datastructures.json;
+    requires org.xbib.time;
+    requires org.xbib.datastructures.common;
+    requires org.xbib.datastructures.tiny;
+    requires org.xbib.datastructures.json.tiny;
     exports org.xbib.event.bus;
     exports org.xbib.event.clock;
     exports org.xbib.event.common;
-    exports org.xbib.event.generic;
     exports org.xbib.event.log;
     exports org.xbib.event.path;
     exports org.xbib.event.persistence;
     exports org.xbib.event.timer;
+    exports org.xbib.event.util;
     exports org.xbib.event.wal;
+    uses Event;
+    uses EventConsumer;
     uses EventManagerService;
-    uses org.xbib.event.EventConsumer;
-    uses org.xbib.event.Event;
-    requires org.xbib.event.api;
-    requires org.xbib.settings.api;
-    requires org.xbib.settings.datastructures.json;
-    requires org.xbib.net;
-    requires org.xbib.time;
-    requires org.xbib.datastructures.common;
-    requires org.xbib.datastructures.json.tiny;
-    requires java.logging;
 }

event-common/src/main/java/org/xbib/event/clock/ClockEventManagerService.java

@@ -1,6 +1,8 @@
 package org.xbib.event.clock;
 
-import org.xbib.event.bus.EventBus;
+import java.util.List;
+import org.xbib.event.Event;
+import org.xbib.event.common.AbstractEventManagerService;
 import org.xbib.event.common.EventManager;
 import org.xbib.event.common.EventManagerService;
 import org.xbib.settings.Settings;
@@ -8,30 +10,29 @@ import org.xbib.time.schedule.CronExpression;
 import org.xbib.time.schedule.CronSchedule;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import org.xbib.time.schedule.Entry;
 
-public class ClockEventManagerService implements EventManagerService {
+public class ClockEventManagerService extends AbstractEventManagerService implements EventManagerService {
 
     private static final Logger logger = Logger.getLogger(ClockEventManagerService.class.getName());
 
-    private CronSchedule<Integer> cronSchedule;
+    private EventManager eventManager;
 
-    private List<String> suspended;
+    private CronSchedule<Boolean> cronSchedule;
 
     public ClockEventManagerService() {
+        super();
     }
 
+    @Override
     public ClockEventManagerService init(EventManager eventManager) {
-        Settings settings = eventManager.getSettings();
-        EventBus eventBus = eventManager.getEventBus();
-        this.suspended = new ArrayList<>();
+        this.eventManager = eventManager;
         ThreadFactory threadFactory = new ThreadFactory() {
             int n = 1;
             @Override
@@ -40,45 +41,44 @@ public class ClockEventManagerService implements EventManagerService {
             }
         };
         ScheduledExecutorService scheduledExecutorService =
-                Executors.newScheduledThreadPool(settings.getAsInt("pool.size", 2), threadFactory);
+                Executors.newScheduledThreadPool(eventManager.getSettings().getAsInt("pool.size", 2), threadFactory);
         this.cronSchedule = new CronSchedule<>(scheduledExecutorService);
-        for (Map.Entry<String, Settings> mapEntry : settings.getGroups("event.clock").entrySet()) {
+        for (Map.Entry<String, Settings> mapEntry : eventManager.getSettings().getGroups("event.clock").entrySet()) {
             String name = mapEntry.getKey();
             Settings entrySettings = mapEntry.getValue();
             if (entrySettings.getAsBoolean("enabled", true)) {
                 String entry = entrySettings.get("entry");
-                if (entry != null) {
+                String type = entrySettings.get("type", "clock");
+                if (entry != null && type != null) {
                     try {
-                        ClockEventService clockEventService = new ClockEventService(this, eventBus, name);
+                        ClockEventService clockEventService = new ClockEventService(this, name, type);
                         cronSchedule.add(name, CronExpression.parse(entry), clockEventService);
-                        logger.log(Level.INFO, "cron job " + name + " scheduled on " + entry);
+                        logger.log(Level.INFO, "cron job " + name + "with type " + type + " scheduled on " + entry);
                     } catch (Exception e) {
                         logger.log(Level.WARNING, "unable to schedule cron job " + mapEntry.getKey() + ", reason " + e.getMessage());
                     }
+                } else {
+                    logger.log(Level.WARNING, "clock event service in configuration is incompletely defined, name = " + name );
                 }
             } else {
-                logger.log(Level.WARNING, "clock event service " + name + " in configuration not enabled");
+                logger.log(Level.WARNING, "clock event service in configuration not enabled, name = " + name );
             }
         }
-        logger.log(Level.INFO, "entries = " + cronSchedule.getEntries());
         cronSchedule.start();
+        logger.log(Level.INFO, "after init: entries = " + cronSchedule.getEntries());
         return this;
     }
 
-    public List<String> getSuspended() {
-        return suspended;
-    }
-
-    public void suspend(String name) {
-        suspended.add(name);
-    }
-
-    public void resume(String name) {
-        suspended.remove(name);
-    }
-
     @Override
     public void shutdown() throws IOException {
         cronSchedule.close();
     }
+
+    public void publish(Event event) {
+        eventManager.publish(event);
+    }
+
+    public List<Entry<Boolean>> getCronScheduleEntries() {
+        return cronSchedule.getEntries();
+    }
 }

event-common/src/main/java/org/xbib/event/clock/ClockEventService.java

@@ -1,47 +1,48 @@
 package org.xbib.event.clock;
 
+import java.util.Objects;
 import java.util.concurrent.Callable;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.xbib.event.Event;
-import org.xbib.event.bus.EventBus;
 import org.xbib.event.common.EventManager;
 
-public class ClockEventService implements Callable<Integer> {
+public class ClockEventService implements Callable<Boolean> {
 
     private static final Logger logger = Logger.getLogger(ClockEventService.class.getName());
 
-    private final ClockEventManagerService manager;
-
-    private final EventBus eventBus;
+    private final ClockEventManagerService clockEventManagerService;
 
     private final String name;
 
-    public ClockEventService(ClockEventManagerService manager,
-                             EventBus eventBus,
-                             String name) {
-        this.manager = manager;
-        this.eventBus = eventBus;
+    private final String eventType;
+
+    public ClockEventService(ClockEventManagerService clockEventManagerService,
+                             String name,
+                             String eventType) {
+        this.clockEventManagerService = clockEventManagerService;
         this.name = name;
+        this.eventType = Objects.requireNonNull(eventType, "clock event type must not be null");
     }
 
     @Override
-    public Integer call() {
+    public Boolean call() {
         try {
-            if (manager.getSuspended().contains(name)) {
-                logger.log(Level.FINE, "clock event " + name + " suspended");
-                return 1;
+            if (clockEventManagerService.getSuspended().contains(name)) {
+                logger.log(Level.FINE, "clock event service " + name + " is suspended, unable to continue");
+                return false;
             } else {
                 Event clockEvent = EventManager.eventBuilder()
-                        .setType("clock")
+                        .setType(eventType)
+
                         .build();
-                eventBus.post(clockEvent);
-                return 0;
+                clockEventManagerService.publish(clockEvent);
+                return true;
             }
         } catch (Throwable t) {
             logger.log(Level.WARNING, t.getMessage(), t);
-            return 1;
+            return false;
         }
     }
 }

event-common/src/main/java/org/xbib/event/clock/SimpleClockEventConsumer.java

@@ -1,23 +0,0 @@
-package org.xbib.event.clock;
-
-import org.xbib.event.ClockEvent;
-import org.xbib.event.EventConsumer;
-
-import java.util.logging.Logger;
-import org.xbib.event.bus.AllowConcurrentEvents;
-import org.xbib.event.bus.Subscribe;
-
-public class SimpleClockEventConsumer implements EventConsumer {
-
-    private static final Logger logger = Logger.getLogger(SimpleClockEventConsumer.class.getName());
-
-    public SimpleClockEventConsumer() {
-    }
-
-    @Subscribe
-    @AllowConcurrentEvents
-    void onEvent(ClockEvent event) {
-        logger.info("received demo clock event, created = " + event.getCreated());
-    }
-
-}

event-common/src/main/java/org/xbib/event/common/AbstractEventManagerService.java

@@ -0,0 +1,25 @@
+package org.xbib.event.common;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+public abstract class AbstractEventManagerService {
+
+    private List<String> suspended;
+
+    public AbstractEventManagerService() {
+        this.suspended = new CopyOnWriteArrayList<>();
+    }
+
+    public void suspend(String name) {
+        suspended.add(name);
+    }
+
+    public void resume(String name) {
+        suspended.remove(name);
+    }
+
+    public List<String> getSuspended() {
+        return suspended;
+    }
+}

event-common/src/main/java/org/xbib/event/common/EventManager.java

@@ -3,6 +3,7 @@ package org.xbib.event.common;
 import java.lang.reflect.InvocationTargetException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.nio.file.StandardCopyOption;
 import java.nio.file.attribute.FileTime;
 import java.time.Instant;
@@ -11,16 +12,17 @@ import java.util.Map;
 import java.util.ServiceLoader;
 
 import org.xbib.datastructures.json.tiny.Json;
+import org.xbib.datastructures.json.tiny.JsonBuilder;
 import org.xbib.event.Event;
 import org.xbib.event.EventConsumer;
 import org.xbib.event.Listener;
-import org.xbib.event.Payload;
 import org.xbib.event.bus.AsyncEventBus;
 import org.xbib.event.bus.EventBus;
+import org.xbib.event.bus.Subscriber;
 import org.xbib.event.bus.SubscriberExceptionContext;
 import org.xbib.event.bus.SubscriberExceptionHandler;
+import org.xbib.event.bus.SubscriberRegistry;
 import org.xbib.event.clock.ClockEventManagerService;
-import org.xbib.event.generic.GenericEventManagerService;
 import org.xbib.event.path.FileFollowEventManagerService;
 import org.xbib.event.path.PathEventManagerService;
 import org.xbib.event.timer.TimerEventManagerService;
@@ -31,12 +33,15 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import org.xbib.time.schedule.Entry;
 
-public final class EventManager {
+public final class EventManager extends AbstractEventManagerService implements EventManagerService {
 
     private static final Logger logger = Logger.getLogger(EventManager.class.getName());
 
@@ -47,6 +52,7 @@ public final class EventManager {
     private final Map<Class<? extends EventManagerService>, EventManagerService> eventManagerServices;
 
     private EventManager(EventManagerBuilder builder) {
+        super();
         this.builder = builder;
         eventTypes.put("null", NullEvent.class);
         eventTypes.put("generic", GenericEventImpl.class);
@@ -59,7 +65,7 @@ public final class EventManager {
         }
         logger.log(Level.INFO, "installed events = " + eventTypes.keySet());
         this.eventManagerServices = new HashMap<>();
-        eventManagerServices.put(GenericEventManagerService.class, new GenericEventManagerService().init(this));
+        eventManagerServices.put(this.getClass(), this);
         eventManagerServices.put(ClockEventManagerService.class, new ClockEventManagerService().init(this));
         eventManagerServices.put(TimerEventManagerService.class, new TimerEventManagerService().init(this));
         eventManagerServices.put(FileFollowEventManagerService.class, new FileFollowEventManagerService().init(this));
@@ -70,6 +76,70 @@ public final class EventManager {
         logger.log(Level.INFO, "installed event service managers = " + eventManagerServices.keySet());
     }
 
+    @SuppressWarnings("unchecked")
+    public static <E extends Event> E eventOf(String eventType, Class<E> eventClass) {
+        return (E) eventBuilder()
+                .setType(eventType)
+                .build();
+    }
+
+    public static Event eventOf(String eventType,
+                                String code,
+                                String message,
+                                Path path) {
+        return eventBuilder()
+                .setType(eventType)
+                .setCode(code)
+                .setMessage(message)
+                .setPath(path)
+                .setPayload(PayloadImpl.fromPath(path))
+                .build();
+    }
+
+    public static Event eventOf(String eventType,
+                                Instant instant) {
+        return eventBuilder()
+                .setType(eventType)
+                .setInstant(instant)
+                .build();
+    }
+
+    public static Event eventFromFile(Path file) throws IOException {
+        return eventFromJson(Files.readString(file));
+    }
+
+    @SuppressWarnings("unchecked")
+    public static Event eventFromJson(String json) {
+        Map<String, Object> map = Json.toMap(json);
+        EventBuilder builder = eventBuilder();
+        if (map.containsKey("type")) {
+            builder.setType(map.getOrDefault("type", "generic").toString());
+        }
+        if (map.containsKey("code")) {
+            builder.setCode(map.getOrDefault("code", "").toString());
+        }
+        if (map.containsKey("message")) {
+            builder.setMessage(map.getOrDefault("message", "").toString());
+        }
+        if (map.containsKey("instant")) {
+            String instant = map.getOrDefault("instant", "").toString();
+            builder.setInstant(Instant.parse(instant));
+        }
+        if (map.containsKey("payload")) {
+            PayloadImpl payload = new PayloadImpl((Map<String, Object>) map.get("payload"));
+            builder.setPayload(payload);
+        }
+        if (map.containsKey("path")) {
+            Path path = Paths.get((String) map.get("path"));
+            builder.setPath(path);
+        }
+        return builder.build();
+    }
+
+    public EventManagerService init(EventManager eventManager) {
+        return this;
+    }
+
     public static EventManagerBuilder builder() {
         return new EventManagerBuilder();
     }
@@ -94,19 +164,11 @@ public final class EventManager {
         return builder.executorService;
     }
 
-    public void dispatch(Event event) {
-        getGenericEventManagerService().post(event);
-    }
-
     @SuppressWarnings("unchecked")
     public <T extends EventManagerService> T getEventManagerService(Class<T> cl) {
         return (T) eventManagerServices.get(cl);
     }
 
-    public GenericEventManagerService getGenericEventManagerService() {
-        return (GenericEventManagerService) eventManagerServices.get(GenericEventManagerService.class);
-    }
-
     public ClockEventManagerService getClockEventManagerService() {
         return (ClockEventManagerService) eventManagerServices.get(ClockEventManagerService.class);
     }
@@ -123,7 +185,24 @@ public final class EventManager {
         return (PathEventManagerService) eventManagerServices.get(PathEventManagerService.class);
     }
 
-    public void close() throws IOException {
+    public void publish(Event event) {
+        getEventBus().post(event);
+    }
+
+    public void publish(GenericEventImpl event,
+                        CompletableFuture<Event> future) {
+        SubscriberRegistry subscriberRegistry = getEventBus().getSubscribers();
+        Set<Subscriber> set = subscriberRegistry.getSubscribersForTesting(event.getClass());
+        event.setListener(new WrappedListener(event.getListener(), set.size(), future));
+        publish(event);
+    }
+
+    public List<Entry<Boolean>> getCronScheduleEntries() {
+        ClockEventManagerService clockEventManagerService = getClockEventManagerService();
+        return clockEventManagerService != null ? clockEventManagerService.getCronScheduleEntries() : List.of();
+    }
+
+    public void shutdown() throws IOException {
         for (EventConsumer eventConsumer : builder.eventConsumers) {
             if (eventConsumer instanceof Closeable closeable) {
                 closeable.close();
@@ -131,7 +210,9 @@ public final class EventManager {
         }
         for (EventManagerService service : eventManagerServices.values()) {
             try {
-                service.shutdown();
+                if (service != this) {
+                    service.shutdown();
+                }
             } catch (IOException e) {
                 logger.log(Level.SEVERE, e.getMessage(), e);
             }
@@ -244,36 +325,31 @@ public final class EventManager {
         }
     }
 
-    public static Event eventOf(String eventType,
-                                String code,
-                                String message,
-                                Path path) {
-        return eventBuilder()
-                .setType(eventType)
-                .setCode(code)
-                .setMessage(message)
-                .setPath(path)
-                .build();
-    }
+    private static class WrappedListener implements Listener {
 
-    public static Event eventOf(String eventType,
-                                Instant scheduled) {
-        return eventBuilder()
-                .setType(eventType)
-                .setScheduledFor(scheduled)
-                .build();
-    }
+        private final Listener listener;
 
-    public static Event eventFromFile(Path file) throws IOException {
-        return eventFromJson(Files.readString(file));
-    }
+        private int size;
 
-    public static Event eventFromJson(String json) {
-        Map<String, Object> map = Json.toMap(json);
-        return eventBuilder()
-                .setType(map.getOrDefault("type", "generic").toString())
-                .setPayload(new Payload(map))
-                .build();
+        private final CompletableFuture<Event> future;
+
+        public WrappedListener(Listener listener, int size, CompletableFuture<Event> future) {
+            this.listener = listener;
+            this.size = size;
+            this.future = future;
+        }
+
+        @Override
+        public void listen(Event event) {
+            if (listener != null) {
+                listener.listen(event);
+            } else {
+                logger.log(Level.WARNING, "listener not set");
+            }
+            if (--size == 0) {
+                future.complete(event);
+            }
+        }
     }
 
     public static class EventImpl implements Event {
@@ -304,18 +380,13 @@ public final class EventManager {
         }
 
         @Override
-        public Payload getPayload() {
+        public PayloadImpl getPayload() {
             return builder.payload;
         }
 
         @Override
-        public Instant getCreated() {
-            return builder.created;
-        }
-
-        @Override
-        public Instant getScheduledFor() {
-            return builder.scheduled;
+        public Instant getInstant() {
+            return builder.instant;
         }
 
         @Override
@@ -338,8 +409,24 @@ public final class EventManager {
             return builder.fileSize;
         }
 
+        @Override
         public String toJson() throws IOException {
-            return Json.toString(builder.payload);
+            JsonBuilder builder = JsonBuilder.builder();
+            builder.beginMap();
+            builder.fieldIfNotNull("type", getType());
+            builder.fieldIfNotNull("code", getCode());
+            builder.fieldIfNotNull("message", getMessage());
+            if (getPayload() != null && !getPayload().isEmpty()) {
+                builder.buildKey("payload").buildMap(getPayload());
+            }
+            if (getInstant() != null) {
+                builder.fieldIfNotNull("instant", getInstant().toString());
+            }
+            if (getPath() != null) {
+                builder.fieldIfNotNull("path", getPath().toAbsolutePath().toString());
+            }
+            builder.endMap();
+            return builder.build();
         }
 
         @Override
@@ -395,11 +482,9 @@ public final class EventManager {
 
         String message;
 
-        Instant scheduled;
+        Instant instant;
 
-        Instant created;
-
-        Payload payload;
+        PayloadImpl payload;
 
         Path path;
 
@@ -439,12 +524,12 @@ public final class EventManager {
             return this;
         }
 
-        public EventBuilder setScheduledFor(Instant scheduled) {
-            this.scheduled = scheduled;
+        public EventBuilder setInstant(Instant instant) {
+            this.instant = instant;
             return this;
         }
 
-        public EventBuilder setPayload(Payload payload) {
+        public EventBuilder setPayload(PayloadImpl payload) {
             this.payload = payload;
             return this;
         }
@@ -482,7 +567,7 @@ public final class EventManager {
                 cl = NullEvent.class;
             }
             try {
-                this.created = Instant.now();
+                this.instant = Instant.now();
                 if (listener != null) {
                     return cl.getDeclaredConstructor(EventBuilder.class, Listener.class).newInstance(this, listener);
                 } else {
@@ -507,5 +592,4 @@ public final class EventManager {
             return pos >= 0 ? name.substring(pos + 1) : null;
         }
     }
-
 }

event-common/src/main/java/org/xbib/event/common/GenericEventImpl.java

@@ -1,28 +1,27 @@
 package org.xbib.event.common;
 
+import org.xbib.event.GenericEvent;
 import org.xbib.event.Listener;
 
 import java.util.Objects;
 
-public class GenericEventImpl extends EventManager.EventImpl {
+public class GenericEventImpl extends EventManager.EventImpl implements GenericEvent {
 
     private final EventManager.EventBuilder builder;
 
-    public GenericEventImpl(EventManager.EventBuilder builder) {
-        this(builder, null);
-    }
-
     public GenericEventImpl(EventManager.EventBuilder builder, Listener listener) {
         super(builder);
         this.builder = builder;
-        this.builder.listener = Objects.requireNonNull(listener);
+        this.builder.listener = listener;
     }
 
+    @Override
     public GenericEventImpl setListener(Listener listener) {
         this.builder.listener = Objects.requireNonNull(listener);
         return this;
     }
 
+    @Override
     public void received() {
         builder.listener.listen(this);
     }

event-common/src/main/java/org/xbib/event/common/PayloadImpl.java

@@ -0,0 +1,35 @@
+package org.xbib.event.common;
+
+import org.xbib.datastructures.json.tiny.Json;
+import org.xbib.event.Payload;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+@SuppressWarnings("serial")
+public class PayloadImpl extends LinkedHashMap<String, Object> implements Payload {
+
+    public PayloadImpl() {
+        super();
+    }
+
+    public PayloadImpl(Map<String, Object> map) {
+        super(map);
+    }
+
+    public static PayloadImpl fromPath(Path path) {
+        PayloadImpl payload = new PayloadImpl();
+        try (InputStream inputStream = Files.newInputStream(path)) {
+            String content = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
+            payload.putAll(Json.toMap(content));
+        } catch (IOException e) {
+            throw new IllegalArgumentException("broken json content in path " + path);
+        }
+        return payload;
+    }
+}

event-common/src/main/java/org/xbib/event/generic/GenericEventManagerService.java

@@ -1,77 +0,0 @@
-package org.xbib.event.generic;
-
-import java.io.IOException;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-import org.xbib.event.Event;
-import org.xbib.event.Listener;
-import org.xbib.event.bus.EventBus;
-import org.xbib.event.bus.Subscriber;
-import org.xbib.event.bus.SubscriberRegistry;
-import org.xbib.event.common.EventManager;
-import org.xbib.event.common.EventManagerService;
-import org.xbib.event.common.GenericEventImpl;
-
-public class GenericEventManagerService implements EventManagerService {
-
-    private static final Logger logger = Logger.getLogger(GenericEventManagerService.class.getName());
-
-    private EventBus eventBus;
-
-    public GenericEventManagerService() {
-    }
-
-    @Override
-    public GenericEventManagerService init(EventManager eventManager) {
-        this.eventBus = eventManager.getEventBus();
-        return this;
-    }
-
-    @Override
-    public void shutdown() throws IOException {
-
-    }
-
-    public void post(Object event) {
-        eventBus.post(event);
-    }
-
-    public void post(GenericEventImpl event,
-                     CompletableFuture<Event> future) {
-        SubscriberRegistry subscriberRegistry = eventBus.getSubscribers();
-        Set<Subscriber> set = subscriberRegistry.getSubscribersForTesting(event.getClass());
-        logger.log(Level.INFO, "set = " + set);
-        event.setListener(new WrappedListener(event.getListener(), set.size(), future));
-        post(event);
-    }
-
-    static class WrappedListener implements Listener {
-
-        private final Listener listener;
-
-        private int size;
-
-        private final CompletableFuture<Event> future;
-
-        public WrappedListener(Listener listener, int size, CompletableFuture<Event> future) {
-            this.listener = listener;
-            this.size = size;
-            this.future = future;
-        }
-
-        @Override
-        public void listen(Event event) {
-            if (listener != null) {
-                listener.listen(event);
-            } else {
-                logger.log(Level.WARNING, "listener not set");
-            }
-            if (--size == 0) {
-                future.complete(event);
-            }
-        }
-    }
-}

event-common/src/main/java/org/xbib/event/path/FileFollowEventManagerService.java

@@ -1,6 +1,7 @@
 package org.xbib.event.path;
 
-import org.xbib.event.bus.EventBus;
+import org.xbib.event.Event;
+import org.xbib.event.common.AbstractEventManagerService;
 import org.xbib.event.common.EventManager;
 import org.xbib.event.common.EventManagerService;
 import org.xbib.settings.Settings;
@@ -17,24 +18,28 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.regex.Pattern;
 
-public class FileFollowEventManagerService implements EventManagerService {
+public class FileFollowEventManagerService extends AbstractEventManagerService implements EventManagerService {
 
     private static final Logger logger = Logger.getLogger(FileFollowEventManagerService.class.getName());
 
-    private Map<Future<?>, FileFollowEventService> eventServiceMap;
+    private EventManager eventManager;
+
+    private final Map<Future<?>, FileFollowEventService> services;
 
     public FileFollowEventManagerService() {
+        super();
+        this.services = new LinkedHashMap<>();
     }
 
     @Override
     public FileFollowEventManagerService init(EventManager eventManager) {
-        Settings settings = eventManager.getSettings();
-        EventBus eventBus = eventManager.getEventBus();
+        this.eventManager = eventManager;
         ExecutorService executorService = eventManager.getExecutorService();
-        this.eventServiceMap = new LinkedHashMap<>();
-        for (Map.Entry<String, Settings> entry : settings.getGroups("event.filefollow").entrySet()) {
+        for (Map.Entry<String, Settings> entry : eventManager.getSettings().getGroups("event.filefollow").entrySet()) {
+            String name = entry.getKey();
             Settings definition = entry.getValue();
             if (definition.getAsBoolean("enabled", true)) {
+                String type = definition.get("type", "filefollow");
                 String baseStr = definition.get("base");
                 Objects.requireNonNull(baseStr);
                 String patternStr = definition.get("pattern");
@@ -42,21 +47,25 @@ public class FileFollowEventManagerService implements EventManagerService {
                 try {
                     Path base = Paths.get(baseStr);
                     Pattern pattern = Pattern.compile(patternStr);
-                    FileFollowEventService fileFollowEventService = new FileFollowEventService(definition, eventBus, base, pattern);
+                    FileFollowEventService fileFollowEventService = new FileFollowEventService(this,
+                            type, name, base, pattern);
                     Future<?> future = executorService.submit(fileFollowEventService);
-                    eventServiceMap.put(future, fileFollowEventService);
-                    logger.log(Level.INFO, "file follow service " + entry.getKey() + " with base " + base + " and pattern " + pattern + " added");
+                    services.put(future, fileFollowEventService);
+                    logger.log(Level.INFO, "file follow service " + name +
+                            " with type " + type +
+                            " with base " + base + " and pattern " + pattern + " added");
                 } catch (Exception e) {
-                    logger.log(Level.SEVERE, "unable to create file follow service " + entry.getKey() + ", reason " + e.getMessage(), e);
+                    logger.log(Level.SEVERE, "unable to create file follow service " +name + ", reason " + e.getMessage(), e);
                 }
             }
         }
+        logger.log(Level.INFO, "after init: event services = " + services);
         return this;
     }
 
     @Override
     public void shutdown() throws IOException {
-        for (Map.Entry<Future<?>, FileFollowEventService> entry : eventServiceMap.entrySet()) {
+        for (Map.Entry<Future<?>, FileFollowEventService> entry : services.entrySet()) {
             try {
                 entry.getValue().setKeepWatching(false);
                 entry.getKey().cancel(true);
@@ -65,4 +74,8 @@ public class FileFollowEventManagerService implements EventManagerService {
             }
         }
     }
+
+    public void publish(Event event) {
+        eventManager.publish(event);
+    }
 }

event-common/src/main/java/org/xbib/event/path/FileFollowEventService.java

@@ -1,9 +1,7 @@
 package org.xbib.event.path;
 
 import org.xbib.event.Event;
-import org.xbib.event.bus.EventBus;
 import org.xbib.event.common.EventManager;
-import org.xbib.settings.Settings;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -33,7 +31,11 @@ public class FileFollowEventService implements Callable<Integer>, Closeable {
 
     private static final Logger logger = Logger.getLogger(FileFollowEventService.class.getName());
 
-    private final EventBus eventBus;
+    private final FileFollowEventManagerService fileFollowEventManagerService;
+
+    private final String eventType;
+
+    private final String name;
 
     private final Path base;
 
@@ -45,11 +47,14 @@ public class FileFollowEventService implements Callable<Integer>, Closeable {
 
     private volatile boolean keepWatching;
 
-    public FileFollowEventService(Settings settings,
-                                  EventBus eventBus,
+    public FileFollowEventService(FileFollowEventManagerService fileFollowEventManagerService,
+                                  String eventType,
+                                  String name,
                                   Path base,
                                   Pattern pattern) throws IOException {
-        this.eventBus = eventBus;
+        this.fileFollowEventManagerService = fileFollowEventManagerService;
+        this.eventType = eventType;
+        this.name = name;
         this.base = base;
         this.pattern = pattern;
         FileSystem fileSystem = base.getFileSystem();
@@ -58,7 +63,8 @@ public class FileFollowEventService implements Callable<Integer>, Closeable {
         kinds[0] = StandardWatchEventKinds.ENTRY_MODIFY;
         base.register(watchService, kinds);
         this.fileSizes = new LinkedHashMap<>();
-        fillFileSizes(base, pattern);
+        // initialize the sizes of existing files, so we can safely create first event
+        setFileSizes(base, pattern);
         this.keepWatching = true;
     }
 
@@ -87,15 +93,20 @@ public class FileFollowEventService implements Callable<Integer>, Closeable {
                         long currentSize = p.toFile().length();
                         fileSizes.put(p, currentSize);
                         String content = readRange(channel, lastSize, currentSize);
-                        // split content by line, this allows pattern matching without preprocessing in worker
+                        // split file content by line
+                        // this prevents swalloed events and allows pattern matching without preprocessing in worker
                         for (String line : content.split("\n")) {
-                            Event event = EventManager.eventBuilder()
-                                    .setType("filefollow")
-                                    .setCode(base.toString())
-                                    .setPath(path)
-                                    .setMessage(line)
-                                    .build();
-                            eventBus.post(event);
+                            if (fileFollowEventManagerService.getSuspended().contains(name)) {
+                                logger.log(Level.WARNING, name + " is suspended");
+                            } else {
+                                Event event = EventManager.eventBuilder()
+                                        .setType(eventType)
+                                        .setCode(base.toString())
+                                        .setPath(path)
+                                        .setMessage(line)
+                                        .build();
+                                fileFollowEventManagerService.publish(event);
+                            }
                         }
                     }
                 }
@@ -134,7 +145,7 @@ public class FileFollowEventService implements Callable<Integer>, Closeable {
         return charBuffer.toString();
     }
 
-    private void fillFileSizes(Path base, Pattern pattern) throws IOException {
+    private void setFileSizes(Path base, Pattern pattern) throws IOException {
         if (!Files.exists(base)) {
             return;
         }

event-common/src/main/java/org/xbib/event/path/PathEventManagerService.java

@@ -3,7 +3,7 @@ package org.xbib.event.path;
 import org.xbib.datastructures.api.TimeValue;
 import org.xbib.datastructures.json.tiny.Json;
 import org.xbib.event.Event;
-import org.xbib.event.bus.EventBus;
+import org.xbib.event.common.AbstractEventManagerService;
 import org.xbib.event.common.EventManager;
 import org.xbib.event.common.EventManagerService;
 import org.xbib.settings.Settings;
@@ -14,73 +14,102 @@ import java.io.Writer;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.ArrayList;
+import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Stream;
 
-public class PathEventManagerService implements EventManagerService {
+public class PathEventManagerService extends AbstractEventManagerService implements EventManagerService {
 
     private static final Logger logger = Logger.getLogger(PathEventManagerService.class.getName());
 
-    private EventBus eventBus;
+    private EventManager eventManager;
 
     private ExecutorService executorService;
 
-    private Map<Future<?>, PathEventService> eventServiceMap;
-
-    private List<String> suspendedQueues;
+    private final Map<Future<?>, PathEventService> services;
 
     public PathEventManagerService() {
+        super();
+        this.services = new LinkedHashMap<>();
     }
 
     @Override
     public PathEventManagerService init(EventManager eventManager) {
-        Settings settings = eventManager.getSettings();
-        this.eventBus = eventManager.getEventBus();
+        this.eventManager = eventManager;
         this.executorService = eventManager.getExecutorService();
-        this.eventServiceMap = new LinkedHashMap<>();
-        this.suspendedQueues = new ArrayList<>();
-        for (Map.Entry<String, Settings> entry : settings.getGroups("event.path").entrySet()) {
+        for (Map.Entry<String, Settings> entry : eventManager.getSettings().getGroups("event.path").entrySet()) {
             try {
                 String name = entry.getKey();
                 Settings definition = entry.getValue();
                 if (definition.getAsBoolean("enabled", true)) {
-                    Path path = Paths.get(definition.get("path", "/var/tmp/" + name));
+                    String pathName = Objects.requireNonNull(definition.get("path"), "path must not be null");
+                    Path path = Paths.get(pathName);
                     TimeValue lifetime = definition.getAsTime("lifetime", TimeValue.timeValueHours(72));
                     String eventType = definition.get("type", "path");
-                    createPathEventService(name, path, eventType, lifetime);
+                    createQueue(name, path);
+                    PathEventService pathEventService = new PathEventService(this, name, path, eventType, lifetime);
+                    add(pathEventService);
+                    pathEventService.drainIncoming();
                 } else {
-                    logger.log(Level.WARNING, "path servive definition not enabled in configuration");
+                    logger.log(Level.WARNING, "path service definition not enabled in configuration");
                 }
-            } catch (Exception e) {
+            } catch (IOException e) {
                 logger.log(Level.SEVERE, "unable to create event path service " + entry.getKey() + ", reason " + e.getMessage(), e);
             }
         }
+        logger.log(Level.INFO, "after init: event services = " + services);
         return this;
     }
 
-    public void createPathEventService(String name, Path path, String eventType, TimeValue lifetime) throws IOException {
+    public void add(String name, Path path, String eventType) throws IOException{
         createQueue(name, path);
-        PathEventService pathEventService = new PathEventService(this, eventBus, name, path, eventType, lifetime);
-        add(pathEventService);
+        add(new PathEventService(this, name, path, eventType, TimeValue.timeValueHours(72)));
+    }
+
+    public void add(String name, Path path, String eventType, TimeValue lifetime) throws IOException{
+        createQueue(name, path);
+        add(new PathEventService(this, name, path, eventType, lifetime));
     }
 
     public void add(PathEventService pathEventService) {
         Future<?> future = executorService.submit(pathEventService);
-        eventServiceMap.put(future, pathEventService);
-        logger.log(Level.INFO, "path event service " + pathEventService + " added");
+        services.put(future, pathEventService);
+        logger.log(Level.INFO, "path event service " + pathEventService + " added for path " + pathEventService.getPath());
+    }
+
+    public Collection<PathEventService> getPathEventServices() {
+        return services.values();
+    }
+
+    public PathEventService getPathEventService(String name) {
+        return services.values().stream().filter(p -> p.getName().equals(name)).findFirst().orElse(null);
+    }
+
+    public void publish(String eventType, Path path) {
+        Event event = EventManager.eventOf(eventType, null, null, path);
+        eventManager.publish(event);
+    }
+
+    public void failEvent(String eventType, Path path) {
+        try {
+            Event event = EventManager.eventOf(eventType, null, null, path);
+            event.fail();
+        } catch (IOException e) {
+            logger.log(Level.SEVERE, "unable to fail event because of  " + e.getMessage(), e);
+        }
     }
 
     @Override
     public void shutdown() throws IOException {
-        logger.log(Level.INFO, "shut down all path event services");
-        eventServiceMap.forEach((k, v) -> {
+        logger.log(Level.INFO, "shutting down all path event services");
+        services.forEach((k, v) -> {
             k.cancel(true);
             try {
                 v.close();
@@ -92,7 +121,7 @@ public class PathEventManagerService implements EventManagerService {
 
     public void destroy() {
         logger.log(Level.INFO, "shutting down and destroying all path event files");
-        eventServiceMap.forEach((k, v) -> {
+        services.forEach((k, v) -> {
             k.cancel(true);
             try {
                 v.close();
@@ -103,31 +132,28 @@ public class PathEventManagerService implements EventManagerService {
         });
     }
 
-    public List<String> getSuspendedQueues() {
-        return suspendedQueues;
-    }
-
-    public void suspend(String queue) {
-        suspendedQueues.add(queue);
-    }
-
-    public void resume(String queue) {
-        suspendedQueues.remove(queue);
-    }
-
-    public boolean put(Path path, String key, Map<String,Object> map) throws IOException {
-         return put(path, key, ".json", Json.toString(map));
-    }
-
-    public boolean putIfNotExists(Path path, String key, Map<String,Object> map) throws IOException {
+    public boolean publishJsonIfNotExists(PathEventService pathEventService,
+                                          String key,
+                                          Map<String,Object> map) throws IOException {
+        Path path = pathEventService.getPath();
         if (!exists(path, key, ".json")) {
-            return put(path, key, ".json", Json.toString(map));
+            return publishJson(pathEventService, key, map);
         } else {
             return false;
         }
     }
 
-    public boolean put(Path path, String key, String suffix, String string) throws IOException {
+    public boolean publishJson(PathEventService pathEventService,
+                               String key,
+                               Map<String,Object> map) throws IOException {
+        return publishJson(pathEventService, key, ".json", Json.toString(map));
+    }
+
+    public boolean publishJson(PathEventService pathEventService,
+                               String key,
+                               String suffix,
+                               String payload) throws IOException {
+        Path path = pathEventService.getPath();
         String keyFileName = key + suffix;
         if (Files.exists(path.resolve(Event.INCOMING).resolve(keyFileName)) ||
                 Files.exists(path.resolve(Event.SUCCESS).resolve(keyFileName))) {
@@ -136,10 +162,10 @@ public class PathEventManagerService implements EventManagerService {
         }
         Path p = path.resolve(Event.INCOMING).resolve(keyFileName);
         try (Writer writer = Files.newBufferedWriter(p)) {
-            writer.write(string);
+            writer.write(payload);
         }
         // obligatory purge. This is hacky.
-        eventServiceMap.forEach((k, v) -> {
+        services.forEach((k, v) -> {
             try {
                 v.purge();
             } catch (IOException e) {
@@ -149,12 +175,6 @@ public class PathEventManagerService implements EventManagerService {
         return true;
     }
 
-    public boolean exists(Path path, String key, String suffix) {
-        String keyFileName = key + suffix;
-        return Files.exists(path.resolve(Event.INCOMING).resolve(keyFileName)) ||
-                Files.exists(path.resolve(Event.SUCCESS).resolve(keyFileName));
-    }
-
     public long sizeOfIncoming(Path path) throws IOException {
         return sizeOf(path.resolve(Event.INCOMING));
     }
@@ -173,13 +193,19 @@ public class PathEventManagerService implements EventManagerService {
         }
     }
 
-    private static void createQueue(String name, Path p) throws IOException {
-        logger.log(Level.FINE, "creating queue " + name + " at " + p);
-        if (!Files.exists(p)) {
-            Files.createDirectories(p);
+    private static boolean exists(Path path, String key, String suffix) {
+        String keyFileName = key + suffix;
+        return Files.exists(path.resolve(Event.INCOMING).resolve(keyFileName)) ||
+                Files.exists(path.resolve(Event.SUCCESS).resolve(keyFileName));
+    }
+
+    private static void createQueue(String name, Path path) throws IOException {
+        logger.log(Level.FINE, "creating queue " + name + " at " + path);
+        if (!Files.exists(path)) {
+            Files.createDirectories(path);
         }
         for (String s: List.of(Event.INCOMING, Event.SUCCESS, Event.FAIL)) {
-            Path dir = p.resolve(s);
+            Path dir = path.resolve(s);
             if (!Files.exists(dir)) {
                 logger.log(Level.FINE, "creating queue " + name + " dir " + dir);
                 Files.createDirectories(dir);

event-common/src/main/java/org/xbib/event/path/PathEventService.java

@@ -4,8 +4,6 @@ import java.nio.file.FileVisitResult;
 import java.nio.file.SimpleFileVisitor;
 import org.xbib.datastructures.api.TimeValue;
 import org.xbib.event.Event;
-import org.xbib.event.bus.EventBus;
-import org.xbib.event.common.EventManager;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -21,18 +19,17 @@ import java.nio.file.WatchService;
 import java.nio.file.attribute.BasicFileAttributeView;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.time.Instant;
+import java.util.Objects;
 import java.util.concurrent.Callable;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-public class PathEventService implements Callable<Integer>, Closeable {
+public class PathEventService implements Callable<Boolean>, Closeable {
 
     private static final Logger logger = Logger.getLogger(PathEventService.class.getName());
 
     private final PathEventManagerService pathEventManager;
 
-    private final EventBus eventBus;
-
     private final Path path;
 
     private final String name;
@@ -43,20 +40,16 @@ public class PathEventService implements Callable<Integer>, Closeable {
 
     private final WatchService watchService;
 
-    private int eventCount;
-
     private volatile boolean keepWatching;
 
     public PathEventService(PathEventManagerService pathEventManager,
-                            EventBus eventBus,
                             String name,
                             Path path,
                             String eventType,
                             TimeValue lifetime) throws IOException {
         this.pathEventManager = pathEventManager;
-        this.eventBus = eventBus;
         this.name = name;
-        this.path = path;
+        this.path = Objects.requireNonNull(path, "path must not be null");
         this.eventType = eventType;
         this.lifetime = lifetime;
         this.watchService = path.getFileSystem().newWatchService();
@@ -67,9 +60,17 @@ public class PathEventService implements Callable<Integer>, Closeable {
         logger.log(Level.INFO, "path event service created for files at " + path);
     }
 
+    public String getName() {
+        return name;
+    }
+
+    public Path getPath() {
+        return path;
+    }
+
     @SuppressWarnings("unchecked")
     @Override
-    public Integer call() {
+    public Boolean call() {
         try {
             logger.log(Level.INFO, "watch service running on " + path.resolve(Event.INCOMING));
             while (keepWatching && watchService != null) {
@@ -86,10 +87,10 @@ public class PathEventService implements Callable<Integer>, Closeable {
                     String watchEventContext = pathWatchEvent.context().toString();
                     Path p = path.resolve(Event.INCOMING).resolve(watchEventContext);
                     logger.log(Level.FINE, "watch event " + pathWatchEvent + " context = " + watchEventContext + " path = " + p);
-                    if (pathEventManager.getSuspendedQueues().contains(name)) {
-                        failEvent(p);
+                    if (pathEventManager.getSuspended().contains(name)) {
+                        pathEventManager.failEvent(eventType, p);
                     } else {
-                        postEvent(p);
+                        pathEventManager.publish(eventType, p);
                     }
                 }
                 watchKey.reset();
@@ -102,7 +103,7 @@ public class PathEventService implements Callable<Integer>, Closeable {
         } catch (Exception e) {
             logger.log(Level.SEVERE, e.getMessage(), e);
         }
-        return eventCount;
+        return true;
     }
 
     @Override
@@ -120,7 +121,7 @@ public class PathEventService implements Callable<Integer>, Closeable {
 
     public void drainIncoming() throws IOException {
         try (DirectoryStream<Path> stream = Files.newDirectoryStream(path.resolve(Event.INCOMING))) {
-            stream.forEach(this::postEvent);
+            stream.forEach(p -> pathEventManager.publish(eventType, p));
         }
     }
 
@@ -149,21 +150,6 @@ public class PathEventService implements Callable<Integer>, Closeable {
         }
     }
 
-    private void postEvent(Path path) {
-        Event event = EventManager.eventOf(eventType, null, null, path);
-        eventBus.post(event);
-        eventCount++;
-    }
-
-    private void failEvent(Path file) {
-        try {
-            Event event = EventManager.eventFromFile(file);
-            event.fail();
-        } catch (IOException e) {
-            logger.log(Level.SEVERE, "unable to fail event because of  " + e.getMessage(), e);
-        }
-    }
-
     private static void delete(Path path) {
         if (path == null) {
             return;

event-common/src/main/java/org/xbib/event/persistence/Journal.java

@@ -1,6 +1,6 @@
 package org.xbib.event.persistence;
 
-import org.xbib.net.util.ExceptionFormatter;
+import org.xbib.event.util.ExceptionFormatter;
 
 import java.io.IOException;
 import java.io.OutputStream;

event-common/src/main/java/org/xbib/event/timer/TimerEventManagerService.java

@@ -1,7 +1,8 @@
 package org.xbib.event.timer;
 
-import org.xbib.event.Payload;
-import org.xbib.event.bus.EventBus;
+import org.xbib.event.Event;
+import org.xbib.event.common.PayloadImpl;
+import org.xbib.event.common.AbstractEventManagerService;
 import org.xbib.event.common.EventManager;
 import org.xbib.event.common.EventManagerService;
 import org.xbib.event.persistence.FilePersistenceStore;
@@ -20,38 +21,53 @@ import java.util.Map;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-public class TimerEventManagerService implements EventManagerService {
+public class TimerEventManagerService extends AbstractEventManagerService implements EventManagerService {
 
     private static final Logger logger = Logger.getLogger(TimerEventManagerService.class.getName());
 
-    private Map<String, TimerEventService> services;
+    private EventManager eventManager;
+
+    private final Map<String, TimerEventService> services;
 
     public TimerEventManagerService() {
+        super();
+        this.services = new LinkedHashMap<>();
     }
 
     @Override
     public TimerEventManagerService init(EventManager eventManager) {
-        Settings settings = eventManager.getSettings();
-        EventBus eventBus = eventManager.getEventBus();
-        this.services = new LinkedHashMap<>();
-        for (Map.Entry<String, Settings> entry : settings.getGroups("event.timer").entrySet()) {
+        this.eventManager = eventManager;
+        for (Map.Entry<String, Settings> entry : eventManager.getSettings().getGroups("event.timer").entrySet()) {
             String name = entry.getKey();
             Settings timerSettings = entry.getValue();
             try {
                 PersistenceStore<String, Object> persistenceStore = new FilePersistenceStore(timerSettings, name);
-                TimerEventService timerEventService = new TimerEventService(eventBus, name, ZoneId.systemDefault(), persistenceStore);
+                TimerEventService timerEventService = new TimerEventService(this, name, ZoneId.systemDefault(), persistenceStore);
                 services.put(name, timerEventService);
                 logger.log(Level.INFO, "timer " + name + " active: " + timerEventService);
             } catch (Exception e) {
                 logger.log(Level.WARNING, "unable to activate timer " + name + ", reason: " + e.getMessage(), e);
             }
         }
+        logger.log(Level.INFO, "after init: event services = " + services);
         return this;
     }
 
-    public boolean publish(String name,
-                           String timeSpec,
-                           Payload payload) throws ParseException, IOException {
+    @Override
+    public void shutdown() throws IOException {
+        for (Map.Entry<String, TimerEventService> entry : services.entrySet()) {
+            logger.log(Level.INFO, "closing timer " + entry.getKey());
+            entry.getValue().close();
+        }
+    }
+
+    public void publish(Event event) {
+        eventManager.publish(event);
+    }
+
+    public boolean schedule(String name,
+                            String timeSpec,
+                            PayloadImpl payload) throws ParseException, IOException {
         if (services.containsKey(name)) {
             Span span = Chronic.parse(timeSpec);
             if (span != null) {
@@ -69,9 +85,9 @@ public class TimerEventManagerService implements EventManagerService {
         return false;
     }
 
-    public boolean publish(String service,
-                           Instant instant,
-                           Payload payload) throws IOException {
+    public boolean schedule(String service,
+                            Instant instant,
+                            PayloadImpl payload) throws IOException {
         if (services.containsKey(service)) {
             services.get(service).schedule(instant, payload);
             return true;
@@ -87,12 +103,4 @@ public class TimerEventManagerService implements EventManagerService {
             entry.getValue().purge();
         }
     }
-
-    @Override
-    public void shutdown() throws IOException {
-        for (Map.Entry<String, TimerEventService> entry : services.entrySet()) {
-            logger.log(Level.INFO, "closing timer " + entry.getKey());
-            entry.getValue().close();
-        }
-    }
 }

event-common/src/main/java/org/xbib/event/timer/TimerEventService.java

@@ -1,8 +1,7 @@
 package org.xbib.event.timer;
 
 import org.xbib.event.Event;
-import org.xbib.event.Payload;
-import org.xbib.event.bus.EventBus;
+import org.xbib.event.common.PayloadImpl;
 import org.xbib.event.common.EventManager;
 import org.xbib.event.persistence.PersistenceStore;
 
@@ -25,8 +24,10 @@ import java.util.logging.Logger;
 class TimerEventService implements Closeable {
 
     private static final Logger logger = Logger.getLogger(TimerEventService.class.getName());
-    
-    private final EventBus eventBus;
+
+    private final TimerEventManagerService timerEventManagerService;
+
+    private final String name;
 
     private final ZoneId zoneId;
 
@@ -34,11 +35,12 @@ class TimerEventService implements Closeable {
 
     private final Timer timer;
 
-    public TimerEventService(EventBus eventBus,
+    public TimerEventService(TimerEventManagerService timerEventManagerService,
                              String name,
                              ZoneId zoneId,
                              PersistenceStore<String, Object> persistenceStore) throws IOException {
-        this.eventBus = eventBus;
+        this.timerEventManagerService = timerEventManagerService;
+        this.name = name;
         this.zoneId = zoneId;
         this.persistenceStore = persistenceStore;
         this.timer = new Timer();
@@ -46,7 +48,7 @@ class TimerEventService implements Closeable {
         logger.log(Level.INFO, "timer event service " + name + " loaded and rescheduled, " +  tasknum + " timer tasks");
     }
 
-    void schedule(Instant instant, Payload payload) throws IOException {
+    void schedule(Instant instant, PayloadImpl payload) throws IOException {
         String scheduled = instant.atZone(zoneId).format(DateTimeFormatter.ISO_DATE_TIME);
         payload.put("scheduled", scheduled);
         TimerEventTask timerEventTask = new TimerEventTask(payload);
@@ -62,7 +64,7 @@ class TimerEventService implements Closeable {
         persistenceStore.clear();
         persistenceStore.commit();
         for (Map<String, Object> task : tasks) {
-            Payload payload = new Payload(task);
+            PayloadImpl payload = new PayloadImpl(task);
             ZonedDateTime scheduledDate = ZonedDateTime.parse((String) task.get("scheduled"), DateTimeFormatter.ISO_DATE_TIME);
             if (scheduledDate.isBefore(ZonedDateTime.now())) {
                 logger.log(Level.WARNING, "scheduled timer task " + task + " date already passed");
@@ -87,9 +89,9 @@ class TimerEventService implements Closeable {
 
     public class TimerEventTask extends TimerTask {
 
-        private final Payload payload;
+        private final PayloadImpl payload;
 
-        public TimerEventTask(Payload payload) throws IOException {
+        public TimerEventTask(PayloadImpl payload) throws IOException {
             this.payload = payload;
             persistenceStore.insert("tasks", this.payload);
         }
@@ -97,12 +99,16 @@ class TimerEventService implements Closeable {
         @Override
         public void run() {
             try {
+                if (timerEventManagerService.getSuspended().contains(name)) {
+                    logger.log(Level.FINE, "timer event " + name + " suspended");
+                    return;
+                }
                 Event timerEvent = EventManager.eventBuilder()
                         .setType("timer")
+                        .setCode(name)
                         .setPayload(payload)
                         .build();
-                logger.log(Level.FINE, "posting timer event " + timerEvent.getClass().getName() + " with payload = " + payload);
-                eventBus.post(timerEvent);
+                timerEventManagerService.publish(timerEvent);
                 logger.log(Level.FINE, "persistence before remove: " + persistenceStore.get("tasks"));
                 if (persistenceStore.remove("tasks", this.payload)) {
                     logger.log(Level.FINE, "removal done");

event-common/src/main/java/org/xbib/event/util/ExceptionFormatter.java

@@ -0,0 +1,56 @@
+package org.xbib.event.util;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+/**
+ * Format exception messages and stack traces.
+ */
+public final class ExceptionFormatter {
+
+    private ExceptionFormatter() {
+    }
+
+    /**
+     * Format exception with stack trace.
+     *
+     * @param t the thrown object
+     * @return the formatted exception
+     */
+    public static String format(Throwable t) {
+        StringBuilder sb = new StringBuilder();
+        append(sb, t, 0, true);
+        return sb.toString();
+    }
+
+    /**
+     * Append Exception to string builder.
+     */
+    private static void append(StringBuilder sb, Throwable t, int level, boolean details) {
+        if (((t != null) && (t.getMessage() != null)) && (!t.getMessage().isEmpty())) {
+            if (details && (level > 0)) {
+                sb.append("\n\nCaused by\n");
+            }
+            sb.append(t.getMessage());
+        }
+        if (details) {
+            if (t != null) {
+                if ((t.getMessage() != null) && (t.getMessage().isEmpty())) {
+                    sb.append("\n\nCaused by ");
+                } else {
+                    sb.append("\n\n");
+                }
+            }
+            StringWriter sw = new StringWriter();
+            if (t != null) {
+                t.printStackTrace(new PrintWriter(sw));
+            }
+            sb.append(sw.toString());
+        }
+        if (t != null) {
+            if (t.getCause() != null) {
+                append(sb, t.getCause(), level + 1, details);
+            }
+        }
+    }
+}

event-common/src/test/java/org/xbib/event/clock/ClockEventManagerTest.java

@@ -13,13 +13,15 @@ public class ClockEventManagerTest {
         TestClockEventConsumer clockEventConsumer = new TestClockEventConsumer();
         Settings settings = Settings.settingsBuilder()
                 .put("event.clock.testclockevent.enabled", "true")
+                .put("event.clock.testclockevent.type", "test-clock-event")
                 .put("event.clock.testclockevent.entry", "*/1 6-21 * * *")
                 .build();
         EventManager eventManager = EventManager.builder()
                 .setSettings(settings)
+                .register("test-clock-event", TestClockEvent.class)
                 .register(clockEventConsumer)
                 .build();
         Thread.sleep(90000L);
-        eventManager.close();
+        eventManager.shutdown();
     }
 }

event-common/src/test/java/org/xbib/event/clock/TestClockEvent.java

@@ -0,0 +1,11 @@
+package org.xbib.event.clock;
+
+import org.xbib.event.common.ClockEventImpl;
+import org.xbib.event.common.EventManager;
+
+public class TestClockEvent extends ClockEventImpl {
+
+    public TestClockEvent(EventManager.EventBuilder builder) {
+        super(builder);
+    }
+}

event-common/src/test/java/org/xbib/event/clock/TestClockEventConsumer.java

@@ -1,6 +1,5 @@
 package org.xbib.event.clock;
 
-import org.xbib.event.ClockEvent;
 import org.xbib.event.EventConsumer;
 import org.xbib.event.bus.AllowConcurrentEvents;
 import org.xbib.event.bus.Subscribe;
@@ -15,7 +14,8 @@ public class TestClockEventConsumer implements EventConsumer {
 
     @Subscribe
     @AllowConcurrentEvents
-    void onEvent(ClockEvent event) {
-        logger.log(Level.INFO, "received clock event on  " + Instant.now() + " event instant = " + event.getCreated());
+    void onEvent(TestClockEvent event) {
+        logger.log(Level.INFO, "received test clock event on " + Instant.now() +
+                " event instant = " + event.getInstant());
     }
 }

event-common/src/test/java/org/xbib/event/common/GenericEventManagerTest.java

@@ -1,11 +1,9 @@
-package org.xbib.event.generic;
+package org.xbib.event.common;
 
 import org.junit.jupiter.api.Test;
 import org.xbib.event.Event;
 import org.xbib.event.EventConsumer;
 import org.xbib.event.bus.Subscribe;
-import org.xbib.event.common.EventManager;
-import org.xbib.event.common.GenericEventImpl;
 import org.xbib.settings.Settings;
 
 import java.util.concurrent.CompletableFuture;
@@ -32,7 +30,7 @@ public class GenericEventManagerTest {
                 .setType("generic")
                 .setListener(e -> logger.log(Level.INFO, "received event " + e))
                 .build();
-        eventManager.getGenericEventManagerService().post(event);
+        eventManager.publish(event);
         // we must wait for a certain time because we do not use a future
         Thread.sleep(500L);
         assertEquals(1, consumer.getCount());
@@ -52,7 +50,7 @@ public class GenericEventManagerTest {
                     future.complete(e);
                 })
                 .build();
-        eventManager.getGenericEventManagerService().post(event);
+        eventManager.publish(event);
         Event e = future.get(1000L, TimeUnit.MILLISECONDS);
         assertNotNull(e);
     }
@@ -76,7 +74,7 @@ public class GenericEventManagerTest {
                     future.complete(e);
                 })
                 .build();
-        eventManager.getGenericEventManagerService().post((GenericEventImpl) event, future);
+        eventManager.publish((GenericEventImpl) event, future);
         Event e = future.get(5L, TimeUnit.SECONDS);
         if (e != null) {
             logger.log(Level.INFO, "the event " + e + " was received by all consumers");

event-common/src/test/java/org/xbib/event/path/FileFollowEventManagerTest.java

@@ -21,11 +21,13 @@ public class FileFollowEventManagerTest {
         TestFileFollowEventConsumer consumer = new TestFileFollowEventConsumer();
         Settings settings = Settings.settingsBuilder()
                 .put("event.filefollow.testfilefollowevent.enabled", "true")
+                .put("event.filefollow.testfilefollowevent.type", "filefollow-test")
                 .put("event.filefollow.testfilefollowevent.base", path.toString())
                 .put("event.filefollow.testfilefollowevent.pattern", ".*")
                 .build();
         EventManager eventManager = EventManager.builder()
                 .setSettings(settings)
+                .register("filefollow-test", TestFileFollowEvent.class)
                 .register(consumer)
                 .build();
         Thread.sleep(1000L);
@@ -37,6 +39,6 @@ public class FileFollowEventManagerTest {
         Thread.sleep(1000L);
         Files.delete(testTxt);
         Files.delete(path);
-        eventManager.close();
+        eventManager.shutdown();
     }
 }

event-common/src/test/java/org/xbib/event/path/PathEventManagerTest.java

@@ -1,7 +1,11 @@
 package org.xbib.event.path;
 
 import org.junit.jupiter.api.Test;
+import org.xbib.datastructures.api.TimeValue;
+import org.xbib.event.EventConsumer;
 import org.xbib.event.PathEvent;
+import org.xbib.event.bus.AllowConcurrentEvents;
+import org.xbib.event.bus.Subscribe;
 import org.xbib.event.common.EventManager;
 import org.xbib.event.common.PathEventImpl;
 import org.xbib.settings.Settings;
@@ -10,6 +14,7 @@ import java.io.BufferedWriter;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Map;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -31,10 +36,11 @@ public class PathEventManagerTest {
                 .build();
         Path testTxt = path.resolve("incoming").resolve("test.txt");
         try (BufferedWriter bufferedWriter = Files.newBufferedWriter(testTxt)) {
-            bufferedWriter.write("Hello");
+            bufferedWriter.write("{\"Hello\":\"world\"}");
         }
-        Thread.sleep(2000L);
-        eventManager.close();
+        Thread.sleep(1000L);
+        eventManager.shutdown();
+        // extra destroy to clean up test
         eventManager.getPathEventManagerService().destroy();
     }
 
@@ -44,8 +50,8 @@ public class PathEventManagerTest {
         TestPathEventConsumer consumer = new TestPathEventConsumer();
         Settings settings = Settings.settingsBuilder()
                 .put("event.path.testpathevent.enabled", "true")
-                .put("event.path.testpathevent.path", path.toString())
                 .put("event.path.testpathevent.type", "path-ext")
+                .put("event.path.testpathevent.path", path.toString())
                 .build();
         EventManager eventManager = EventManager.builder()
                 .setSettings(settings)
@@ -54,10 +60,32 @@ public class PathEventManagerTest {
                 .build();
         Path testTxt = path.resolve("incoming").resolve("test.txt");
         try (BufferedWriter bufferedWriter = Files.newBufferedWriter(testTxt)) {
-            bufferedWriter.write("Hello");
+            bufferedWriter.write("{\"Hello\":\"world\"}");
         }
-        Thread.sleep(2000L);
-        eventManager.close();
+        Thread.sleep(1000L);
+        eventManager.shutdown();
+        // extra destroy to clean up test
+        eventManager.getPathEventManagerService().destroy();
+    }
+
+    @Test
+    public void testPathEventByApi() throws Exception {
+        TestPathExtEventConsumer consumer = new TestPathExtEventConsumer();
+        EventManager eventManager = EventManager.builder()
+                .register("path-ext", PathExtEvent.class)
+                .register(consumer)
+                .build();
+        // create by API
+        Path path = Files.createTempDirectory("testpath");
+        eventManager.getPathEventManagerService().add("test", path, "path-ext", TimeValue.timeValueHours(72));
+        // publish
+        PathEventService pathEventService = eventManager.getPathEventManagerService().getPathEventService("test");
+        eventManager.getPathEventManagerService()
+                .publishJson(pathEventService, "key", Map.of("hello", "world"));
+        // everything done
+        Thread.sleep(1000L);
+        eventManager.shutdown();
+        // extra destroy to clean up test
         eventManager.getPathEventManagerService().destroy();
     }
 
@@ -68,4 +96,25 @@ public class PathEventManagerTest {
             logger.log(Level.INFO, "I'm the path ext event");
         }
     }
+
+    public static class TestPathEventConsumer implements EventConsumer {
+
+        @Subscribe
+        @AllowConcurrentEvents
+        void onEvent(PathEvent event) {
+            logger.log(Level.INFO, "received path event, path = " + event.getPath() +
+                    " payload = " + event.getPayload());
+        }
+    }
+
+
+    public static class TestPathExtEventConsumer implements EventConsumer {
+
+        @Subscribe
+        @AllowConcurrentEvents
+        void onEvent(PathExtEvent event) {
+            logger.log(Level.INFO, "received path ext event, path = " + event.getPath() +
+                    " payload = " + event.getPayload());
+        }
+    }
 }

event-common/src/test/java/org/xbib/event/path/TestFileFollowEvent.java

@@ -0,0 +1,10 @@
+package org.xbib.event.path;
+
+import org.xbib.event.common.EventManager;
+import org.xbib.event.common.FileFollowEventImpl;
+
+public class TestFileFollowEvent extends FileFollowEventImpl {
+    public TestFileFollowEvent(EventManager.EventBuilder builder) {
+        super(builder);
+    }
+}

event-common/src/test/java/org/xbib/event/path/TestFileFollowEventConsumer.java

@@ -1,7 +1,6 @@
 package org.xbib.event.path;
 
 import org.xbib.event.EventConsumer;
-import org.xbib.event.FileFollowEvent;
 import org.xbib.event.bus.AllowConcurrentEvents;
 import org.xbib.event.bus.Subscribe;
 
@@ -14,7 +13,7 @@ public class TestFileFollowEventConsumer implements EventConsumer {
 
     @Subscribe
     @AllowConcurrentEvents
-    void onEvent(FileFollowEvent event) {
-        logger.log(Level.INFO, "received filefollow event, path = " + event.getPath() + " content = " + event.getMessage());
+    void onEvent(TestFileFollowEvent event) {
+        logger.log(Level.INFO, "received test filefollow event, path = " + event.getPath() + " content = " + event.getMessage());
     }
 }

event-common/src/test/java/org/xbib/event/timer/TestTimerEventConsumer.java

@@ -16,7 +16,7 @@ public class TestTimerEventConsumer implements EventConsumer {
     @Subscribe
     @AllowConcurrentEvents
     void onEvent(TimerEvent event) {
-        logger.log(Level.INFO, "received timer event on  " + Instant.now() + " event instant = " + event.getCreated());
+        logger.log(Level.INFO, "received timer event on  " + Instant.now() + " event instant = " + event.getInstant());
     }
 
 }

event-common/src/test/java/org/xbib/event/timer/TimerEventManagerTest.java

@@ -1,7 +1,7 @@
 package org.xbib.event.timer;
 
 import org.junit.jupiter.api.Test;
-import org.xbib.event.Payload;
+import org.xbib.event.common.PayloadImpl;
 import org.xbib.event.common.EventManager;
 import org.xbib.settings.Settings;
 
@@ -22,8 +22,8 @@ public class TimerEventManagerTest {
                 .register(consumer)
                 .build();
         TimerEventManagerService timerEventManager = eventManager.getTimerEventManagerService();
-        Payload payload = new Payload(Map.of("a", "b"));
-        timerEventManager.publish("testtimerevent", Instant.now().plusSeconds(5L), payload);
+        PayloadImpl payload = new PayloadImpl(Map.of("a", "b"));
+        timerEventManager.schedule("testtimerevent", Instant.now().plusSeconds(5L), payload);
         Thread.sleep(10000L);
         timerEventManager.shutdown();
     }

event-journald/build.gradle

@@ -0,0 +1,3 @@
+dependencies {
+    implementation libs.jna
+}

event-journald/src/main/java/org/xbib/event/journald/DefaultJournalEntry.java

@@ -0,0 +1,421 @@
+package org.xbib.event.journald;
+
+public class DefaultJournalEntry implements JournalEntry {
+
+    private String message;
+
+    private String messageId;
+
+    private int priority;
+
+    private String codeFile;
+
+    private String codeLine;
+
+    private String codeFunc;
+
+    private String errno;
+
+    private String syslogFacility;
+
+    private String syslogIdentifier;
+
+    private String syslogPid;
+
+    private String syslogTimestamp;
+
+    private String syslogRaw;
+
+    private String pid;
+
+    private String uid;
+
+    private String gid;
+
+    private String comm;
+
+    private String exe;
+
+    private String cmdLine;
+
+    private String capEffective;
+
+    private String auditSession;
+
+    private String auditLoginUid;
+
+    private String systemdCgroup;
+
+    private String systemdSlice;
+
+    private String systemdUnit;
+
+    private String systemdUserSlice;
+
+    private String systemdUserUnit;
+
+    private String systemdSession;
+
+    private String systemdOwnerUid;
+
+    private String selinuxContext;
+
+    private String sourceRealtimeTimestamp;
+
+    private String bootId;
+
+    private String machineId;
+
+    private String systemdInvocationId;
+
+    private String hostname;
+
+    private String transport;
+
+    private String streamId;
+
+    private String lineBreak;
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    @Override
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessageId(String messageId) {
+        this.messageId = messageId;
+    }
+
+    @Override
+    public String getMessageId() {
+        return messageId;
+    }
+
+    public void setPriority(int priority) {
+        this.priority = priority;
+    }
+
+    @Override
+    public int getPriority() {
+        return priority;
+    }
+
+    public void setCodeFile(String codeFile) {
+        this.codeFile = codeFile;
+    }
+
+    @Override
+    public String getCodeFile() {
+        return codeFile;
+    }
+
+    public void setCodeLine(String codeLine) {
+        this.codeLine = codeLine;
+    }
+
+    @Override
+    public String getCodeLine() {
+        return codeLine;
+    }
+
+    public void setCodeFunc(String codeFunc) {
+        this.codeFunc = codeFunc;
+    }
+
+    @Override
+    public String getCodeFunc() {
+        return codeFunc;
+    }
+
+    public void setErrno(String errno) {
+        this.errno = errno;
+    }
+
+    @Override
+    public String getErrno() {
+        return errno;
+    }
+
+    public void setSyslogFacility(String syslogFacility) {
+        this.syslogFacility = syslogFacility;
+    }
+
+    @Override
+    public String getSyslogFacility() {
+        return syslogFacility;
+    }
+
+    public void setSyslogIdentifier(String syslogIdentifier) {
+        this.syslogIdentifier = syslogIdentifier;
+    }
+
+    @Override
+    public String getSyslogIdentifier() {
+        return syslogIdentifier;
+    }
+
+    public void setSyslogPid(String syslogPid) {
+        this.syslogPid = syslogPid;
+    }
+
+    @Override
+    public String getSyslogPid() {
+        return syslogPid;
+    }
+
+    public void setSyslogTimestamp(String syslogTimestamp) {
+        this.syslogTimestamp = syslogTimestamp;
+    }
+
+    @Override
+    public String getSyslogTimestamp() {
+        return syslogTimestamp;
+    }
+
+    public void setSyslogRaw(String syslogRaw) {
+        this.syslogRaw = syslogRaw;
+    }
+
+    @Override
+    public String getSyslogRaw() {
+        return syslogRaw;
+    }
+
+    public void setPid(String pid) {
+        this.pid = pid;
+    }
+
+    @Override
+    public String getPid() {
+        return pid;
+    }
+
+    public void setUid(String uid) {
+        this.uid = uid;
+    }
+
+    @Override
+    public String getUid() {
+        return uid;
+    }
+
+    public void setGid(String gid) {
+        this.gid = gid;
+    }
+
+    @Override
+    public String getGid() {
+        return gid;
+    }
+
+    public void setComm(String comm) {
+        this.comm = comm;
+    }
+
+    @Override
+    public String getComm() {
+        return comm;
+    }
+
+    public void setExe(String exe) {
+        this.exe = exe;
+    }
+
+    @Override
+    public String getExe() {
+        return exe;
+    }
+
+    public void setCmdLine(String cmdline) {
+        this.cmdLine = cmdline;
+    }
+
+    @Override
+    public String getCmdLine() {
+        return cmdLine;
+    }
+
+    public void setCapEffective(String capEffective) {
+        this.capEffective = capEffective;
+    }
+
+    @Override
+    public String getCapEffective() {
+        return capEffective;
+    }
+
+    public void setAuditSession(String auditSession) {
+        this.auditSession = auditSession;
+    }
+
+    @Override
+    public String getAuditSession() {
+        return auditSession;
+    }
+
+    public void setAuditLoginUid(String auditLoginUid) {
+        this.auditLoginUid = auditLoginUid;
+    }
+
+    @Override
+    public String getAuditLoginUid() {
+        return auditLoginUid;
+    }
+
+    public void setSystemdCgroup(String systemdCgroup) {
+        this.systemdCgroup = systemdCgroup;
+    }
+
+    @Override
+    public String getSystemdCgroup() {
+        return systemdCgroup;
+    }
+
+    public void setSystemdSlice(String systemdSlice) {
+        this.systemdSlice = systemdSlice;
+    }
+
+    @Override
+    public String getSystemdSlice() {
+        return systemdSlice;
+    }
+
+    public void setSystemdUnit(String systemdUnit) {
+        this.systemdUnit = systemdUnit;
+    }
+
+    @Override
+    public String getSystemdUnit() {
+        return systemdUnit;
+    }
+
+    public void setSystemdUserSlice(String systemdUserSlice) {
+        this.systemdUserSlice = systemdUserSlice;
+    }
+
+    @Override
+    public String getSystemdUserSlice() {
+        return systemdUserSlice;
+    }
+
+    public void setSystemdUserUnit(String systemdUserUnit) {
+        this.systemdUserUnit = systemdUserUnit;
+    }
+
+    @Override
+    public String getSystemdUserUnit() {
+        return systemdUserUnit;
+    }
+
+    public void setSystemdSession(String systemdSession) {
+        this.systemdSession = systemdSession;
+    }
+
+    @Override
+    public String getSystemdSession() {
+        return systemdSession;
+    }
+
+    public void setSystemdOwnerUid(String systemdOwnerUid) {
+        this.systemdOwnerUid = systemdOwnerUid;
+    }
+
+    @Override
+    public String getSystemdOwnerUid() {
+        return systemdOwnerUid;
+    }
+
+    public void setSelinuxContext(String selinuxContext) {
+        this.selinuxContext = selinuxContext;
+    }
+
+    @Override
+    public String getSelinuxContext() {
+        return selinuxContext;
+    }
+
+    public void setSourceRealtimeTimestamp(String sourceRealtimeTimestamp) {
+        this.sourceRealtimeTimestamp = sourceRealtimeTimestamp;
+    }
+
+    @Override
+    public String getSourceRealtimeTimestamp() {
+        return sourceRealtimeTimestamp;
+    }
+
+    public void setBootId(String bootId) {
+        this.bootId = bootId;
+    }
+
+    @Override
+    public String getBootId() {
+        return bootId;
+    }
+
+    public void setMachineId(String machineId) {
+        this.machineId = machineId;
+    }
+
+    @Override
+    public String getMachineId() {
+        return machineId;
+    }
+
+    public void setSystemdInvocationId(String systemdInvocationId) {
+        this.systemdInvocationId = systemdInvocationId;
+    }
+
+    @Override
+    public String getSystemdInvocationId() {
+        return systemdInvocationId;
+    }
+
+    public void setHostname(String hostname) {
+        this.hostname = hostname;
+    }
+
+    @Override
+    public String getHostname() {
+        return hostname;
+    }
+
+    public void setTransport(String transport) {
+        this.transport = transport;
+    }
+
+    @Override
+    public String getTransport() {
+        return transport;
+    }
+
+    public void setStreamId(String streamId) {
+        this.streamId = streamId;
+    }
+
+    @Override
+    public String getStreamId() {
+        return streamId;
+    }
+
+    public void setLineBreak(String lineBreak) {
+        this.lineBreak = lineBreak;
+    }
+
+    @Override
+    public String getLineBreak() {
+        return lineBreak;
+    }
+
+    @Override
+    public String getFieldValue(String fieldName) {
+        return null;
+    }
+
+    @Override
+    public String toString() {
+        return "priority=" + this.priority + ",message=" + this.message;
+    }
+}

event-journald/src/main/java/org/xbib/event/journald/JournalEntry.java

@@ -0,0 +1,80 @@
+package org.xbib.event.journald;
+
+public interface JournalEntry {
+
+    String getMessage();
+
+    String getMessageId();
+
+    int getPriority();
+
+    String getCodeFile();
+
+    String getCodeLine();
+
+    String getCodeFunc();
+
+    String getErrno();
+
+    String getSyslogFacility();
+
+    String getSyslogIdentifier();
+
+    String getSyslogPid();
+
+    String getSyslogTimestamp();
+
+    String getSyslogRaw();
+
+    String getPid();
+
+    String getUid();
+
+    String getGid();
+
+    String getComm();
+
+    String getExe();
+
+    String getCmdLine();
+
+    String getCapEffective();
+
+    String getAuditSession();
+
+    String getAuditLoginUid();
+
+    String getSystemdCgroup();
+
+    String getSystemdSlice();
+
+    String getSystemdUnit();
+
+    String getSystemdUserSlice();
+
+    String getSystemdUserUnit();
+
+    String getSystemdSession();
+
+    String getSystemdOwnerUid();
+
+    String getSelinuxContext();
+
+    String getSourceRealtimeTimestamp();
+
+    String getBootId();
+
+    String getMachineId();
+
+    String getSystemdInvocationId();
+
+    String getHostname();
+
+    String getTransport();
+
+    String getStreamId();
+
+    String getLineBreak();
+
+    String getFieldValue(String fieldName);
+}

event-journald/src/main/java/org/xbib/event/journald/SystemdJournalConsumer.java

@@ -0,0 +1,240 @@
+package org.xbib.event.journald;
+
+import com.sun.jna.Memory;
+import com.sun.jna.Native;
+import com.sun.jna.Pointer;
+import com.sun.jna.StringArray;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import static org.xbib.event.journald.SystemdLibraryInstance.SD_JOURNAL_LOCAL_ONLY;
+import static org.xbib.event.journald.SystemdLibraryInstance.SD_JOURNAL_NOP;
+import static org.xbib.event.journald.SystemdLibraryInstance.SD_JOURNAL_SYSTEM;
+
+public class SystemdJournalConsumer implements Runnable {
+
+    private static final Logger logger = Logger.getLogger(SystemdJournalConsumer.class.getName());
+
+    private final String match;
+
+    private final String field;
+
+    private final SystemdJournalListener listener;
+
+    public SystemdJournalConsumer(String match, SystemdJournalListener listener) {
+        this(match, null, listener);
+    }
+
+    public SystemdJournalConsumer(String match, String field, SystemdJournalListener listener) {
+        this.match = match;
+        this.field = field;
+        this.listener = listener;
+    }
+
+    @Override
+    public void run() {
+        try {
+            loop();
+        } catch (Throwable e) {
+            logger.log(Level.SEVERE, e.getMessage(), e);
+        }
+    }
+
+    /**
+     * <a href="https://gist.github.com/portante/ff7fb429c6f973aab377f7bb77b0ffdb">...</a>
+     * @throws IOException if loop fails
+     */
+    private void loop() throws IOException {
+        SystemdLibraryInstance api = SystemdLibraryInstance.getInstance();
+        SystemdLibrary.SdJournal sdJournal = new SystemdLibrary.SdJournal();
+        int rc = api.sd_journal_open(sdJournal, SD_JOURNAL_LOCAL_ONLY);
+        if (rc < 0) {
+            throw new IOException("error opening journal for read: " + rc);
+        }
+        logger.log(Level.INFO, "top fd = " + sdJournal.top_level_fd +
+                " path = " + sdJournal.path +
+                " prefix = " + sdJournal.prefix +
+                " namespace = " + sdJournal.namespace);
+        if (match != null) {
+            rc = api.sd_journal_add_match(sdJournal, match, match.length());
+            logger.log(Level.INFO, "add_match: " + rc);
+            if (rc < 0) {
+                throw new IOException("error in add_match: " + rc);
+            }
+        }
+        rc = api.sd_journal_get_fd(sdJournal);
+        logger.log(Level.INFO, "get_fd: " + rc);
+        if (rc < 0) {
+            throw new IOException("failed to get file descriptor: " + rc);
+        }
+        rc = api.sd_journal_seek_tail(sdJournal);
+        logger.log(Level.INFO, "seek_tail: " + rc);
+        if (rc < 0) {
+            throw new IOException("failed to seek to tail of journal");
+        }
+        rc = api.sd_journal_previous(sdJournal);
+        logger.log(Level.INFO, "previous: " + rc);
+        rc = api.sd_journal_next(sdJournal);
+        logger.log(Level.INFO, "next: " + rc);
+        String[] strings = new String[1];
+        StringArray cursor = new StringArray(strings);
+        rc = api.sd_journal_get_cursor(sdJournal, cursor);
+        logger.log(Level.INFO, "get_cursor: " + rc);
+        System.exit(1);
+        while (true) {
+            do {
+                rc = api.sd_journal_wait(sdJournal, -1L);
+                logger.log(Level.INFO, "wait rc = " + rc);
+            } while (rc == SD_JOURNAL_NOP);
+            while ((rc = api.sd_journal_next(sdJournal)) > 0) {
+                logger.log(Level.INFO, "next: " + rc);
+                if (field != null) {
+                    Pointer dataPointer = new Memory(Native.POINTER_SIZE);
+                    Pointer sizePointer = new Memory(Native.POINTER_SIZE);
+                    rc = api.sd_journal_get_data(sdJournal, field, dataPointer, sizePointer);
+                    logger.log(Level.INFO, "get_data: " + rc);
+                    if (rc != 0) {
+                        throw new IOException("error in get_data: " + rc);
+                    }
+                    int size = sizePointer.getInt(0);
+                    byte[] b = dataPointer.getByteArray(0, size);
+                    String s = new String(b, StandardCharsets.UTF_8);
+                    if (listener != null) {
+                        listener.handleEntry(makeEntry(Collections.singletonList(s)));
+                    }
+                } else {
+                    String[] strings2 = new String[1];
+                    StringArray nextCursor = new StringArray(strings2);
+                    rc = api.sd_journal_get_cursor(sdJournal, nextCursor);
+                    logger.log(Level.INFO, "get_cursor: " + rc);
+                    if (!cursor.getString(0).equals(nextCursor.getString(0))) {
+                        cursor = nextCursor;
+                        Pointer dataPointer = new Memory(Native.POINTER_SIZE);
+                        Pointer sizePointer = new Memory(Native.POINTER_SIZE);
+                        List<String> list = new ArrayList<>();
+                        while ((rc = api.sd_journal_enumerate_data(sdJournal, dataPointer, sizePointer)) > 0) {
+                            logger.log(Level.INFO, "enumerate_data: " + rc);
+                            int size = sizePointer.getInt(0);
+                            byte[] b = dataPointer.getByteArray(0, size);
+                            String s = new String(b, StandardCharsets.UTF_8);
+                            list.add(s);
+                        }
+                        rc = api.sd_journal_restart_data(sdJournal);
+                        logger.log(Level.INFO, "restart_data: " + rc);
+                        if (listener != null) {
+                            listener.handleEntry(makeEntry(list));
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private JournalEntry makeEntry(List<String> list) {
+        DefaultJournalEntry journalEntry = new DefaultJournalEntry();
+        for (String string : list) {
+            if (string.startsWith("MESSAGE=")) {
+                journalEntry.setMessage(string.substring(8));
+                continue;
+            }
+            if (string.startsWith("MESSAGE_ID=")) {
+                journalEntry.setMessageId(string.substring(11));
+                continue;
+            }
+            if (string.startsWith("PRIORITY=")) {
+                journalEntry.setPriority(Integer.parseInt(string.substring(9)));
+                continue;
+            }
+            if (string.startsWith("CODE_FILE=")) {
+                journalEntry.setCodeFile(string.substring(10));
+                continue;
+            }
+            if (string.startsWith("CODE_LINE=")) {
+                journalEntry.setCodeLine(string.substring(10));
+                continue;
+            }
+            if (string.startsWith("CODE_FUNC=")) {
+                journalEntry.setCodeFunc(string.substring(10));
+                continue;
+            }
+            if (string.startsWith("ERRNO=")) {
+                journalEntry.setErrno(string.substring(6));
+                continue;
+            }
+            if (string.startsWith("SYSLOG_FACILITY=")) {
+                journalEntry.setSyslogFacility(string.substring(16));
+                continue;
+            }
+            if (string.startsWith("SYSLOG_IDENTIFIER=")) {
+                journalEntry.setSyslogIdentifier(string.substring(18));
+                continue;
+            }
+            if (string.startsWith("SYSLOG_PID=")) {
+                journalEntry.setSyslogPid(string.substring(11));
+                continue;
+            }
+            if (string.startsWith("SYSLOG_TIMESTAMP=")) {
+                journalEntry.setSyslogTimestamp(string.substring(17));
+                continue;
+            }
+            if (string.startsWith("SYSLOG_RAW=")) {
+                journalEntry.setSyslogRaw(string.substring(11));
+                continue;
+            }
+            if (string.startsWith("_PID=")) {
+                journalEntry.setPid(string.substring(5));
+                continue;
+            }
+            if (string.startsWith("_UID=")) {
+                journalEntry.setUid(string.substring(5));
+                continue;
+            }
+            if (string.startsWith("_GID=")) {
+                journalEntry.setGid(string.substring(5));
+                continue;
+            }
+            if (string.startsWith("_COMM=")) {
+                journalEntry.setComm(string.substring(6));
+                continue;
+            }
+            if (string.startsWith("_EXE=")) {
+                journalEntry.setExe(string.substring(5));
+                continue;
+            }
+            if (string.startsWith("_CMDLINE=")) {
+                journalEntry.setCmdLine(string.substring(9));
+                continue;
+            }
+            if (string.startsWith("_CAP_EFFECTIVE=")) {
+                journalEntry.setCapEffective(string.substring(15));
+                continue;
+            }
+            if (string.startsWith("_TRANSPORT=")) {
+                journalEntry.setTransport(string.substring(11));
+                continue;
+            }
+            if (string.startsWith("_SYSTEMD_OWNER_UID=")) {
+                journalEntry.setSystemdOwnerUid(string.substring(19));
+                continue;
+            }
+            if (string.startsWith("_SYSTEMD_UNIT=")) {
+                journalEntry.setSystemdUnit(string.substring(13));
+                continue;
+            }
+            if (string.startsWith("_SYSTEMD_USER_SLICE=")) {
+                journalEntry.setSystemdUserSlice(string.substring(19));
+                continue;
+            }
+            if (string.startsWith("_SYSTEMD_USER_UNIT=")) {
+                journalEntry.setSystemdUserUnit(string.substring(18));
+            }
+        }
+        return journalEntry;
+    }
+}

event-journald/src/main/java/org/xbib/event/journald/SystemdJournalListener.java

@@ -0,0 +1,8 @@
+package org.xbib.event.journald;
+
+import java.io.IOException;
+
+public interface SystemdJournalListener {
+
+    void handleEntry(JournalEntry entry) throws IOException;
+}

event-journald/src/main/java/org/xbib/event/journald/SystemdLibrary.java

@@ -0,0 +1,49 @@
+package org.xbib.event.journald;
+
+import com.sun.jna.Library;
+import com.sun.jna.Pointer;
+import com.sun.jna.StringArray;
+import com.sun.jna.Structure;
+import java.util.Arrays;
+import java.util.List;
+
+public interface SystemdLibrary extends Library {
+
+    int sd_notify(int fd, String msg);
+
+    int sd_journal_send(String format, Object... args);
+
+    int sd_journal_open(SdJournal sdJournal, int flag);
+
+    int sd_journal_get_fd(SdJournal sdJournal);
+
+    int sd_journal_seek_tail(SdJournal sdJournal);
+
+    int sd_journal_previous(SdJournal sdJournal);
+
+    int sd_journal_next(SdJournal sdJournal);
+
+    int sd_journal_get_cursor(SdJournal sdJournal, StringArray cursor);
+
+    int sd_journal_add_match(SdJournal sdJournal, String match, int len);
+
+    int sd_journal_wait(SdJournal sdJournal, long timeout);
+
+    int sd_journal_get_data(SdJournal sdJournal, String field, Pointer data, Pointer length);
+
+    int sd_journal_enumerate_data(SdJournal sdJournal, Pointer data, Pointer length);
+
+    int sd_journal_restart_data(SdJournal sdJournal);
+
+    class SdJournal extends Structure {
+        public int top_level_fd;
+        public String path;
+        public String prefix;
+        public String namespace;
+
+        @Override
+        protected List<String> getFieldOrder() {
+            return Arrays.asList("top_level_fd","path","prefix","namespace");
+        }
+    }
+}

event-journald/src/main/java/org/xbib/event/journald/SystemdLibraryInstance.java

@@ -0,0 +1,105 @@
+package org.xbib.event.journald;
+
+import com.sun.jna.Native;
+import com.sun.jna.Pointer;
+import com.sun.jna.StringArray;
+
+import java.util.List;
+
+/**
+ * The systemd library API, loaded by Java Native Access (JNA).
+ *
+ * The native library is loaded only once, so this class is a singleton.
+ */
+public class SystemdLibraryInstance {
+
+    private static final SystemdLibraryInstance instance = new SystemdLibraryInstance();
+
+    private final SystemdLibrary systemdLibrary;
+
+    private SystemdLibraryInstance() {
+        this.systemdLibrary = loadLibrary();
+    }
+
+    public static SystemdLibraryInstance getInstance() {
+        return instance;
+    }
+
+    public static final int SD_JOURNAL_LOCAL_ONLY = 1;
+
+    public static final int SD_JOURNAL_RUNTIME_ONLY = 2;
+
+    public static final int SD_JOURNAL_SYSTEM = 4;
+
+    public static final int SD_JOURNAL_CURRENT_USER = 8;
+
+    public static final int SD_JOURNAL_NOP = 0;
+
+    public static final int SD_JOURNAL_APPEND = 1;
+
+    public static final int SD_JOURNAL_INVALIDATE = 2;
+
+    public static final String SD_ID128_FORMAT_STR = "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x";
+
+    public int sd_journal_send(String format, Object... args) {
+        return systemdLibrary.sd_journal_send(format, args);
+    }
+
+    public int sd_journal_send(String format, List<Object> args) {
+        return systemdLibrary.sd_journal_send(format, args.toArray());
+    }
+
+    public int sd_journal_open(SystemdLibrary.SdJournal sdJournal, int flag) {
+        return systemdLibrary.sd_journal_open(sdJournal, flag);
+    }
+
+    public int sd_journal_get_fd(SystemdLibrary.SdJournal sdJournal) {
+        return systemdLibrary.sd_journal_get_fd(sdJournal);
+    }
+
+    public int sd_journal_seek_tail(SystemdLibrary.SdJournal sdJournal) {
+        return systemdLibrary.sd_journal_seek_tail(sdJournal);
+    }
+
+    public int sd_journal_previous(SystemdLibrary.SdJournal sdJournal) {
+        return systemdLibrary.sd_journal_previous(sdJournal);
+    }
+
+    public int sd_journal_next(SystemdLibrary.SdJournal sdJournal) {
+        return systemdLibrary.sd_journal_next(sdJournal);
+    }
+
+    public int sd_journal_get_cursor(SystemdLibrary.SdJournal sdJournal, StringArray cursor) {
+        return systemdLibrary.sd_journal_get_cursor(sdJournal, cursor);
+    }
+
+    public int sd_journal_add_match(SystemdLibrary.SdJournal sdJournal, String match, int len) {
+        return systemdLibrary.sd_journal_add_match(sdJournal, match, len);
+    }
+
+    public int sd_journal_wait(SystemdLibrary.SdJournal sdJournal, long timeout_musec) {
+        return systemdLibrary.sd_journal_wait(sdJournal, timeout_musec);
+    }
+
+    public int sd_journal_get_data(SystemdLibrary.SdJournal sdJournal, String field, Pointer data, Pointer length) {
+        return systemdLibrary.sd_journal_get_data(sdJournal, field, data, length);
+    }
+
+    public int sd_journal_enumerate_data(SystemdLibrary.SdJournal sdJournal, Pointer data, Pointer length) {
+        return systemdLibrary.sd_journal_enumerate_data(sdJournal, data, length);
+    }
+
+    public int sd_journal_restart_data(SystemdLibrary.SdJournal sdJournal) {
+        return systemdLibrary.sd_journal_restart_data(sdJournal);
+    }
+
+    private static SystemdLibrary loadLibrary() {
+        try {
+            return Native.load("systemd", SystemdLibrary.class);
+        } catch (UnsatisfiedLinkError e) {
+            throw new RuntimeException("Failed to load systemd library." +
+                    " Please note that JNA requires an executable temporary folder." +
+                    " It can be explicitly defined with -Djna.tmpdir", e);
+        }
+    }
+}

event-journald/src/test/java/org/xbib/event/journald/test/SystemdJournalReaderTest.java

@@ -0,0 +1,26 @@
+package org.xbib.event.journald.test;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import java.util.concurrent.Executors;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.xbib.event.journald.SystemdJournalConsumer;
+
+@EnabledOnOs({OS.LINUX})
+public class SystemdJournalReaderTest {
+
+    private static final Logger logger = Logger.getLogger(SystemdJournalReaderTest.class.getName());
+
+    @Test
+    void testConsumer() throws InterruptedException {
+        SystemdJournalConsumer consumer = new SystemdJournalConsumer(null,
+                entry -> logger.log(Level.INFO, entry.toString()));
+        Executors.newSingleThreadExecutor().submit(consumer);
+        //  consuming for some seconds
+        Thread.sleep(60000L);
+    }
+}
+

event-journald/src/test/resources/logging.properties

@@ -0,0 +1,6 @@
+handlers=java.util.logging.ConsoleHandler
+.level=ALL
+java.util.logging.SimpleFormatter.format=%1$tY-%1$tm-%1$td %1$tH:%1$tM:%1$tS.%1$tL %4$-7s [%3$s] %5$s %6$s%n
+java.util.logging.ConsoleHandler.level=ALL
+java.util.logging.ConsoleHandler.formatter=java.util.logging.SimpleFormatter
+jdk.event.security.level=INFO

event-net-http/build.gradle

@@ -1,7 +1,7 @@
 dependencies {
     api project(':event-common')
-    api libs.net.http.server.netty
-    api libs.net.http.client.netty
+    api libs.net.http.server.netty.secure
+    api libs.net.http.client.netty.secure
     implementation libs.settings.datastructures.json
     implementation libs.datastructures.json.tiny
-}
+}

event-net-http/src/main/java/org/xbib/event/net/http/HttpEventReceiverService.java

@@ -24,16 +24,16 @@ public class HttpEventReceiverService {
     public HttpService createService(String prefix) {
         return BaseHttpService.builder()
                 .setPrefix(prefix)
-                .setPath("/event/{type}")
+                .setPath("/event")
                 .setMethod(HttpMethod.POST)
                 .setHandler(ctx -> {
                     Event event = EventManager.eventFromJson(ctx.getRequest().asJson());
                     if (event.isNullEvent()) {
                         ctx.status(NOT_FOUND).done();
                     } else {
-                        eventManager.dispatch(event);
+                        eventManager.publish(event);
                         ctx.status(OK)
-                                .header(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON)
+                                .setHeader(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON)
                                 .charset(StandardCharsets.UTF_8)
                                 .body(event.toJson())
                                 .done();

event-net-http/src/test/java/org/xbib/event/net/http/test/HttpEventTest.java

@@ -1,8 +1,12 @@
 package org.xbib.event.net.http.test;
 
 import io.netty.bootstrap.Bootstrap;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
+import org.xbib.event.Event;
+import org.xbib.event.Listener;
 import org.xbib.event.common.EventManager;
+import org.xbib.event.common.GenericEventImpl;
 import org.xbib.event.net.http.HttpEventManagerService;
 import org.xbib.net.NetworkClass;
 import org.xbib.net.URL;
@@ -22,17 +26,16 @@ import org.xbib.net.http.server.netty.NettyHttpServerConfig;
 import org.xbib.net.http.server.route.BaseHttpRouter;
 import org.xbib.net.http.server.route.HttpRouter;
 import org.xbib.net.http.server.service.BaseHttpService;
-import org.xbib.net.util.JsonUtil;
 
-import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+@Disabled("for now")
 public class HttpEventTest {
 
     private static final Logger logger = Logger.getLogger(HttpEventTest.class.getName());
@@ -48,6 +51,7 @@ public class HttpEventTest {
         nettyHttpServerConfig.setDebug(true);
 
         EventManager eventManager = EventManager.builder()
+                .register("dummy", DummyEvent.class)
                 .build();
 
         HttpEventManagerService httpEventManagerService = eventManager.getEventManagerService(HttpEventManagerService.class);
@@ -58,10 +62,12 @@ public class HttpEventTest {
                         .addService(BaseHttpService.builder()
                                 .setPath("/event")
                                 .setHandler(ctx -> {
+                                    DummyEvent dummyEvent = EventManager.eventOf("dummy", DummyEvent.class);
+                                    dummyEvent.setListener(new DummyEventListener());
                                     ctx.status(HttpResponseStatus.OK)
-                                            .header(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN)
+                                            .setHeader(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN)
                                             .charset(StandardCharsets.UTF_8)
-                                            .body(ctx.getRequest().asJson())
+                                            .body(dummyEvent.toJson())
                                             .done();
                                 })
                                 .build())
@@ -93,14 +99,9 @@ public class HttpEventTest {
                                     " status = " + resp.getStatus() +
                                     " header = " + resp.getHeaders() +
                                     " body = " + body);
-                            try {
-                                Map<String, Object> map = JsonUtil.toMap(body);
-                                org.xbib.net.http.server.netty.HttpRequest httpRequest = org.xbib.net.http.server.netty.HttpRequest.builder()
-                                        .parse(map).build();
-                                logger.log(Level.INFO, "parsed http request = " + httpRequest.asJson());
-                            } catch (IOException e) {
-                                throw new RuntimeException(e);
-                            }
+                            DummyEvent dummyEvent = (DummyEvent) EventManager.eventFromJson(body);
+                            assertNotNull(dummyEvent);
+                            logger.log(Level.INFO, "dummy event transported = " + dummyEvent);
                             received.set(true);
                         })
                         .build();
@@ -109,4 +110,19 @@ public class HttpEventTest {
             assertTrue(received.get());
         }
     }
+
+    public static class DummyEvent extends GenericEventImpl {
+
+        public DummyEvent(EventManager.EventBuilder builder, Listener listener) {
+            super(builder, listener);
+        }
+    }
+
+    public static class DummyEventListener implements Listener {
+
+        @Override
+        public void listen(Event event) {
+            logger.log(Level.INFO, "got event " + event);
+        }
+    }
 }

event-net-http/src/test/resources/META-INF/services/org.xbib.net.http.server.netty.HttpChannelInitializer

@@ -0,0 +1,4 @@
+org.xbib.net.http.server.netty.http1.Http1ChannelInitializer
+org.xbib.net.http.server.netty.http2.Http2ChannelInitializer
+org.xbib.net.http.server.netty.secure.http1.Https1ChannelInitializer
+org.xbib.net.http.server.netty.secure.http2.Https2ChannelInitializer

gradle.properties

@@ -1,3 +1,3 @@
 group = org.xbib
 name = event
-version = 0.2.0
+version = 0.4.0

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.7-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

gradlew.bat

@@ -43,11 +43,11 @@ set JAVA_EXE=java.exe
 %JAVA_EXE% -version >NUL 2>&1
 if %ERRORLEVEL% equ 0 goto execute
 
-echo.
-echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
-echo.
-echo Please set the JAVA_HOME variable in your environment to match the
-echo location of your Java installation.
+echo. 1>&2
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. 1>&2
+echo. 1>&2
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
+echo location of your Java installation. 1>&2
 
 goto fail
 
@@ -57,11 +57,11 @@ set JAVA_EXE=%JAVA_HOME%/bin/java.exe
 
 if exist "%JAVA_EXE%" goto execute
 
-echo.
-echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
-echo.
-echo Please set the JAVA_HOME variable in your environment to match the
-echo location of your Java installation.
+echo. 1>&2
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% 1>&2
+echo. 1>&2
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
+echo location of your Java installation. 1>&2
 
 goto fail
 

settings.gradle

@@ -15,25 +15,24 @@ pluginManagement {
 dependencyResolutionManagement {
     versionCatalogs {
         libs {
-            version('gradle', '8.5')
-            version('datastructures', '5.0.6')
-            version('net', '4.0.4')
-            version('net-http', '4.1.0')
-            version('netty', '4.1.104.Final')
+            version('gradle', '8.7')
+            version('datastructures', '5.0.7')
+            version('net-http', '4.5.0')
+            version('netty', '4.1.109.Final')
             library('netty-handler', 'io.netty', 'netty-handler').versionRef('netty')
-            library('net', 'org.xbib', 'net').versionRef('net')
             library('datastructures-common', 'org.xbib', 'datastructures-common').versionRef('datastructures')
             library('datastructures-json-tiny', 'org.xbib', 'datastructures-json-tiny').versionRef('datastructures')
             library('settings-api', 'org.xbib', 'settings-api').versionRef('datastructures')
             library('settings-datastructures-json', 'org.xbib', 'settings-datastructures-json').versionRef('datastructures')
             library('time', 'org.xbib', 'time').version('4.0.0')
-            library('net-http-server-netty', 'org.xbib', 'net-http-server-netty-secure').versionRef('net-http')
-            library('net-http-client-netty', 'org.xbib', 'net-http-client-netty-secure').versionRef('net-http')
+            library('net-http-server-netty-secure', 'org.xbib', 'net-http-server-netty-secure').versionRef('net-http')
+            library('net-http-client-netty-secure', 'org.xbib', 'net-http-client-netty-secure').versionRef('net-http')
             library('reactivestreams', 'org.reactivestreams', 'reactive-streams').version('1.0.3')
             library('rxjava3', 'io.reactivex.rxjava3', 'rxjava').version('3.0.3')
+            library('jna', 'net.java.dev.jna', 'jna').version('5.14.0')
         }
         testLibs {
-            version('junit', '5.10.1')
+            version('junit', '5.10.2')
             library('junit-jupiter-api', 'org.junit.jupiter', 'junit-jupiter-api').versionRef('junit')
             library('junit-jupiter-params', 'org.junit.jupiter', 'junit-jupiter-params').versionRef('junit')
             library('junit-jupiter-engine', 'org.junit.jupiter', 'junit-jupiter-engine').versionRef('junit')
@@ -46,6 +45,7 @@ dependencyResolutionManagement {
 include 'event-api'
 include 'event-async'
 include 'event-common'
+include 'event-journald'
 include 'event-loop'
 include 'event-net-http'
 include 'event-pubsub'