Fix s3 db connector

Author: MoeweX

ImplementationSubscriptionManagement.md

@@ -1,5 +1,7 @@
 # Implementation Subscription Management
 
+Numbers are not as written in thesis
+
 ## Events
 
 ### [1] Missing heartbeats detected

README.md

@@ -21,7 +21,7 @@ For some of the tests, a running naming service is required. The naming service
 
 ### Handling Missed Messages
 - [x] Add message history size to machine config (node specific), cleanup config files
-- [ ] (Add message history cleanup functionality (on receiver and sender side)), thesis states not implemented
+- [ ] (Add message history cleanup functionality (on receiver and sender side), not implemented in thesis)
 
 ## Controlling FBase with Clients
  - [ ] Enable encryption and authentication

src/main/java/control/FBase.java

@@ -52,7 +52,6 @@ public class FBase {
 
 	public FBase(String configName) {
 		configuration = new Configuration(configName);
-		publisher = new Publisher("tcp://0.0.0.0", configuration.getPublisherPort());
 	}
 
 	public void startup(boolean announce, boolean backgroundTasks) throws InterruptedException,
@@ -71,6 +70,8 @@ public void startup(boolean announce, boolean backgroundTasks) throws Interrupte
 			server = new WebServer(this);
 			server.startServer();
 		}
+		publisher = new Publisher("tcp://0.0.0.0", configuration.getPublisherPort());
+
 		namingServiceSender = new NamingServiceSender(configuration.getNamingServiceAddress(),
 				configuration.getNamingServicePort(), this);
 

src/main/java/de/hasenburg/fbase/rest/jersey/NodesResource.java

@@ -22,9 +22,6 @@
  * operations.
  * 
  * 
- * TODO C <br>
- * Do not return the whole config (e.g. keys)
- * 
  * @author jonathanhasenburg
  *
  */

src/main/java/storageconnector/AbstractDBConnector.java

@@ -5,6 +5,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
 import java.util.Set;
 
 import org.javatuples.Pair;
@@ -416,5 +417,18 @@ public boolean keygroup_delete(String app, String tenant, String group)
 			throws FBaseStorageConnectorException {
 		return keygroup_delete(new KeygroupID(app, tenant, group));
 	}
+	
+	protected String generateNameString(int size) {
+		String possibleCharacterString = "abcdefghijklmnopqrstuvwxyz";
+		char[] chars = possibleCharacterString.toCharArray();
+		StringBuilder builder = new StringBuilder();
+		Random random = new Random();
+		for (int i = 0; i < size; i++) {
+			char c = chars[random.nextInt(chars.length)];
+			builder.append(c);
+		}
+		String output = builder.toString();
+		return output;
+	}
 
 }

src/main/java/storageconnector/S3DBConnector.java

@@ -73,7 +73,7 @@ public S3DBConnector(NodeID nodeID, String bucketName) {
 		this(nodeID);
 		this.bucketPrefix = bucketName;
 	}
