Add in response handling and output for client processing

DeFi-Data-Engine/DeFi Data Engine/.classpath

@@ -25,11 +25,6 @@
 		</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/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"/>
@@ -38,20 +33,21 @@
 	<classpathentry kind="con" path="org.eclipse.jdt.junit.JUNIT_CONTAINER/4"/>
 	<classpathentry kind="src" path="target/generated-sources/annotations">
 		<attributes>
+			<attribute name="ignore_optional_problems" value="true"/>
 			<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="ignore_optional_problems" value="true"/>
+			<attribute name="test" value="true"/>
 			<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="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
 	<classpathentry kind="output" path="target/classes"/>
 </classpath>

DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/out/socket/SocketManager.java

@@ -62,38 +62,57 @@ public void run() {
 				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());
+				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];
+
+						// retrieve destination
+						String destination = "";
+						for(int i = 0; i < data.length; i++) {
+							if(data[i].equals("destination") && data.length - 1 != i)
+								destination = data[i + 1];
+						}
+
+						// if no destination found then continue
+						if(destination.equals(""))
+							continue;
+
+						// execute valid response to engine
+						Response response = producer.send(tag, sub_tag, data);
+
+						// send response signifier
+						producer.send("OUT", "EDAT", 
+								"data", "<<<response>>>",
+								"destination", destination);
+						// send response details
+						producer.send("OUT", "EDAT", 
+								"data", new JSONObject()
+								.put("response", "200")
+								.put("code", response.code())
+								.put("message", response.message())
+								.put("data", response.data())
+								.toString(), 
+								"destination", destination);
+
+					} catch(Exception e) {
+						break;
 					}
-
-					// 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();
-
-				} catch(Exception e) {
-					break;
 				}
 
 				Logger.log(String.format("Terminating thread for Socket with key <%s>", key));

DeFi-Data-Engine/DeFi Data Engine/src/main/java/org/stream/registry/StreamRegistryController.java

@@ -141,7 +141,7 @@ else if(dated) {
 				}
 
 				// send end response
-				return send("SRC", "EDAT", "data", "<<<end>>>", "destination", packet.getData("destination"));
+				return ResponseFactory.response200();
 
 			} catch(Exception e) {
 				return ResponseFactory.response503(Config.getProperty("app", "general.data.dateformat"), packet.getData("start_date"), packet.getData("end_date"));