Merge branch 'main' of https://github.rpi.edu/DataINCITE/IDEA-DeFi-CRAFT

into main
DataINCITE · Dec 14, 2022 · 11db9f4 · 11db9f4
2 parents f09e27d + d6e1de8
commit 11db9f4
Show file tree

Hide file tree

Showing 30 changed files with 346 additions and 192 deletions.
diff --git a/.github/workflows/dockerpushandrun.yml b/.github/workflows/dockerpushandrun.yml
@@ -0,0 +1,40 @@
+name: CD to Docker
+on:
+  push:
+    branches: [ main ]
+    paths: 'DeFi-Data-Engine/**'
+  workflow_dispatch:
+
+jobs:
+  buildDocker:
+    runs-on: [ self-hosted ]
+    defaults:
+      run:
+        working-directory: "DeFi-Data-Engine/"
+
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Stop Existing Images
+      run: sudo docker compose down
+
+    - name: Build
+      run: sudo docker compose build
+
+    - name: Push to Docker
+      run: sudo docker compose push
+
+  runDocker:
+    needs: buildDocker
+    runs-on: [ self-hosted ]
+
+    defaults:
+      run:
+        working-directory: "DeFi-Data-Engine/"
+
+    steps:
+    - name: Create Containers
+      run: sudo docker compose create
+
+    - name: Start services
+      run: sudo docker compose start
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
diff --git a/DeFi-Data-Engine/DeFi Data Engine/.classpath b/DeFi-Data-Engine/DeFi Data Engine/.classpath
@@ -25,12 +25,33 @@
 		</attributes>
 	</classpathentry>
 	<classpathentry kind="src" path="src/main/webapp/WEB-INF"/>
-	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/jdk-17.0.2"/>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
 	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
 		<attributes>
 			<attribute name="maven.pomderived" value="true"/>
 		</attributes>
 	</classpathentry>
 	<classpathentry kind="con" path="org.eclipse.jdt.junit.JUNIT_CONTAINER/4"/>
+	<classpathentry kind="src" path="target/generated-sources/annotations">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+			<attribute name="ignore_optional_problems" value="true"/>
+			<attribute name="m2e-apt" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="target/generated-test-sources/test-annotations">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+			<attribute name="ignore_optional_problems" value="true"/>
+			<attribute name="m2e-apt" value="true"/>
+			<attribute name="test" value="true"/>
+		</attributes>
+	</classpathentry>
 	<classpathentry kind="output" path="target/classes"/>
 </classpath>
diff --git a/DeFi-Data-Engine/DeFi Data Engine/.settings/org.eclipse.jdt.apt.core.prefs b/DeFi-Data-Engine/DeFi Data Engine/.settings/org.eclipse.jdt.apt.core.prefs
@@ -0,0 +1,2 @@
+eclipse.preferences.version=1
+org.eclipse.jdt.apt.aptEnabled=false
diff --git a/DeFi-Data-Engine/DeFi Data Engine/.settings/org.eclipse.jdt.core.prefs b/DeFi-Data-Engine/DeFi Data Engine/.settings/org.eclipse.jdt.core.prefs
@@ -1,9 +1,9 @@
 eclipse.preferences.version=1
 org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
 org.eclipse.jdt.core.compiler.codegen.methodParameters=do not generate
-org.eclipse.jdt.core.compiler.codegen.targetPlatform=17
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
-org.eclipse.jdt.core.compiler.compliance=17
+org.eclipse.jdt.core.compiler.compliance=1.8
 org.eclipse.jdt.core.compiler.debug.lineNumber=generate
 org.eclipse.jdt.core.compiler.debug.localVariable=generate
 org.eclipse.jdt.core.compiler.debug.sourceFile=generate
@@ -12,5 +12,6 @@ org.eclipse.jdt.core.compiler.problem.enablePreviewFeatures=disabled
 org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
 org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
 org.eclipse.jdt.core.compiler.problem.reportPreviewFeatures=warning