-	
+
 	/**
 	 * Mainly used for tests.
 	 * 
@@ -86,7 +86,7 @@ public void setNodeIDAndMachineName(NodeID nodeID, String machineName) {
 	}
 
 	@Override
-	public void dbConnection_initiate() throws FBaseStorageConnectorException {
+	public String dbConnection_initiate() throws FBaseStorageConnectorException {
 		try {
 			s3 = AmazonS3ClientBuilder.defaultClient();
 		} catch (SdkClientException e) {
@@ -105,9 +105,19 @@ public void dbConnection_initiate() throws FBaseStorageConnectorException {
 				}
 			}
 		}
-		// TODO S3: set this.machinename to generated name
+
+		// read all machine names
+		Map<String, Pair<String, Long>> heartbeats_listAll = heartbeats_listAll();
+
+		// generate names until we found one that is not used, yet.
+		while (true) {
+			String name = generateNameString(5);
+			if (!heartbeats_listAll.containsKey(name)) {
+				this.machineName = name;
+				return this.machineName;
+			}
+		}
 
-		return this.machineName;
 	}
 
 	public void deleteBuckets() throws FBaseStorageConnectorException {
@@ -544,19 +554,20 @@ private String getHeartbeatsPath(String machine) {
 	}
 
 	@Override
-	public void heartbeats_update(String machine) throws FBaseStorageConnectorException {
+	public void heartbeats_update(String machine, String address)
+			throws FBaseStorageConnectorException {
 		try {
 			s3.putObject(getHeartbeatBucketName(), getHeartbeatsPath(machine),
-					Long.toString(System.currentTimeMillis()));
+					Long.toString(System.currentTimeMillis()) + "\n" + address);
 		} catch (AmazonServiceException e) {
 			throw new FBaseStorageConnectorException(e);
 		}
 	}
 
 	@Override
-	public Map<String, Long> heartbeats_listAll() throws FBaseStorageConnectorException {
+	public Map<String, Pair<String, Long>> heartbeats_listAll() throws FBaseStorageConnectorException {
 		try {
-			Map<String, Long> heartbeats = new HashMap<>();
+			Map<String, Pair<String, Long>> heartbeats = new HashMap<>();
 			ObjectListing ol = s3.listObjects(getHeartbeatBucketName());
 			while (true) {
 				List<S3ObjectSummary> objects = ol.getObjectSummaries();
@@ -565,7 +576,8 @@ public Map<String, Long> heartbeats_listAll() throws FBaseStorageConnectorExcept
 							.getObject(getHeartbeatBucketName(), os.getKey()).getObjectContent()));
 					try {
 						Long time = Long.parseLong(reader.readLine());
-						heartbeats.put(os.getKey(), time);
+						String address = reader.readLine();
+						heartbeats.put(os.getKey(), new Pair<String, Long>(address, time));
 					} catch (NumberFormatException | IOException e) {
 						logger.error("Cannot parse time from " + os.getKey(), e);
 					}
@@ -582,6 +594,16 @@ public Map<String, Long> heartbeats_listAll() throws FBaseStorageConnectorExcept
 			throw new FBaseStorageConnectorException(e);
 		}
 	}
+	
+	@Override
+	public boolean heartbeats_remove(String machine) throws FBaseStorageConnectorException {
+		try {
+			s3.deleteObject(getHeartbeatBucketName(), getHeartbeatsPath(machine));
+			return true;
+		} catch (AmazonServiceException e) {
+			throw new FBaseStorageConnectorException(e);
+		}
+	}
 
 	private String getMessageBucketName() {
 		return bucketPrefix + suffixMap.get(Suffix.MESSAGE_HISTORY);

src/test/java/storageconnector/S3DBConnectorTest.java

@@ -50,9 +50,9 @@ public static void tearDownAfterClass() throws Exception {
 
 	@Before
 	public void setUp() throws Exception {
-		connector = new S3DBConnector(new NodeID("N1"),
-				"de.hasenburg.fbase.s3dbconnector-testbucket");
-		connector.dbConnection_initiate();
+		connector =
+				new S3DBConnector(new NodeID("N1"), "de.hasenburg.fbase.s3dbconnector-testbucket");
+		logger.debug("Machine name: " + connector.dbConnection_initiate());
 		keygroupID1 = new KeygroupID("smartlight", "h1", "lightning");
 		keygroupID2 = new KeygroupID("smartlight", "h1", "brightness");
 	}
@@ -235,26 +235,36 @@ public void testHeartbeats() throws FBaseStorageConnectorException, InterruptedE
 
 		// store data
 		Long time = System.currentTimeMillis();
-		connector.heartbeats_update("M1");
-		connector.heartbeats_update("M2");
+		connector.heartbeats_update("M1", "addressM1");
+		connector.heartbeats_update("M2", "addressM2");
 
 		// check map
-		Map<String, Long> heartbeats = connector.heartbeats_listAll();
+		Map<String, Pair<String, Long>> heartbeats = connector.heartbeats_listAll();
 		assertEquals(2, heartbeats.keySet().size());
-		assertTrue(heartbeats.get("M1") >= time && heartbeats.get("M1") < (time + 1000));
-		assertTrue(heartbeats.get("M2") >= time && heartbeats.get("M2") < (time + 1000));
+		assertTrue(heartbeats.get("M1").getValue1() >= time
+				&& heartbeats.get("M1").getValue1() < (time + 1000));
+		assertTrue(heartbeats.get("M2").getValue1() >= time
+				&& heartbeats.get("M2").getValue1() < (time + 1000));
+		
+		assertEquals("addressM1", heartbeats.get("M1").getValue0());
+		assertEquals("addressM2", heartbeats.get("M2").getValue0());
 
 		Thread.sleep(1000);
-		connector.heartbeats_update("M1");
+		connector.heartbeats_update("M1", "addressMX");
 		heartbeats = connector.heartbeats_listAll();
-		assertTrue(heartbeats.get("M1") >= time + 1000 && heartbeats.get("M1") < (time + 2000));
+		assertTrue(heartbeats.get("M1").getValue1() >= time + 1000
+				&& heartbeats.get("M1").getValue1() < (time + 2000));
+		
+		assertEquals("addressMX", heartbeats.get("M1").getValue0());
 
 		logger.debug("Finished testHeartbeats.");
 	}
 
 	@Test
 	public void testMessageHistory() throws InterruptedException, FBaseException {
-		logger.debug("-------Starting testMessageHistory-------");
+		logger.debug("-------Starting testMessageHistory-------");		
+		connector.setNodeIDAndMachineName(new NodeID("N1"), "M1");
+
 		MessageID mID1 = new MessageID();
 		mID1.setMessageIDString("N1/M1/1");
 		MessageID mID2 = new MessageID();