Skip to content

Commit

Permalink
add properties and url_path to LSH collection name
Browse files Browse the repository at this point in the history
  • Loading branch information
Conor Flynn committed Mar 21, 2023
1 parent 5101e0c commit 9095e63
Show file tree
Hide file tree
Showing 7 changed files with 82 additions and 318 deletions.
180 changes: 0 additions & 180 deletions ...ine/DeFi Data Engine/src/main/java/org/stream/external/handler/ExternalStreamHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,184 +56,4 @@ public Response processRQST(Packet packet) {
// return valid
return ResponseFactory.response200();
}

// // source: source of data
// public Response processEXSR(Packet packet) {
// String validate;
// if((validate = packet.validate("source")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// return ResponseFactory.response200(String.format("%s", manager.containsTemplate(packet.getData("source"))));
// }
//
// // key: key of the given stream
// public Response processEXST(Packet packet) {
// String validate;
// if((validate = packet.validate("key")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// return ResponseFactory.response200(String.format("%s", manager.containsStream(packet.getData("key"))));
// }
//
// // source: source of the given stream
// // auth_data[]: all keys associated with data
// public Response processINIT(Packet packet) {
// String validate;
// if((validate = packet.validate("source")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// String source = packet.getData("source");
//
// // validate data
// String tempHash = manager.getHash(source, packet.getData());
// if(tempHash == null)
// return ResponseFactory.response430(packet.getData());
// if(manager.containsStream(tempHash))
// return ResponseFactory.response220(tempHash);
//
// // attempt to add stream
// Object[] output = manager.addStream(source, packet.getData());
// boolean success = (Boolean) output[0];
// String hash = (String) output[1];
//
// if(!success)
// return ResponseFactory.response420(source);
//
// boolean authorized = manager.authorizeStream(hash);
//
// // if successful authorize
// if(authorized)
// return ResponseFactory.response200(String.format("%s", hash));
//
// manager.removeStream(hash);
// return ResponseFactory.response422(source);
// }
//
// // key: stream key
// public Response processIATH(Packet packet) {
// String validate;
// if((validate = packet.validate("key")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// if(!manager.containsStream(packet.getData("key")))
// return ResponseFactory.response421(packet.getData("key"));
//
// return ResponseFactory.response200(String.format("%s", manager.isStreamAuthorized(packet.getData("key"))));
// }
//
// public Response processIATV(Packet packet) {
// String validate;
// if((validate = packet.validate("key")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// String key = packet.getData("key");
// if(!manager.containsStream(key))
// return ResponseFactory.response421(key);
//
// return ResponseFactory.response200(String.format("%s", manager.isStreamActive(key)));
// }
//
// public Response processEXEC(Packet packet) {
// String validate;
// if((validate = packet.validate("key")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// String key = packet.getData("key");
// if(!manager.containsStream(key))
// return ResponseFactory.response421(key);
//
// if(!manager.isStreamReady(key))
// return ResponseFactory.response423(key);
//
// if(manager.isStreamActive(key))
// return ResponseFactory.response424(key);
//
// return ResponseFactory.response200(String.format("%s", manager.executeStream(key)));
// }
//
// public Response processKILL(Packet packet) {
// String validate;
// if((validate = packet.validate("key")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// String key = packet.getData("key");
// if(!manager.containsStream(key))
// return ResponseFactory.response421(key);
//
// if(!manager.isStreamActive(key))
// return ResponseFactory.response425(key);
//
// return ResponseFactory.response200(String.format("%s", manager.killStream(key)));
// }
//
// public Response processSUBS(Packet packet) {
// String validate;
// if((validate = packet.validate("key", "subscription")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// String key = packet.getData("key");
// String subscription = packet.getData("subscription");
//
// if(!manager.containsStream(key))
// return ResponseFactory.response421(key);
//
// if(!manager.isStreamActive(key))
// return ResponseFactory.response425(key);
//
// if(!manager.containsSubscriptionType(key, subscription))
// return ResponseFactory.response426(key, subscription);
//
// Object[] response = manager.subscribe(key, subscription);
//
// if((Boolean)response[0])
// return ResponseFactory.response200(String.format("%s", "true"));
//
// return ResponseFactory.response427(key, response[0].toString(), response[1].toString());
// }
//
// public Response processRQST(Packet packet) {
// String validate;
// if((validate = packet.validate("key", "request", "query", "destination")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// String key = packet.getData("key");
// String request = packet.getData("request");
//
// if(!manager.containsStream(key))
// return ResponseFactory.response421(key);
//
// if(!manager.isStreamReady(key))
// return ResponseFactory.response423(key);
//
// if(!manager.containsRequestType(key, request))
// return ResponseFactory.response428(key, request);
//
// Object[] response = manager.request(key, packet.getData());
//
// if(response == null)
// return ResponseFactory.response501("Response from Manager.request should never be null. "
// + "Improper implementation of ExternalStreamConnection.");
//
// if((Boolean)response[0])
// return ResponseFactory.response200(String.format("%s", (String)response[1]));
//
// return ResponseFactory.response429(key, request, response[1].toString());
// }
//
// public Response processTYPE(Packet packet) {
// String validate;
// if((validate = packet.validate("key")) != null)
// return ResponseFactory.response500("ExternalStreamHandler", validate);
//
// String key = packet.getData("key");
// if(!manager.containsStream(key))
// return ResponseFactory.response421(key);
//
// String uuid = manager.getStreamType(key);
// if(uuid == null)
// return ResponseFactory.response501("Stream was removed in different thread mid observation.");
//
// return ResponseFactory.response200(uuid);
//
// }
}
10 changes: 7 additions & 3 deletions ...ine/DeFi Data Engine/src/main/java/org/stream/external/handler/ExternalStreamManager.java
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package org.stream.external.handler;

