Stable updated environment

DataINCITE · Oct 18, 2022 · ee21a56 · ee21a56
1 parent 6d21f64
commit ee21a56
Show file tree

Hide file tree

Showing 17 changed files with 156 additions and 182 deletions.
diff --git a/Data Engine/Documents/Internal Manual/Packet Spreadsheet.xlsx b/Data Engine/Documents/Internal Manual/Packet Spreadsheet.xlsx
diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/core/engine/Engine.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/core/engine/Engine.java
@@ -22,7 +22,7 @@ public Response processSTRT(Packet packet) {
 		// start local stream handler processes:
 		String lsh_type = Config.getProperty("stream", "local.stream.type");
 		if(!lsh_type.equals("null")) {
-			Response lsh_response = send("LSH", "INIT", "local_source", lsh_type);
+			Response lsh_response = send("LSH", "INIT", "source", lsh_type);
 			if(lsh_response.code() != 200)
 				return lsh_response;			
 		}

diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/consumers/SocketConsumer.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/consumers/SocketConsumer.java
@@ -4,6 +4,7 @@
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.SocketException;
+import java.util.Arrays;
 
 import org.framework.router.Response;
 import org.json.JSONObject;
@@ -46,75 +47,30 @@ public void run() {
 
 					// listen for data packets from rest socket
 					while(true) {
-						String[] data = ((String)in.readUTF()).split(Config.getProperty("app", "general.transfer.delim"));
-						String tag, sub_tag, sub_data, data_str;
-
-						if(data.length == 3) {
-							tag = data[0];
-							sub_tag = data[1];
-							data_str = data[2];
-
-							Response response = send(tag, sub_tag, data_str);
-							out.writeUTF(new JSONObject()
-									.put("response", "200")
-									.put("code", response.code())
-									.put("message", response.message())
-									.put("data", response.data())
-									.toString());
-
-						} else if(data.length == 4) {
-							tag = data[0];
-							sub_tag = data[1];
-							sub_data = data[2];
-							data_str = data[3];
-
-							Response response = send(tag, sub_tag, data_str, sub_data);
-							out.writeUTF(new JSONObject()
-									.put("response", "200")
-									.put("code", response.code())
-									.put("message", response.message())
-									.put("data", response.data())
-									.toString());
-
-						} else {
+						String[] input = ((String)in.readUTF()).split(Config.getProperty("app", "general.transfer.delim"));
+
+						// validate length is greater than 2
+						if(input.length <= 2) {
 							out.writeUTF(new JSONObject()
 									.put("response", "502")
-									.put("message", "Communication between REST API and SocketConsumer is not consistent. Differing input lengths.")
+									.put("message", "Packet processed from REST API does not contain a TAG or SUB_TAG. Review REST API endpoint code.")
 									.toString());
 						}
 
+						// extract non-essential data
+						String[] data = Arrays.copyOfRange(input, 2, input.length);
+
+						String tag = input[0];
+						String sub_tag = input[1];
 
-//						int protocol_index = 0, data_index = 0;
-//						if(data.length == 2) {
-//							protocol_index = 0;
-//							data_index = 1;
-//						}
-//						
-//						else if(data.length == 3) {
-//							protocol_index = 1;
-//							data_index = 2;
-//						}
-//						
-//						else {
-//							System.err.println("Invalid data format for SocketConsumer. Terminating.");
-//							System.exit(1);
-//						}
-//						
-//						String[] protocol = ProtocolDirectory.getProtocol(data[protocol_index]);
-//						if(protocol == null) {
-//							out.writeUTF(new JSONObject()
-//									.put("response", "403")
-//									.put("message", "Protocol does not exist. Please reference documentation.")
-//									.toString());
-//						} else {
-//							Response response = send(protocol[0], protocol[1], data[data_index], data[0]);
-//							out.writeUTF(new JSONObject()
-//									.put("response", "200")
-//									.put("code", response.code())
-//									.put("message", response.message())
-//									.put("data", response.data())
-//									.toString());	
-//						}
+						// execute valid response to engine
+						Response response = send(tag, sub_tag, data);
+						out.writeUTF(new JSONObject()
+								.put("response", "200")
+								.put("code", response.code())
+								.put("message", response.message())
+								.put("data", response.data())
+								.toString());
 					}
 				} catch(SocketException e) {
 					System.err.println("Rest Application has unexpectedly closed.");

diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/handler/OutputConsumer.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/handler/OutputConsumer.java
@@ -11,14 +11,10 @@ public OutputConsumer(OutputManager manager) {
 		this.manager = manager;
 	}
 
-	protected final Response send(String tag, String sub_tag, String data) {
+	protected final Response send(String tag, String sub_tag, String... data) {
 		return manager.send(tag, sub_tag, data);
 	}
 
-	protected final Response send(String tag, String sub_tag, String data, String sub_data) {
-		return manager.send(tag, sub_tag, data, sub_data);
-	}
-
 	protected abstract boolean init();
 	protected abstract boolean listen();
 	protected abstract boolean kill();

diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/properties/Config.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/properties/Config.java
@@ -1,7 +1,5 @@
 package org.properties;
 
-import java.io.File;
-import java.io.FileInputStream;
 import java.util.HashMap;
 import java.util.Properties;
 
@@ -10,23 +8,57 @@ public class Config {
 	private static final HashMap<String, Properties> properties;
 
 static {
+//	properties = new HashMap<String, Properties>();
+//	
+//	String[] files = new File("src/main/resources/config").list();
+//	
+//	for(String file : files) {
+//		if(file.lastIndexOf(".properties") == file.length() - 11) {
+//			String name = file.substring(0, file.length() - 11);
+//			properties.put(name, new Properties());
+//			
+//			try (FileInputStream in = new FileInputStream("src/main/resources/config/" + file)) {
+//				properties.get(name).load(in);
+//			} catch(Exception e) {
+//				e.printStackTrace();
+//				System.exit(1);
+//			}
+//		}
+//	}
+
 	properties = new HashMap<String, Properties>();
 
-	String[] files = new File("src/main/resources/config").list();
-
-	for(String file : files) {
-		if(file.lastIndexOf(".properties") == file.length() - 11) {
-			String name = file.substring(0, file.length() - 11);
-			properties.put(name, new Properties());
-
-			try (FileInputStream in = new FileInputStream("src/main/resources/config/" + file)) {
-				properties.get(name).load(in);
-			} catch(Exception e) {
-				e.printStackTrace();
-				System.exit(1);
-			}
-		}
-	}
+	Properties app_properties = new Properties();
+	app_properties.put("general.internal.delim", ":::");
+	app_properties.put("general.data.delim", ",");
+	app_properties.put("general.collection.delim", "=");
+	app_properties.put("general.transfer.delim", "&&&");
+	app_properties.put("general.data.dateformat", "yyyy-MM-dd");
+	app_properties.put("general.logging.packets", "true");
+	app_properties.put("general.logging.responses", "true");
+	properties.put("app", app_properties);
+
+	Properties stream_properties = new Properties();
+	stream_properties.put("general.consumer.types", "socket_consumer");
+	stream_properties.put("general.producer.types", "socket_producer");
+	stream_properties.put("rest.socket.address", "localhost");
+	stream_properties.put("rest.socket.port", "61100");
+	stream_properties.put("rest.socket.key", "rest-key-reserved");
+	stream_properties.put("output.socket.address", "localhost");
+	stream_properties.put("output.socket.port", "61200");
+	stream_properties.put("local.stream.type", "mongo_db");
+	stream_properties.put("mongodb.properties.uri", "mongodb://localhost:27017");
+	stream_properties.put("mongodb.database.state", "main-state-db");
+	stream_properties.put("mongodb.database.main", "main-db");
+	stream_properties.put("mongodb.auth.collection", "auth-collection");
+	stream_properties.put("mongodb.query.delim", ",");
+	stream_properties.put("polygon.request.delim", "-");
+	properties.put("stream", stream_properties);
+
+	Properties testing_properties = new Properties();
+	testing_properties.put("lsh.authorized", "true");
+	testing_properties.put("lsh.ready", "true");
+	properties.put("testing", testing_properties);
 }
 
 	public static final Properties getProperties(String name) {

diff --git a/...e/src/main/java/org/stream/external/connected/connections/TemplateExternalConnection.java b/...e/src/main/java/org/stream/external/connected/connections/TemplateExternalConnection.java
@@ -82,14 +82,18 @@ public boolean containsSubscriptionType(String type) {
 
 	@Override
 	public boolean containsRequestType(String type) {
-		// TODO Auto-generated method stub
-		return false;
+		return type.equals("valid");
 	}
 
 	@Override
 	public Object[] request(HashMap<String, String> request) {
-		// TODO Auto-generated method stub
-		return null;
+		if(request.get("key").equals("valid")) {
+			this.processRequest("valid", "col1", "data1");
+			this.processRequest("valid", "col2", "data2");
+			return new Object[] {true, ""};
+		}
+
+		return new Object[] {false, ""};
 	}
 
 	@Override

diff --git a/.../DeFi Data Engine/src/main/java/org/stream/external/handler/ExternalStreamConnection.java b/.../DeFi Data Engine/src/main/java/org/stream/external/handler/ExternalStreamConnection.java
@@ -35,16 +35,16 @@ public void processSubscription(String subscription, String data) {
 		manager.processSubscription(hash, subscription, data);
 	}
 
+	public void processRequest(String request, String data) {
+		this.processRequest(request, null, data);
+	}
+
 	public void processRequest(String request, String date, String data) {
 		if(date != null)
-			manager.processRequest(hash, 
-				getUUID() + Config.getProperty("app", "general.collection.delim") + request + Config.getProperty("app", "general.collection.delim") + date,
-				data);
+			manager.processRequest(getUUID() + Config.getProperty("app", "general.collection.delim") + request + Config.getProperty("app", "general.collection.delim") + date, data);
 
 		else
-			manager.processRequest(hash, 
-					getUUID() + Config.getProperty("app", "general.collection.delim") + request,
-					data);
+			manager.processRequest(getUUID() + Config.getProperty("app", "general.collection.delim") + request,	data);
 	}
 
 	public final String getHash() {

diff --git a/...ine/DeFi Data Engine/src/main/java/org/stream/external/handler/ExternalStreamManager.java b/...ine/DeFi Data Engine/src/main/java/org/stream/external/handler/ExternalStreamManager.java
@@ -392,7 +392,7 @@ protected void processSubscription(String hash, String subscription, String data
 	 * @param request Request which the data was received by.
 	 * @param data Data sent by the given subscription.
 	 */
-	protected void processRequest(String hash, String request, String data) {
+	protected void processRequest(String request, String data) {
 		Response lsh_response = handler.send("LSH", "PUSH", "request", request, "data", data);
 		if(lsh_response.code() != 200)
 			Logger.warn(lsh_response);

diff --git a/... Engine/src/main/java/org/stream/local/connected/connections/TemplateLocalConnection.java b/... Engine/src/main/java/org/stream/local/connected/connections/TemplateLocalConnection.java
@@ -72,7 +72,9 @@ else if(query[0].equals("corrupted"))
 	public Set<String> get(String... query) {
 		if(query[0].equals("irregular"))
 			return null;
-		return new HashSet<String>();
+		HashSet<String> out = new HashSet<String>();
+		out.add("123");
+		return out;
 	}
 
 	public boolean push(String data, String collection) {

diff --git a/...ta-Engine/DeFi Data Engine/src/main/java/org/stream/local/handler/LocalStreamHandler.java b/...ta-Engine/DeFi Data Engine/src/main/java/org/stream/local/handler/LocalStreamHandler.java
@@ -89,9 +89,9 @@ public Response processRQST(Packet packet) {
 		String request;
 		String delim = Config.getProperty("app", "general.collection.delim");
 		if(packet.containsKey("date")) {
-			request = packet.getData("uuid" + delim + packet.getData("request") + delim + packet.getData("date"));
+			request = "uuid" + delim + packet.getData("request") + delim + packet.getData("date");
 		} else {
-			request = packet.getData("uuid" + delim + packet.getData("request"));
+			request = "uuid" + delim + packet.getData("request");
 		}
 
 		if(query == null)

diff --git a/...ta-Engine/DeFi Data Engine/src/main/java/org/stream/local/handler/LocalStreamManager.java b/...ta-Engine/DeFi Data Engine/src/main/java/org/stream/local/handler/LocalStreamManager.java
@@ -8,6 +8,7 @@
 
 public class LocalStreamManager {
 
+	@SuppressWarnings("unused")
 	private final LocalStreamHandler handler;
 	private final HashMap<String, Class<? extends LocalStreamConnection>> templates;
 	private LocalStreamConnection stream;
@@ -56,7 +57,7 @@ protected boolean setStream(String type) {
 			e.printStackTrace();
 			System.exit(1);
 		}
-
+		
 		return stream.init();
 	}
 

diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/test/java/test/lsh/mongodb/TestMongoDatabase.java b/DeFi-Data-Engine/DeFi Data Engine/src/test/java/test/lsh/mongodb/TestMongoDatabase.java
@@ -46,24 +46,24 @@ public void TestSCAN() {
 		Core core = new Core();
 
 		// test contains_collection
-		assertEquals(200, core.send("LSH", "SCAN", "contains_collection, test-mongo-database").code());
-		assertEquals("true", core.send("LSH", "SCAN", "contains_collection, test-mongo-database").data());	
-		assertEquals("false", core.send("LSH", "SCAN", "contains_collection, dne").data());
-		assertEquals(445, core.send("LSH", "SCAN", "contains_collection, test-mongo-database, invalid").code());
+		assertEquals(200, core.send("LSH", "SCAN", "query", "contains_collection, test-mongo-database").code());
+		assertEquals("true", core.send("LSH", "SCAN", "query", "contains_collection, test-mongo-database").data());	
+		assertEquals("false", core.send("LSH", "SCAN", "query", "contains_collection, dne").data());
+		assertEquals(445, core.send("LSH", "SCAN", "query", "contains_collection, test-mongo-database, invalid").code());
 
 		// test contains_type
-		assertEquals(200, core.send("LSH", "SCAN", "contains_type, test-mongo-database, element1").code());
-		assertEquals("true", core.send("LSH", "SCAN", "contains_type, test-mongo-database, element1").data());
-		assertEquals("false", core.send("LSH", "SCAN", "contains_type, test-mongo-database, dne").data());
-		assertEquals(445, core.send("LSH", "SCAN", "contains_type, test-mongo-database, element1, invalid").code());
+		assertEquals(200, core.send("LSH", "SCAN", "query", "contains_type, test-mongo-database, element1").code());
+		assertEquals("true", core.send("LSH", "SCAN", "query", "contains_type, test-mongo-database, element1").data());
+		assertEquals("false", core.send("LSH", "SCAN", "query", "contains_type, test-mongo-database, dne").data());
+		assertEquals(445, core.send("LSH", "SCAN", "query", "contains_type, test-mongo-database, element1, dne").code());
 
 		// test contains_item
-		assertEquals(200, core.send("LSH", "SCAN", "contains_item, test-mongo-database, element1, e1").code());
-		assertEquals("true", core.send("LSH", "SCAN", "contains_item, test-mongo-database, element1, e1").data());
-		assertEquals("false", core.send("LSH", "SCAN", "contains_item, test-mongo-database, element1, dne").data());
-		assertEquals(445, core.send("LSH", "SCAN", "contains_item, test-mongo-database, element1, e1, invalid").code());
+		assertEquals(200, core.send("LSH", "SCAN", "query", "contains_item, test-mongo-database, element1, e1").code());
+		assertEquals("true", core.send("LSH", "SCAN", "query", "contains_item, test-mongo-database, element1, e1").data());
+		assertEquals("false", core.send("LSH", "SCAN", "query", "contains_item, test-mongo-database, element1, dne").data());
+		assertEquals(445, core.send("LSH", "SCAN", "query", "contains_item, test-mongo-database, element1, e1, invalid").code());
 
-		assertEquals(500, core.send("LSH", "SCAN", "").code());
+		assertEquals(500, core.send("LSH", "SCAN", "query", "").code());
 	}
 
 	@Test
@@ -72,36 +72,37 @@ public void TestRQST() {
 		Core core = new Core();
 
 		// test get_all
-		assertEquals(200, core.send("LSH", "RQST", "get_all, test-mongo-database", "null").code());
-		assertEquals(445, core.send("LSH", "RQST", "get_all, test-mongo-database, invalid", "null").code());
-		assertEquals(446, core.send("LSH", "RQST", "get_all, dne", "null").code());
+		assertEquals(200, core.send("LSH", "RQST", "uuid", "external_template", "request", "test-mongo-database", "query", "get_all, test-mongo-database", "destination", "null").code());
+		assertEquals(445, core.send("LSH", "RQST", "uuid", "external_template", "request", "test-mongo-database", "query", "get_all, test-mongo-database, invalid", "destination", "null").code());
+		assertEquals(446, core.send("LSH", "RQST", "uuid", "external_template", "request", "test-mongo-database", "query", "get_all, dne", "destination", "null").code());
 
 		// test get_item
-		assertEquals(200, core.send("LSH", "RQST", "get_item, test-mongo-database, element1, e1", "null").code());
-		assertEquals(446, core.send("LSH", "RQST", "get_item, test-mongo-database, element1, e2", "null").code());
+		assertEquals(200, core.send("LSH", "RQST", "uuid", "external_template", "request", "test-mongo-database", "query", "get_item, test-mongo-database, element1, e1", "destination", "null").code());
+		assertEquals(446, core.send("LSH", "RQST", "uuid", "external_template", "request", "test-mongo-database", "query", "get_item, test-mongo-database, element1, e2", "destination", "null").code());
 
-		assertEquals(500, core.send("LSH", "RQST", "").code());
+		assertEquals(500, core.send("LSH", "RQST", "uuid", "").code());
 	}
 
 	@Test
 	public void TestRQSTDated() {
 		Config.setProperty("stream", "mongodb.database.main", "testing");
 		Core core = new Core();
 
-		assertEquals(200, core.send("SRC", "INIT", "external_template:::key").code());
+		assertEquals(200, core.send("SRC", "INIT", "source", "external_template", "key", "key").code());
 
-		assertEquals(200, core.send("SRC", "RQST", "key:::10-09-2022:::15-09-2022:::template-external-request:::get_all, template-external-request", "null").code());
+		assertEquals(200, core.send("SRC", "RQST", "key", "key", "request", "test-mongo-database", "query", "get_all, test-mongo-database", "destination", "null",
+									"start_date", "2020-09-01", "end_date", "2020-09-05").code());
 	}
 
 	@Test
 	public void TestPUSH() {
 		Config.setProperty("stream", "mongodb.database.main", "testing");
 		Core core = new Core();
 
-		assertEquals(200, core.send("LSH", "PUSH", "element1, e1:::test-mongo-database").code());
-		assertEquals(200, core.send("LSH", "PUSH", "element1, e1, element2, e2:::test-mongo-database").code());
-		assertEquals(449, core.send("LSH", "PUSH", "element1, e1, invalid:::test-mongo-database").code());
+		assertEquals(200, core.send("LSH", "PUSH", "data", "element1, e1", "collection", "test-mongo-database").code());
+		assertEquals(200, core.send("LSH", "PUSH", "data", "element1, e2", "collection", "test-mongo-database").code());
+		assertEquals(449, core.send("LSH", "PUSH", "data", "element1, e2, invalid", "collection", "test-mongo-database").code());
 
-		assertEquals(500, core.send("LSH", "PUSH", "invalid").code());
+		assertEquals(500, core.send("LSH", "PUSH", "data", "").code());
 	}
 }