EventStream reconnect

src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/EventProcessor.java

diff --git a/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/EventProcessor.java b/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/EventProcessor.java
index c04da49..0831b33 100644
--- a/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/EventProcessor.java
+++ b/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/EventProcessor.java
@@ -44,7 +44,8 @@
 	private PeakNotificationManager odsNotificationManager;
 	private MediaType eventMediaType;
 	private boolean closeInvoked = false;
-
+	private boolean disconnected = false;
+	
 	public EventProcessor(EventInput eventInput, NotificationListener listener,
 			PeakNotificationManager odsNotificationManager, MediaType eventMediaType) {
 		this.eventInput = eventInput;
@@ -65,6 +66,7 @@
 					odsNotificationManager
 							.processException(new NotificationException("Inbound event input stream closed!"));
 				}
+				disconnected = true;
 				return;
 			}
 
@@ -77,9 +79,16 @@
 			} catch (ProcessingException e) {
 				odsNotificationManager
 						.processException(new NotificationException("Cannot deserialize notification event!", e));
+				disconnected = true;
 				return;
 			}
 		}
+		
+		disconnected = true;
+	}
+
+	public boolean isDisconnected() {
+		return disconnected;
 	}
 
 	public void stop() {

src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/PeakNotificationManager.java

diff --git a/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/PeakNotificationManager.java b/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/PeakNotificationManager.java
index f385b2f..c1208c8 100644
--- a/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/PeakNotificationManager.java
+++ b/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/PeakNotificationManager.java
@@ -19,6 +19,7 @@
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import javax.ws.rs.client.Client;
@@ -41,6 +42,7 @@
 import org.eclipse.mdm.api.odsadapter.utils.ODSUtils;
 import org.glassfish.jersey.media.sse.EventInput;
 import org.glassfish.jersey.media.sse.SseFeature;
+import org.omg.CORBA.OBJECT_NOT_EXIST;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -91,6 +93,8 @@
 					.register(JsonMessageBodyProvider.class).build();
 
 			endpoint = client.target(url).path("events");
+			
+			Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> reconnect(), 10, 10, TimeUnit.SECONDS);
 		} catch (Exception e) {
 			throw new NotificationException("Could not create " + PeakNotificationManager.class.getName() + "!", e);
 		}
@@ -128,28 +132,7 @@
 					"Could not create registration at notification service: " + response.readEntity(String.class));
 		}
 
-		try {
-			LOGGER.info("Requesting event input for {}", registration);
-			EventInput eventInput = endpoint.path(registration)
-					.request(SseFeature.SERVER_SENT_EVENTS_TYPE)
-					.get(EventInput.class);
-
-			LOGGER.info("Received event input, starting event processor.");
-			EventProcessor processor = new EventProcessor(eventInput, listener, this, MediaType.APPLICATION_JSON_TYPE);
-
-			executor.submit(processor);
-
-			processors.put(registration, processor);
-			LOGGER.info("Event processor started.");
-		} catch (Exception e) {
-			try {
-				deregister(registration);
-			} catch (Exception ex) {
-				LOGGER.error("Exception upon deregistering!");
-			}
-			throw new NotificationException("Could not create event input stream!", e);
-		}
-
+		recreateEventStream(registration, listener);
 	}
 
 	/*
@@ -183,6 +166,56 @@
 		}
 	}
 
+	/**
+	 * Recreates the {@link EventInput}, attaches an {@link EventProcessor} to it and stores the 
+	 * {@link EventProcessor} in the processors map.
+	 * 
+	 * @param registration
+	 * @param listener
+	 * @throws NotificationException
+	 */
+	private void recreateEventStream(String registration, NotificationListener listener) throws NotificationException {
+		try {
+			LOGGER.info("Requesting event input for {}", registration);
+			EventInput eventInput = endpoint.path(registration)
+					.request(SseFeature.SERVER_SENT_EVENTS_TYPE)
+					.get(EventInput.class);
+
+			LOGGER.info("Received event input, starting event processor.");
+			EventProcessor processor = new EventProcessor(eventInput, listener, this, MediaType.APPLICATION_JSON_TYPE);
+
+			executor.submit(processor);
+
+			processors.put(registration, processor);
+			LOGGER.info("Event processor started.");
+		} catch (Exception e) {
+			try {
+				deregister(registration);
+			} catch (Exception ex) {
+				LOGGER.error("Exception upon deregistering!");
+			}
+			throw new NotificationException("Could not create event input stream!", e);
+		}
+	}
+
+	/**
+	 * Checks if any registered event processor got disconnected and tries to recreate the event stream,
+	 * if necessary.
+	 */
+	private void reconnect() {
+		
+		for (Map.Entry<String, EventProcessor> entry : processors.entrySet()) {
+			if (entry.getValue().isDisconnected()) {
+				try {
+					LOGGER.trace("Registration '{}' was disconnected and will be recreated.", entry.getKey());
+					recreateEventStream(entry.getKey(), entry.getValue().getListener());
+				} catch (NotificationException e) {
+					LOGGER.warn("Cannot recreate event stream for registration " + entry.getKey(), e);
+				}
+			}
+		}
+	}
+
 	private void close(String registration) {
 		if (processors.containsKey(registration)) {
 			EventProcessor processor = processors.get(registration);