joerg/event
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: joerg:ab2cf01520828616c70327cb9ea732e5b808f3b3
Branches Tags
joerg:main
joerg:0.1.0
joerg:0.0.12
joerg:0.0.10
joerg:0.0.9
joerg:0.0.8
joerg:0.0.7
joerg:0.0.6
joerg:0.0.5
joerg:0.0.4
..
compare: joerg:788ce1b7d7859d7211f2b7bcb38271da5c464561
Branches Tags
joerg:main
joerg:0.1.0
joerg:0.0.12
joerg:0.0.10
joerg:0.0.9
joerg:0.0.8
joerg:0.0.7
joerg:0.0.6
joerg:0.0.5
joerg:0.0.4

No commits in common. "ab2cf01520828616c70327cb9ea732e5b808f3b3" and "788ce1b7d7859d7211f2b7bcb38271da5c464561" have entirely different histories.
ab2cf01520 ... 788ce1b7d7

50 changed files with 468 additions and 1631 deletions
Show stats Expand all files Collapse all files

17
build.gradle
View file

@ -13,7 +13,7 @@ ext {
user = 'joerg'
name = 'event'
description = 'Event framework for Java (NIO paths, files, timers, journals)'
inceptionYear = '2024'
inceptionYear = '2021'
url = 'https://xbib.org/' + user + '/' + name
scmUrl = 'https://xbib.org/' + user + '/' + name
scmConnection = 'scm:git:git://xbib.org/' + user + '/' + name + '.git'
@ -28,7 +28,20 @@ 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
}
*/

4
event-api/src/main/java/org/xbib/event/Event.java
View file

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

8
event-api/src/main/java/org/xbib/event/GenericEvent.java
View file

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

12
event-api/src/main/java/org/xbib/event/Payload.java
View file

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

1
event-common/build.gradle
View file

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

24
event-common/src/main/java/module-info.java
View file

@ -1,26 +1,24 @@
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;
}

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

@ -1,8 +1,6 @@
package org.xbib.event.clock;
import java.util.List;
import org.xbib.event.Event;
import org.xbib.event.common.AbstractEventManagerService;
import org.xbib.event.bus.EventBus;
import org.xbib.event.common.EventManager;
import org.xbib.event.common.EventManagerService;
import org.xbib.settings.Settings;
@ -10,29 +8,30 @@ 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 extends AbstractEventManagerService implements EventManagerService {
public class ClockEventManagerService implements EventManagerService {
private static final Logger logger = Logger.getLogger(ClockEventManagerService.class.getName());
private EventManager eventManager;
private CronSchedule<Integer> cronSchedule;
private CronSchedule<Boolean> cronSchedule;
private List<String> suspended;
public ClockEventManagerService() {
super();
}
@Override
public ClockEventManagerService init(EventManager eventManager) {
this.eventManager = eventManager;
Settings settings = eventManager.getSettings();
EventBus eventBus = eventManager.getEventBus();
this.suspended = new ArrayList<>();
ThreadFactory threadFactory = new ThreadFactory() {
int n = 1;
@Override
@ -41,44 +40,45 @@ public class ClockEventManagerService extends AbstractEventManagerService implem
}
};
ScheduledExecutorService scheduledExecutorService =
Executors.newScheduledThreadPool(eventManager.getSettings().getAsInt("pool.size", 2), threadFactory);
Executors.newScheduledThreadPool(settings.getAsInt("pool.size", 2), threadFactory);
this.cronSchedule = new CronSchedule<>(scheduledExecutorService);
for (Map.Entry<String, Settings> mapEntry : eventManager.getSettings().getGroups("event.clock").entrySet()) {
for (Map.Entry<String, Settings> mapEntry : settings.getGroups("event.clock").entrySet()) {
String name = mapEntry.getKey();
Settings entrySettings = mapEntry.getValue();
if (entrySettings.getAsBoolean("enabled", true)) {
String entry = entrySettings.get("entry");
String type = entrySettings.get("type", "clock");
if (entry != null && type != null) {
if (entry != null) {
try {
ClockEventService clockEventService = new ClockEventService(this, name, type);
ClockEventService clockEventService = new ClockEventService(this, eventBus, name);
cronSchedule.add(name, CronExpression.parse(entry), clockEventService);
logger.log(Level.INFO, "cron job " + name + "with type " + type + " scheduled on " + entry);
logger.log(Level.INFO, "cron job " + name + " 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 in configuration not enabled, name = " + name );
logger.log(Level.WARNING, "clock event service " + name + " in configuration not enabled");
}
}
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();
}
}

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

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

23
event-common/src/main/java/org/xbib/event/clock/SimpleClockEventConsumer.java Normal file
View file

@ -0,0 +1,23 @@
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());
}
}

25
event-common/src/main/java/org/xbib/event/common/AbstractEventManagerService.java
View file

@ -1,25 +0,0 @@
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;
}
}