-org.eclipse.jdt.core.compiler.release=enabled
-org.eclipse.jdt.core.compiler.source=17
+org.eclipse.jdt.core.compiler.processAnnotations=disabled
+org.eclipse.jdt.core.compiler.release=disabled
+org.eclipse.jdt.core.compiler.source=1.8
diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/framework/router/Router.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/framework/router/Router.java
@@ -223,11 +223,11 @@ public final Response receive(Packet packet) {
 	 * All {@link Method} objects must contain a single parameter, a {@link Packet} object,
 	 * and return a {@link Response} object.
 	 * 
-	 * @param subtag Subtag of the process to handle the incoming {@link Packet} object.
+	 * @param sub_tag Subtag of the process to handle the incoming {@link Packet} object.
 	 * @param method {@link Method} to pass the {@link Packet} object to.
 	 */
-	public final void addProcess(String subtag, Method method) {
-		processes.put(subtag, method);
+	public final void addProcess(String sub_tag, Method method) {
+		processes.put(sub_tag, method);
 	}
 
 	/**

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
@@ -49,32 +49,45 @@ public void run() {
 					while(true) {
 						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", "Packet processed from REST API does not contain a TAG or SUB_TAG. Review REST API endpoint code.")
-									.toString());
-						}
+						Thread thread = new Thread() {
+							public void run() {
+								try {
+									// validate length is greater than 2
+									if(input.length <= 2) {
+										out.writeUTF(new JSONObject()
+												.put("response", "502")
+												.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];
+
+									// 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());
+									out.flush();
+
+								} catch(SocketException e) {
+									System.err.println("Rest Application has unexpectedly closed.");
+									System.exit(1);
+								} catch (IOException e) {
+									e.printStackTrace();
+									System.exit(1);
+								}
+							}
+						};
 
-						// extract non-essential data
-						String[] data = Arrays.copyOfRange(input, 2, input.length);
-						String tag = input[0];
-						String sub_tag = input[1];
-
-						// 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());
+						thread.start();
 					}
-				} catch(SocketException e) {
-					System.err.println("Rest Application has unexpectedly closed.");
-					System.exit(1);
-				} catch (IOException e) {
+				} catch(Exception e) {
 					e.printStackTrace();
 					System.exit(1);
 				}

diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/destinations/SocketDestination.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/destinations/SocketDestination.java
@@ -21,6 +21,7 @@ public final synchronized boolean send(Packet packet) {
 		try {
 			out.write(packet.getData("data").getBytes());
 			out.write(10);
+			out.flush();
 		} catch (JSONException | IOException e) {
 			return false;
 		}

diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/handler/OutputManager.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/handler/OutputManager.java
@@ -111,7 +111,7 @@ public Object[] producerListen() {
 		return new Object[] {true, ""};
 	}
 
-	protected final Response send(String tag, String sub_tag, String... data) {
+	public final Response send(String tag, String sub_tag, String... data) {
 		return handler.send(tag, sub_tag, data);
 	}
 

diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/producers/SocketProducer.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/producers/SocketProducer.java
@@ -1,5 +1,6 @@
 package org.out.producers;
 
+import org.framework.router.Response;
 import org.out.destinations.SocketDestination;
 import org.out.handler.OutputManager;
 import org.out.handler.OutputProducer;
@@ -9,6 +10,7 @@
 public class SocketProducer extends OutputProducer {
 
 	private Thread listener;
+	public final SocketProducer producer = this;
 
 	public SocketProducer(OutputManager manager) {
 		super(manager);
@@ -24,7 +26,7 @@ protected boolean init() {
 		listener = new Thread() {
 			public void run() {
 				while(true) {
-					String key = SocketManager.accept(Integer.parseInt(Config.getProperty("stream", "output.socket.port")));
+					String key = SocketManager.accept(Integer.parseInt(Config.getProperty("stream", "output.socket.port")), producer);
 					if(key == null) {
 						System.err.println("SocketProducer: Could not create connection to socket port.");
 						System.exit(1);
@@ -55,4 +57,8 @@ protected boolean kill() {
 
 		return true;
 	}
+
+	public Response send(String tag, String sub_tag, String... data) {
+		return manager.send(tag, sub_tag, data);
+	}
 }
diff --git a/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/socket/SocketManager.java b/DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/socket/SocketManager.java
@@ -5,11 +5,16 @@
 import java.io.IOException;
 import java.net.ServerSocket;
 import java.net.Socket;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.UUID;
 
 import org.core.logger.Logger;
+import org.framework.router.Response;
+import org.json.JSONObject;
+import org.out.producers.SocketProducer;
+import org.properties.Config;
 
 public class SocketManager {
 
@@ -38,8 +43,69 @@ public synchronized static boolean exists(String key) {
 		return connections.containsKey(key);
 	}
 
+	public synchronized static void createThread(String key, SocketProducer producer) {
+		Thread thread = new Thread() {
+			public void run() {
+				// perform logger verifications
+				Logger.log(String.format("Starting thread for Socket with key <%s>", key));
+				if(!inflow.containsKey(key)) {
+					Logger.terminate(String.format("Key <%s> not found within inflow thread configuration, manual review recommended.", key));
+					return;
+				}
+
+				if(!outflow.containsKey(key)) {
+					Logger.terminate(String.format("Key <%s> not found within outflow thread configuration, manual review recommended.", key));
+					return;
+				}
+
+				// retrieve inflow stream and listen
+				DataInputStream in = inflow.get(key);
+				DataOutputStream out = outflow.get(key);
+				String str;
+				while(true)
+				try {
+					str = readLine(in);
+
+					// parse input
+					String[] input = str.split(Config.getProperty("app", "general.transfer.delim"));
+
+					// validate input
+					if(input.length <= 2) {
+						out.writeUTF(new JSONObject()
+								.put("response", "502")
+								.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];
+
+					// execute valid response to engine
+					Response response = producer.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());
+					out.flush();
+					System.out.println("<<<responded>>>");
+
+				} catch(Exception e) {
+					break;
+				}
+
+				Logger.log(String.format("Terminating thread for Socket with key <%s>", key));
+			}
+		};
+
+		thread.start();
+	}
+
 	// used for generic channel accepting
-	public synchronized static String accept(int port) {
+	public synchronized static String accept(int port, SocketProducer producer) {
 		if(!servers.containsKey(port))
 			if(!createServer(port))
 				return null;
@@ -68,6 +134,9 @@ public synchronized static String accept(int port) {
 			if(!synced(key))
 				throw new Exception("Connection inflow and outflow not synchronized");
 
+			// start internal thread for socket information parsing
+			createThread(key, producer);
+
 			return key;
 		} catch(Exception e) {
 			e.printStackTrace();
@@ -132,4 +201,13 @@ public static DataInputStream read(String key) {
 	private static boolean synced(String key) {
 		return connections.containsKey(key) && inflow.containsKey(key) && outflow.containsKey(key);
 	}
+
+	private static final String readLine(DataInputStream in) throws IOException {
+		StringBuilder out = new StringBuilder();
+		char c = 0;
+		while((c = (char)in.read()) != 10)
+			out.append(c);
+
+		return out.toString();
+	}
 }