Add code for: [2] No responsibility detected

Author: MoeweX

src/main/java/control/FBase.java

@@ -95,7 +95,9 @@ public void startup(boolean announce) throws InterruptedException, ExecutionExce
 		taskmanager.startBackgroundPollLatesConfigurationDataForResponsibleKeygroupsTask(0);
 		taskmanager.startBackgroundCheckKeygroupConfigurationsOnUpdatesTask(0);
 		taskmanager.startDetectMissingHeartbeatsTask(0, 0);
+		taskmanager.startBackgroundDetectMissingResponsibility(0);
 
+		Thread.sleep(50);
 		logger.info("FBase started, all background tasks up and running.");
 	}
 

src/main/java/tasks/TaskManager.java

@@ -17,6 +17,7 @@
 import model.messages.Envelope;
 import tasks.background.CheckKeygroupConfigurationsOnUpdatesTask;
 import tasks.background.DetectMissingHeartbeats;
+import tasks.background.DetectMissingResponsibility;
 import tasks.background.PollLatestConfigurationDataForResponsibleKeygroupsTask;
 import tasks.background.PutHeartbeatTask;
 
@@ -40,7 +41,7 @@ public enum TaskName {
 		PROCESS_MESSAGE_WITH_UNKNOWN_ENCRYPTION, CHECK_NAMING_SERVICE_CONFIGURATION_DATA,
 		B_POLL_LATEST_CONFIGURATION_DATA_FOR_RESPONSIBLE_KEYGROUPS, B_PUT_HEARTBEAT,
 		B_DETECT_MISSING_HEARTBEATS, REMOVE_MACHINE_FROM_NODE,
-		ANNOUNCE_UPDATE_OF_OWN_NODE_CONFIGURATION
+		ANNOUNCE_UPDATE_OF_OWN_NODE_CONFIGURATION, B_DETECT_MISSING_RESPONSIBILITY
 	}
 
 	public void storeHistory() {
@@ -141,7 +142,7 @@ public Future<Boolean> runRemoveMachineFromNodeTask(String machineName) {
 		Future<Boolean> future = pool.submit(new RemoveMachineFromNodeTask(machineName, fBase));
 		return future;
 	}
-	
+
 	public Future<Boolean> runAnnounceUpdateOfOwnNodeConfigurationTask() {
 		Future<Boolean> future = pool.submit(new AnnounceUpdateOfOwnNodeConfigurationTask(fBase));
 		return future;
@@ -176,4 +177,9 @@ public Future<Boolean> startDetectMissingHeartbeatsTask(int checkInterval,
 		return future;
 	}
 
+	public Future<Boolean> startBackgroundDetectMissingResponsibility(int interval) {
+		Future<Boolean> future = pool.submit(new DetectMissingResponsibility(fBase, interval));
+		return future;
+	}
+
 }

src/main/java/tasks/background/DetectMissingHeartbeats.java

@@ -16,7 +16,8 @@
 
 /**
  * 
- * This background task stores heartbeats inside the node database for this machine.
+ * This background task detects missing heartbeats of other machines and removes them from the
+ * node if detected.
  * 
  * @author jonathanhasenburg
  *
@@ -30,11 +31,11 @@ public class DetectMissingHeartbeats extends Task<Boolean> {
 	}
 
 	/**
-	 * Creates a new {@link DetectMissingHeartbeats}. If pulse <= 0, the default is used (10
+	 * Creates a new {@link DetectMissingHeartbeats}. If checkInterval <= 0, the default is used (10
 	 * sec). If toleratedMissingHeartbeat <= 0, the default is used (100 sec).
 	 * 
 	 * @param fBase
-	 * @param pulse - the interval to put heartbeats in milliseconds
+	 * @param checkInterval - the interval to put heartbeats in milliseconds
 	 * @param toleratedMissingHeartbeat - the maximum number of milliseconds tolerated for a
 	 *            heartbeat to not have been updated
 	 */

src/main/java/tasks/background/DetectMissingResponsibility.java

@@ -0,0 +1,85 @@
+package tasks.background;
+
+import java.util.Map;
+
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.javatuples.Pair;
+
+import control.FBase;
+import exceptions.FBaseException;
+import exceptions.FBaseStorageConnectorException;
+import model.data.KeygroupID;
+import tasks.Task;
+import tasks.TaskManager.TaskName;
+
+/**
+ * 
+ * This background task detects keygroups which do not have a responsiblity assigned
+ * 
+ * @author jonathanhasenburg
+ *
+ */
+public class DetectMissingResponsibility extends Task<Boolean> {
+
+	private static Logger logger = Logger.getLogger(DetectMissingResponsibility.class.getName());
+
+	static {
+		logger.setLevel(Level.INFO);
+	}
+
+	/**
+	 * Creates a new {@link DetectMissingResponsibility}. If checkInterval <= 0, the default
+	 * is used (10 sec).
+	 * 
+	 * @param fBase
+	 * @param checkInterval - the interval to put heartbeats in milliseconds
+	 */
+	public DetectMissingResponsibility(FBase fBase, int checkInterval) {
+		super(TaskName.B_DETECT_MISSING_RESPONSIBILITY, fBase);
+		if (checkInterval > 0) {
+			this.checkInterval = checkInterval;
+		}
+
+	}
+
+	private int checkInterval = 10000;
+
+	@Override
+	public Boolean executeFunctionality() {
+
+		while (!Thread.currentThread().isInterrupted()) {
+			logger.info("Looking for missing responsibilities");
+
+			try {
+				Map<KeygroupID, Pair<String, Integer>> responsibilities =
+						fBase.connector.keyGroupSubscriberMachines_listAll();
+
+				for (KeygroupID keygroupID : responsibilities.keySet()) {
+					if (responsibilities.get(keygroupID).getValue0() == null) {
+						try {
+							fBase.taskmanager.runUpdateKeygroupSubscriptionsTask(
+									fBase.connector.keygroupConfig_get(keygroupID));
+						} catch (FBaseException e) {
+							logger.error(
+									"Could not update responsibilites for keygroup " + keygroupID);
+						}
+					}
+				}
+
+			} catch (FBaseStorageConnectorException e1) {
+				logger.error("Could not read heartbeats, going back to sleep", e1);
+			}
+
+			try {
+				Thread.sleep(checkInterval);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				logger.error("Background task has been interrupted");
+			}
+		}
+		logger.info("Stopping task, because Thread was interrupted");
+		return true;
+	}
+
+}

src/test/java/scenario/SubscriptionUpdateProcessScenario.java

@@ -49,5 +49,10 @@ public void tearDown() throws Exception {
 	public void testMissingHeartbeatDetected() {
 		fail("Not yet implemented");
 	}
+	
+	@Test
+	public void testNoResponsibilityDetected() {
+		fail("Not yet implemented");
+	}
 
 }