202
event-common/src/main/java/org/xbib/event/common/EventManager.java
View file

@ -3,7 +3,6 @@ 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;
@ -12,17 +11,16 @@ 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;
@ -33,15 +31,12 @@ 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 extends AbstractEventManagerService implements EventManagerService {
public final class EventManager {
private static final Logger logger = Logger.getLogger(EventManager.class.getName());
@ -52,7 +47,6 @@ public final class EventManager extends AbstractEventManagerService implements E
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);
@ -65,7 +59,7 @@ public final class EventManager extends AbstractEventManagerService implements E
}
logger.log(Level.INFO, "installed events = " + eventTypes.keySet());
this.eventManagerServices = new HashMap<>();
eventManagerServices.put(this.getClass(), this);
eventManagerServices.put(GenericEventManagerService.class, new GenericEventManagerService().init(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));
@ -76,70 +70,6 @@ public final class EventManager extends AbstractEventManagerService implements E
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();
}
@ -164,11 +94,19 @@ public final class EventManager extends AbstractEventManagerService implements E
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);
}
@ -185,24 +123,7 @@ public final class EventManager extends AbstractEventManagerService implements E
return (PathEventManagerService) eventManagerServices.get(PathEventManagerService.class);
}
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 {
public void close() throws IOException {
for (EventConsumer eventConsumer : builder.eventConsumers) {
if (eventConsumer instanceof Closeable closeable) {
closeable.close();
@ -210,9 +131,7 @@ public final class EventManager extends AbstractEventManagerService implements E
}
for (EventManagerService service : eventManagerServices.values()) {
try {
if (service != this) {
service.shutdown();
}
service.shutdown();
} catch (IOException e) {
logger.log(Level.SEVERE, e.getMessage(), e);
}
@ -325,31 +244,36 @@ public final class EventManager extends AbstractEventManagerService implements E
}
}
private static class WrappedListener implements Listener {
public static Event eventOf(String eventType,
String code,
String message,
Path path) {
return eventBuilder()
.setType(eventType)
.setCode(code)
.setMessage(message)
.setPath(path)
.build();
}
private final Listener listener;
public static Event eventOf(String eventType,
Instant scheduled) {
return eventBuilder()
.setType(eventType)
.setScheduledFor(scheduled)
.build();
}
private int size;
public static Event eventFromFile(Path file) throws IOException {
return eventFromJson(Files.readString(file));
}
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 Event eventFromJson(String json) {
Map<String, Object> map = Json.toMap(json);
return eventBuilder()
.setType(map.getOrDefault("type", "generic").toString())
.setPayload(new Payload(map))
.build();
}
public static class EventImpl implements Event {
@ -380,13 +304,18 @@ public final class EventManager extends AbstractEventManagerService implements E
}
@Override
public PayloadImpl getPayload() {
public Payload getPayload() {
return builder.payload;
}
@Override
public Instant getInstant() {
return builder.instant;
public Instant getCreated() {
return builder.created;
}
@Override
public Instant getScheduledFor() {
return builder.scheduled;
}
@Override
@ -409,24 +338,8 @@ public final class EventManager extends AbstractEventManagerService implements E
return builder.fileSize;
}
@Override
public String toJson() throws IOException {
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();
return Json.toString(builder.payload);
}
@Override
@ -482,9 +395,11 @@ public final class EventManager extends AbstractEventManagerService implements E
String message;
Instant instant;
Instant scheduled;
PayloadImpl payload;
Instant created;
Payload payload;
Path path;
@ -524,12 +439,12 @@ public final class EventManager extends AbstractEventManagerService implements E
return this;
}
public EventBuilder setInstant(Instant instant) {
this.instant = instant;
public EventBuilder setScheduledFor(Instant scheduled) {
this.scheduled = scheduled;
return this;
}
public EventBuilder setPayload(PayloadImpl payload) {
public EventBuilder setPayload(Payload payload) {
this.payload = payload;
return this;
}
@ -567,7 +482,7 @@ public final class EventManager extends AbstractEventManagerService implements E
cl = NullEvent.class;
}
try {
this.instant = Instant.now();
this.created = Instant.now();
if (listener != null) {
return cl.getDeclaredConstructor(EventBuilder.class, Listener.class).newInstance(this, listener);
} else {
@ -592,4 +507,5 @@ public final class EventManager extends AbstractEventManagerService implements E
return pos >= 0 ? name.substring(pos + 1) : null;
}
}
}

11
event-common/src/main/java/org/xbib/event/common/GenericEventImpl.java
View file

@ -1,27 +1,28 @@
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 implements GenericEvent {
public class GenericEventImpl extends EventManager.EventImpl {
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 = listener;
this.builder.listener = Objects.requireNonNull(listener);
}
@Override
public GenericEventImpl setListener(Listener listener) {
this.builder.listener = Objects.requireNonNull(listener);
return this;
}
@Override
public void received() {
builder.listener.listen(this);
}

35
event-common/src/main/java/org/xbib/event/common/PayloadImpl.java
View file

@ -1,35 +0,0 @@
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;
}
}

77
event-common/src/main/java/org/xbib/event/generic/GenericEventManagerService.java Normal file
View file

@ -0,0 +1,77 @@
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);
}
}
}
}

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

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

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

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

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

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

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