import java.util.HashMap;
import java.util.TreeMap;

import org.stream.external.requests.ExternalRequestFramework;
import org.stream.external.requests.ExternalRequestManager;
Expand Down Expand Up @@ -47,13 +48,16 @@ protected Object[] request(String type, HashMap<String, String> data, String sta

// retrieve properties
HashMap<String, String> properties = new HashMap<String, String>();
TreeMap<String, String> user_properties = new TreeMap<String, String>();
if(data.containsKey("properties")) {
String[] raw_properties = data.get("properties").split(",");
if(raw_properties.length % 2 != 0)
return new Object[] {false, String.format("Properties must be in <key, value> pairs.")};

for(int i = 0; i < raw_properties.length; i+=2)
for(int i = 0; i < raw_properties.length; i+=2) {
properties.put(raw_properties[i], raw_properties[i + 1]);
user_properties.put(raw_properties[i], raw_properties[i + 1]);
}
}

// retrieve headers
Expand All @@ -73,9 +77,9 @@ protected Object[] request(String type, HashMap<String, String> data, String sta
// submit response
String response;
if(startDate == null || endDate == null)
response = request.request(url_path, properties, headers);
response = request.request(url_path, properties, headers, user_properties);
else
response = request.request(url_path, properties, headers, startDate, endDate);
response = request.request(url_path, properties, headers, startDate, endDate, user_properties);

if(response != null)
return new Object[] {false, response};
Expand Down
32 changes: 27 additions & 5 deletions ...DeFi Data Engine/src/main/java/org/stream/external/requests/ExternalRequestFramework.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,10 @@
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.TreeMap;
import java.util.stream.Stream;

import org.json.JSONArray;
Expand Down Expand Up @@ -65,7 +67,6 @@ public ExternalRequestFramework(ExternalStreamManager manager, String name, Stri
HashMap<String, String> tags, String[] recursive_location, String recursive_replacement, String[] path,
boolean is_dated, String date_location, String date_start_var, String date_end_var, String date_format) {
this.manager = manager;
this.collection = name;
this.name = name;
this.url = url;
this.url_path = url_path;
Expand Down Expand Up @@ -178,7 +179,6 @@ protected final Request getRequest(String url, HashMap<String, String> propertie
return builder.build();
}


public final synchronized String request(String[] url_path, HashMap<String, String> properties, HashMap<String, String> headers) {
// validate that url_path is correctly formatted
if(url_path.length % 2 != 0) {
Expand Down Expand Up @@ -208,8 +208,21 @@ public final synchronized String request(String[] url_path, HashMap<String, Stri
return process(url_builder.toString(), properties, headers);
}

public final synchronized String request(String[] url_path, HashMap<String, String> properties, HashMap<String, String> headers, TreeMap<String, String> user_properties) {
// update collection to be dated
StringBuilder sb = new StringBuilder();
sb.append(name);
if(!user_properties.isEmpty())
sb.append("-").append(user_properties.toString());
if(url_path.length != 0)
sb.append("-").append(Arrays.toString(url_path));
this.collection = sb.toString();
collection = sb.toString();
return request(url_path, properties, headers);
}

public final synchronized String request(String[] url_path, HashMap<String, String> properties, HashMap<String, String> headers,
String startDate, String endDate) {
String startDate, String endDate, TreeMap<String, String> user_properties) {

// validate that the request can be dated
if(!is_dated) {
Expand Down Expand Up @@ -278,9 +291,18 @@ public final synchronized String request(String[] url_path, HashMap<String, Stri

// submit request with updated properties
// update collection to be dated
collection = name + "-" + date.format(date_format).toString();
String temp_collection = collection;
StringBuilder sb = new StringBuilder();
sb.append(name);
sb.append("-").append(date.format(date_format).toString());
if(!user_properties.isEmpty())
sb.append("-").append(user_properties.toString());
if(url_path.length != 0)
sb.append("-").append(Arrays.toString(url_path));
this.collection = sb.toString();
collection = sb.toString();
String request = request(url_path, properties, headers);
collection = name;
collection = temp_collection;
if(request != null)
return request;
}
Expand Down
3 changes: 2 additions & 1 deletion ...e/DeFi Data Engine/src/main/java/org/stream/external/requests/ExternalRequestManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import java.util.Properties;
import java.util.Set;

import org.core.logger.Logger;
import org.reflections.Reflections;
import org.reflections.util.ConfigurationBuilder;
import org.stream.external.handler.ExternalStreamManager;
Expand Down Expand Up @@ -213,7 +214,7 @@ public void initialize(ExternalStreamManager manager) throws InstantiationExcept
manager, name, url, url_path, properties, headers, tags, recursive_location, recursive_replacement, path,
is_dated, date_location, date_start_var, date_end_var, date_format));

System.out.println(String.format("Successfully added type [%s]", name));
Logger.log(String.format("[ExternalRequestManager] Successfully added Request Framework type [%s]", name));
}
}

Expand Down
4 changes: 0 additions & 4 deletions ...ta-Engine/DeFi Data Engine/src/main/java/org/stream/local/handler/LocalStreamHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,12 +68,8 @@ public Response processRQST(Packet packet) {

String request = packet.getData("query");

//String[] query = packet.getData("query").split(Config.getProperty("app", "general.data.delim"));
String[] query = new String[] {"get_all", request};

// if(query == null)
// return ResponseFactory.response501("Query array was null when attempting to process.");

if(request == null || request.isEmpty())
return ResponseFactory.response501("Request was null when attempting to process.");

Expand Down
47 changes: 43 additions & 4 deletions ...a-Engine/DeFi Data Engine/src/main/java/org/stream/registry/StreamRegistryController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,10 @@

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.TreeMap;
import java.util.stream.Collectors;

import org.framework.router.Packet;
Expand Down Expand Up @@ -54,7 +56,25 @@ public Response processRQST(Packet packet) {

// not dated
if(!dated) {
data.put("query", String.format("get_all, %s", packet.getData("type")));
// format collection name
StringBuilder sb = new StringBuilder();
sb.append(data.get("type"));
if(packet.containsKey("properties")) {
TreeMap<String, String> ordered_properties = new TreeMap<String, String>();
String[] properties = data.get("properties").split(",");
if(properties.length % 2 != 0)
return ResponseFactory.response407("SRC", "RQST", "<null>", data.toString());
for(int i = 0; i < properties.length; i+=2)
ordered_properties.put(properties[i], properties[i + 1]);
sb.append("-").append(ordered_properties.toString());
}

if(packet.containsKey("url_path")) {
String[] url_path = data.get("url_path").split(",");
sb.append("-").append(Arrays.toString(url_path));
}
data.put("query", String.format("%s", sb));

Response lsh_response = send("LSH", "RQST", data);
// if data does not exist send request to external stream handler
if(lsh_response.code() == 446)
Expand All @@ -74,11 +94,30 @@ else if(dated) {

Response lsh_response, esh_response;
for(LocalDate date : dates) {
// perform requests

// initial request
data.put("date", date.format(formatter));
data.put("query", String.format("%s-%s", data.get("type"), data.get("date")));

// format collection name
StringBuilder sb = new StringBuilder();
sb.append(data.get("type"));
sb.append("-").append(data.get("date"));
if(packet.containsKey("properties")) {
TreeMap<String, String> ordered_properties = new TreeMap<String, String>();
String[] properties = data.get("properties").split(",");
if(properties.length % 2 != 0)
return ResponseFactory.response407("SRC", "RQST", "<null>", data.toString());
for(int i = 0; i < properties.length; i+=2)
ordered_properties.put(properties[i], properties[i + 1]);
sb.append("-").append(ordered_properties.toString());
}

if(packet.containsKey("url_path")) {
String[] url_path = data.get("url_path").split(",");
sb.append("-").append(Arrays.toString(url_path));
}
data.put("query", sb.toString());

// initiate request
lsh_response = send("LSH", "RQST", data);
if(lsh_response.code() == 200)
continue;
Expand Down
Loading

0 comments on commit 9095e63

Please sign in to comment.