@ -4,6 +4,8 @@ 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;
@ -19,17 +21,18 @@ 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<Boolean>, Closeable {
public class PathEventService implements Callable<Integer>, 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;
@ -40,16 +43,20 @@ public class PathEventService implements Callable<Boolean>, 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 = Objects.requireNonNull(path, "path must not be null");
this.path = path;
this.eventType = eventType;
this.lifetime = lifetime;
this.watchService = path.getFileSystem().newWatchService();
@ -60,17 +67,9 @@ public class PathEventService implements Callable<Boolean>, 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 Boolean call() {
public Integer call() {
try {
logger.log(Level.INFO, "watch service running on " + path.resolve(Event.INCOMING));
while (keepWatching && watchService != null) {
@ -87,10 +86,10 @@ public class PathEventService implements Callable<Boolean>, 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.getSuspended().contains(name)) {
pathEventManager.failEvent(eventType, p);
if (pathEventManager.getSuspendedQueues().contains(name)) {
failEvent(p);
} else {
pathEventManager.publish(eventType, p);
postEvent(p);
}
}
watchKey.reset();
@ -103,7 +102,7 @@ public class PathEventService implements Callable<Boolean>, Closeable {
} catch (Exception e) {
logger.log(Level.SEVERE, e.getMessage(), e);
}
return true;
return eventCount;
}
@Override
@ -121,7 +120,7 @@ public class PathEventService implements Callable<Boolean>, Closeable {
public void drainIncoming() throws IOException {
try (DirectoryStream<Path> stream = Files.newDirectoryStream(path.resolve(Event.INCOMING))) {
stream.forEach(p -> pathEventManager.publish(eventType, p));
stream.forEach(this::postEvent);
}
}
@ -150,6 +149,21 @@ public class PathEventService implements Callable<Boolean>, 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;

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

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

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

@ -1,8 +1,7 @@
package org.xbib.event.timer;
import org.xbib.event.Event;
import org.xbib.event.common.PayloadImpl;
import org.xbib.event.common.AbstractEventManagerService;
import org.xbib.event.Payload;
import org.xbib.event.bus.EventBus;
import org.xbib.event.common.EventManager;
import org.xbib.event.common.EventManagerService;
import org.xbib.event.persistence.FilePersistenceStore;
@ -21,53 +20,38 @@ import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
public class TimerEventManagerService extends AbstractEventManagerService implements EventManagerService {
public class TimerEventManagerService implements EventManagerService {
private static final Logger logger = Logger.getLogger(TimerEventManagerService.class.getName());
private EventManager eventManager;
private final Map<String, TimerEventService> services;
private Map<String, TimerEventService> services;
public TimerEventManagerService() {
super();
this.services = new LinkedHashMap<>();
}
@Override
public TimerEventManagerService init(EventManager eventManager) {
this.eventManager = eventManager;
for (Map.Entry<String, Settings> entry : eventManager.getSettings().getGroups("event.timer").entrySet()) {
Settings settings = eventManager.getSettings();
EventBus eventBus = eventManager.getEventBus();
this.services = new LinkedHashMap<>();
for (Map.Entry<String, Settings> entry : settings.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(this, name, ZoneId.systemDefault(), persistenceStore);
TimerEventService timerEventService = new TimerEventService(eventBus, 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;
}
@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 {
public boolean publish(String name,
String timeSpec,
Payload payload) throws ParseException, IOException {
if (services.containsKey(name)) {
Span span = Chronic.parse(timeSpec);
if (span != null) {
@ -85,9 +69,9 @@ public class TimerEventManagerService extends AbstractEventManagerService implem
return false;
}
public boolean schedule(String service,
Instant instant,
PayloadImpl payload) throws IOException {
public boolean publish(String service,
Instant instant,
Payload payload) throws IOException {
if (services.containsKey(service)) {
services.get(service).schedule(instant, payload);
return true;
@ -103,4 +87,12 @@ public class TimerEventManagerService extends AbstractEventManagerService implem
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();
}
}
}

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

@ -1,7 +1,8 @@
package org.xbib.event.timer;
import org.xbib.event.Event;
import org.xbib.event.common.PayloadImpl;
import org.xbib.event.Payload;
import org.xbib.event.bus.EventBus;
import org.xbib.event.common.EventManager;
import org.xbib.event.persistence.PersistenceStore;
@ -25,9 +26,7 @@ class TimerEventService implements Closeable {
private static final Logger logger = Logger.getLogger(TimerEventService.class.getName());
private final TimerEventManagerService timerEventManagerService;
private final String name;
private final EventBus eventBus;
private final ZoneId zoneId;
@ -35,12 +34,11 @@ class TimerEventService implements Closeable {
private final Timer timer;
public TimerEventService(TimerEventManagerService timerEventManagerService,
public TimerEventService(EventBus eventBus,
String name,
ZoneId zoneId,
PersistenceStore<String, Object> persistenceStore) throws IOException {
this.timerEventManagerService = timerEventManagerService;
this.name = name;
this.eventBus = eventBus;
this.zoneId = zoneId;
this.persistenceStore = persistenceStore;
this.timer = new Timer();
@ -48,7 +46,7 @@ class TimerEventService implements Closeable {
logger.log(Level.INFO, "timer event service " + name + " loaded and rescheduled, " + tasknum + " timer tasks");
}
void schedule(Instant instant, PayloadImpl payload) throws IOException {
void schedule(Instant instant, Payload payload) throws IOException {
String scheduled = instant.atZone(zoneId).format(DateTimeFormatter.ISO_DATE_TIME);
payload.put("scheduled", scheduled);
TimerEventTask timerEventTask = new TimerEventTask(payload);
@ -64,7 +62,7 @@ class TimerEventService implements Closeable {
persistenceStore.clear();
persistenceStore.commit();
for (Map<String, Object> task : tasks) {
PayloadImpl payload = new PayloadImpl(task);
Payload payload = new Payload(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");
@ -89,9 +87,9 @@ class TimerEventService implements Closeable {
public class TimerEventTask extends TimerTask {
private final PayloadImpl payload;
private final Payload payload;
public TimerEventTask(PayloadImpl payload) throws IOException {
public TimerEventTask(Payload payload) throws IOException {
this.payload = payload;
persistenceStore.insert("tasks", this.payload);
}
@ -99,16 +97,12 @@ 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();
timerEventManagerService.publish(timerEvent);
logger.log(Level.FINE, "posting timer event " + timerEvent.getClass().getName() + " with payload = " + payload);
eventBus.post(timerEvent);
logger.log(Level.FINE, "persistence before remove: " + persistenceStore.get("tasks"));
if (persistenceStore.remove("tasks", this.payload)) {
logger.log(Level.FINE, "removal done");

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

@ -1,56 +0,0 @@
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);
}
}
}
}

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

@ -13,15 +13,13 @@ 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.shutdown();
eventManager.close();
}
}

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

@ -1,11 +0,0 @@
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);
}
}

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

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

10
event-common/src/test/java/org/xbib/event/common/GenericEventManagerTest.java → event-common/src/test/java/org/xbib/event/generic/GenericEventManagerTest.java
View file

@ -1,9 +1,11 @@
package org.xbib.event.common;
package org.xbib.event.generic;
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;
@ -30,7 +32,7 @@ public class GenericEventManagerTest {
.setType("generic")
.setListener(e -> logger.log(Level.INFO, "received event " + e))
.build();
eventManager.publish(event);
eventManager.getGenericEventManagerService().post(event);
// we must wait for a certain time because we do not use a future
Thread.sleep(500L);
assertEquals(1, consumer.getCount());
@ -50,7 +52,7 @@ public class GenericEventManagerTest {
future.complete(e);
})
.build();
eventManager.publish(event);
eventManager.getGenericEventManagerService().post(event);
Event e = future.get(1000L, TimeUnit.MILLISECONDS);
assertNotNull(e);
}
@ -74,7 +76,7 @@ public class GenericEventManagerTest {
future.complete(e);
})
.build();
eventManager.publish((GenericEventImpl) event, future);
eventManager.getGenericEventManagerService().post((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");

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

@ -21,13 +21,11 @@ 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);
@ -39,6 +37,6 @@ public class FileFollowEventManagerTest {
Thread.sleep(1000L);
Files.delete(testTxt);
Files.delete(path);
eventManager.shutdown();
eventManager.close();
}
}

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

@ -1,11 +1,7 @@
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;
@ -14,7 +10,6 @@ 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;
@ -36,11 +31,10 @@ public class PathEventManagerTest {
.build();
Path testTxt = path.resolve("incoming").resolve("test.txt");
try (BufferedWriter bufferedWriter = Files.newBufferedWriter(testTxt)) {
bufferedWriter.write("{\"Hello\":\"world\"}");
bufferedWriter.write("Hello");
}
Thread.sleep(1000L);
eventManager.shutdown();
// extra destroy to clean up test
Thread.sleep(2000L);
eventManager.close();
eventManager.getPathEventManagerService().destroy();
}
@ -50,8 +44,8 @@ public class PathEventManagerTest {
TestPathEventConsumer consumer = new TestPathEventConsumer();
Settings settings = Settings.settingsBuilder()
.put("event.path.testpathevent.enabled", "true")
.put("event.path.testpathevent.type", "path-ext")
.put("event.path.testpathevent.path", path.toString())
.put("event.path.testpathevent.type", "path-ext")
.build();
EventManager eventManager = EventManager.builder()
.setSettings(settings)
@ -60,32 +54,10 @@ public class PathEventManagerTest {
.build();
Path testTxt = path.resolve("incoming").resolve("test.txt");
try (BufferedWriter bufferedWriter = Files.newBufferedWriter(testTxt)) {
bufferedWriter.write("{\"Hello\":\"world\"}");
bufferedWriter.write("Hello");
}
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
Thread.sleep(2000L);
eventManager.close();
eventManager.getPathEventManagerService().destroy();
}
@ -96,25 +68,4 @@ 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());
}
}
}

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

@ -1,10 +0,0 @@
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);
}
}

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

@ -1,6 +1,7 @@
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;
@ -13,7 +14,7 @@ public class TestFileFollowEventConsumer implements EventConsumer {
@Subscribe
@AllowConcurrentEvents
void onEvent(TestFileFollowEvent event) {
logger.log(Level.INFO, "received test filefollow event, path = " + event.getPath() + " content = " + event.getMessage());
void onEvent(FileFollowEvent event) {
logger.log(Level.INFO, "received filefollow event, path = " + event.getPath() + " content = " + event.getMessage());
}
}

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

@ -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.getInstant());
logger.log(Level.INFO, "received timer event on " + Instant.now() + " event instant = " + event.getCreated());
}
}

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

@ -1,7 +1,7 @@
package org.xbib.event.timer;
import org.junit.jupiter.api.Test;
import org.xbib.event.common.PayloadImpl;
import org.xbib.event.Payload;
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();
PayloadImpl payload = new PayloadImpl(Map.of("a", "b"));
timerEventManager.schedule("testtimerevent", Instant.now().plusSeconds(5L), payload);
Payload payload = new Payload(Map.of("a", "b"));
timerEventManager.publish("testtimerevent", Instant.now().plusSeconds(5L), payload);
Thread.sleep(10000L);
timerEventManager.shutdown();
}

3
event-journald/build.gradle
View file

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

421
event-journald/src/main/java/org/xbib/event/journald/DefaultJournalEntry.java
View file

@ -1,421 +0,0 @@
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;
}
}

80
event-journald/src/main/java/org/xbib/event/journald/JournalEntry.java
View file

@ -1,80 +0,0 @@
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);
}

240
event-journald/src/main/java/org/xbib/event/journald/SystemdJournalConsumer.java
View file

@ -1,240 +0,0 @@
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;
}
}

8
event-journald/src/main/java/org/xbib/event/journald/SystemdJournalListener.java
View file

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

49
event-journald/src/main/java/org/xbib/event/journald/SystemdLibrary.java
View file

@ -1,49 +0,0 @@
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");
}
}
}

105
event-journald/src/main/java/org/xbib/event/journald/SystemdLibraryInstance.java
View file

@ -1,105 +0,0 @@
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);
}
}
}

26
event-journald/src/test/java/org/xbib/event/journald/test/SystemdJournalReaderTest.java
View file

@ -1,26 +0,0 @@
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);
}
}

6
event-journald/src/test/resources/logging.properties
View file

@ -1,6 +0,0 @@
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

4
event-net-http/build.gradle
View file

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

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

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

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

@ -1,12 +1,8 @@
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;
@ -26,16 +22,17 @@ 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());
@ -51,7 +48,6 @@ public class HttpEventTest {
nettyHttpServerConfig.setDebug(true);
EventManager eventManager = EventManager.builder()
.register("dummy", DummyEvent.class)
.build();
HttpEventManagerService httpEventManagerService = eventManager.getEventManagerService(HttpEventManagerService.class);
@ -62,12 +58,10 @@ 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)
.setHeader(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN)
.header(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN)
.charset(StandardCharsets.UTF_8)
.body(dummyEvent.toJson())
.body(ctx.getRequest().asJson())
.done();
})
.build())
@ -99,9 +93,14 @@ public class HttpEventTest {
" status = " + resp.getStatus() +
" header = " + resp.getHeaders() +
" body = " + body);
DummyEvent dummyEvent = (DummyEvent) EventManager.eventFromJson(body);
assertNotNull(dummyEvent);
logger.log(Level.INFO, "dummy event transported = " + dummyEvent);
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);
}
received.set(true);
})
.build();
@ -110,19 +109,4 @@ 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);
}
}
}

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

@ -1,4 +0,0 @@
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

2
gradle.properties
View file

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

BIN
gradle/wrapper/gradle-wrapper.jar vendored
View file

Binary file not shown.

2
gradle/wrapper/gradle-wrapper.properties vendored
View file

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

20
gradlew.bat vendored
View file

@ -43,11 +43,11 @@ set JAVA_EXE=java.exe
%JAVA_EXE% -version >NUL 2>&1
if %ERRORLEVEL% equ 0 goto execute
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
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.
goto fail
@ -57,11 +57,11 @@ set JAVA_EXE=%JAVA_HOME%/bin/java.exe
if exist "%JAVA_EXE%" goto execute
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
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.
goto fail

18
settings.gradle
View file

@ -15,24 +15,25 @@ pluginManagement {
dependencyResolutionManagement {
versionCatalogs {
libs {
version('gradle', '8.7')
version('datastructures', '5.0.7')
version('net-http', '4.5.0')
version('netty', '4.1.109.Final')
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')
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-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('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('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.2')
version('junit', '5.10.1')
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')
@ -45,7 +46,6 @@ dependencyResolutionManagement {
include 'event-api'
include 'event-async'
include 'event-common'
include 'event-journald'
include 'event-loop'
include 'event-net-http'
include 'event-pubsub'