0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[#207]: PR feedback

Browse source
This commit is contained in:
Ilnur Farukhshin 2021-07-26 13:52:38 +03:00
parent 232a17b80d
commit 64e2caf55c
8 changed files with 44 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/BaseStrategy.java
View file

@ -16,9 +16,9 @@ import java.util.stream.Stream;
import java.util.stream.StreamSupport;
public abstract class BaseStrategy {
protected static final String ksqlRequestPath = "/ksql";
protected static final String queryRequestPath = "/query";
private static final String mappingExceptionMessage = "KSQL DB response mapping error";
protected static final String KSQL_REQUEST_PATH = "/ksql";
protected static final String QUERY_REQUEST_PATH = "/query";
private static final String MAPPING_EXCEPTION_ERROR = "KSQL DB response mapping error";
protected String host = null;
protected KsqlCommand ksqlCommand = null;
@ -47,36 +47,39 @@ public abstract class BaseStrategy {
return this;
}
protected KsqlCommandResponse serializeTableResponse(JsonNode response, String path) {
if (response.isArray() && response.size() > 0) {
KsqlCommandResponse commandResponse = new KsqlCommandResponse();
JsonNode first = response.get(0);
JsonNode items = first.path(path);
Table table = items.isArray() ? getTableFromArray(items) : getTableFromObject(items);
return commandResponse.data(table);
}
throw new UnprocessableEntityException(mappingExceptionMessage);
protected String getRequestPath() {
return BaseStrategy.KSQL_REQUEST_PATH;
}
protected KsqlCommandResponse serializeMessageResponse(JsonNode response, String path) {
if (response.isArray() && response.size() > 0) {
KsqlCommandResponse commandResponse = new KsqlCommandResponse();
JsonNode first = response.get(0);
JsonNode item = first.path(path);
return commandResponse.message(getMessageFromObject(item));
}
throw new UnprocessableEntityException(mappingExceptionMessage);
protected KsqlCommandResponse serializeTableResponse(JsonNode response, String key) {
JsonNode item = getResponseFirstItemValue(response, key);
Table table = item.isArray() ? getTableFromArray(item) : getTableFromObject(item);
return (new KsqlCommandResponse()).data(table);
}
protected KsqlCommandResponse serializeMessageResponse(JsonNode response, String key) {
JsonNode item = getResponseFirstItemValue(response, key);
return (new KsqlCommandResponse()).message(getMessageFromObject(item));
}
protected KsqlCommandResponse serializeQueryResponse(JsonNode response) {
if (response.isArray() && response.size() > 0) {
KsqlCommandResponse commandResponse = new KsqlCommandResponse();
Table table = (new Table())
.headers(getQueryResponseHeader(response))
.rows(getQueryResponseRows(response));
return commandResponse.data(table);
return (new KsqlCommandResponse()).data(table);
}
throw new UnprocessableEntityException(mappingExceptionMessage);
throw new UnprocessableEntityException(MAPPING_EXCEPTION_ERROR);
}
private JsonNode getResponseFirstItemValue(JsonNode response, String key) {
if (response.isArray() && response.size() > 0) {
JsonNode first = response.get(0);
if (first.has(key)) {
return first.path(key);
}
}
throw new UnprocessableEntityException(MAPPING_EXCEPTION_ERROR);
}
private List<String> getQueryResponseHeader(JsonNode response) {
@ -124,7 +127,7 @@ public abstract class BaseStrategy {
if (node.isObject() && node.has("message")) {
return node.get("message").asText();
}
throw new UnprocessableEntityException(mappingExceptionMessage);
throw new UnprocessableEntityException(MAPPING_EXCEPTION_ERROR);
}
private List<List<String>> getTableRows(JsonNode node, List<String> keys) {
@ -140,7 +143,7 @@ public abstract class BaseStrategy {
if (node.isArray() && node.size() > 0) {
return StreamSupport.stream(node.spliterator(), false);
}
throw new UnprocessableEntityException(mappingExceptionMessage);
throw new UnprocessableEntityException(MAPPING_EXCEPTION_ERROR);
}
private List<String> getJsonObjectKeys(JsonNode node) {
@ -149,7 +152,7 @@ public abstract class BaseStrategy {
Spliterators.spliteratorUnknownSize(node.fieldNames(), Spliterator.ORDERED), false
).collect(Collectors.toList());
}
throw new UnprocessableEntityException(mappingExceptionMessage);
throw new UnprocessableEntityException(MAPPING_EXCEPTION_ERROR);
}
private List<String> getJsonObjectValues(JsonNode node) {
@ -159,7 +162,5 @@ public abstract class BaseStrategy {
public abstract KsqlCommandResponse serializeResponse(JsonNode response);
protected abstract String getRequestPath();
protected abstract String getTestRegExp();
}

9
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/CreateStrategy.java
View file

@ -6,16 +6,11 @@ import org.springframework.stereotype.Component;
@Component
public class CreateStrategy extends BaseStrategy {
private static final String responseValueKey = "commandStatus";
private static final String RESPONSE_VALUE_KEY = "commandStatus";
@Override
public KsqlCommandResponse serializeResponse(JsonNode response) {
return serializeMessageResponse(response, responseValueKey);
}
@Override
protected String getRequestPath() {
return BaseStrategy.ksqlRequestPath;
return serializeMessageResponse(response, RESPONSE_VALUE_KEY);
}
@Override

9
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/DescribeStrategy.java
View file

@ -6,16 +6,11 @@ import org.springframework.stereotype.Component;
@Component
public class DescribeStrategy extends BaseStrategy {
private static final String responseValueKey = "sourceDescription";
private static final String RESPONSE_VALUE_KEY = "sourceDescription";
@Override
public KsqlCommandResponse serializeResponse(JsonNode response) {
return serializeTableResponse(response, responseValueKey);
}
@Override
protected String getRequestPath() {
return BaseStrategy.ksqlRequestPath;
return serializeTableResponse(response, RESPONSE_VALUE_KEY);
}
@Override

9
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/DropStrategy.java
View file

@ -6,16 +6,11 @@ import org.springframework.stereotype.Component;
@Component
public class DropStrategy extends BaseStrategy {
private static final String responseValueKey = "commandStatus";
private static final String RESPONSE_VALUE_KEY = "commandStatus";
@Override
public KsqlCommandResponse serializeResponse(JsonNode response) {
return serializeMessageResponse(response, responseValueKey);
}
@Override
protected String getRequestPath() {
return BaseStrategy.ksqlRequestPath;
return serializeMessageResponse(response, RESPONSE_VALUE_KEY);
}
@Override

9
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/ExplainStrategy.java
View file

@ -6,16 +6,11 @@ import org.springframework.stereotype.Component;
@Component
public class ExplainStrategy extends BaseStrategy {
private static final String responseValueKey = "queryDescription";
private static final String RESPONSE_VALUE_KEY = "queryDescription";
@Override
public KsqlCommandResponse serializeResponse(JsonNode response) {
return serializeTableResponse(response, responseValueKey);
}
@Override
protected String getRequestPath() {
return BaseStrategy.ksqlRequestPath;
return serializeTableResponse(response, RESPONSE_VALUE_KEY);
}
@Override

2
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/SelectStrategy.java
View file

@ -14,7 +14,7 @@ public class SelectStrategy extends BaseStrategy {
@Override
protected String getRequestPath() {
return BaseStrategy.queryRequestPath;
return BaseStrategy.QUERY_REQUEST_PATH;
}
@Override

13
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/ShowStrategy.java
View file

@ -8,9 +8,9 @@ import org.springframework.stereotype.Component;
@Component
public class ShowStrategy extends BaseStrategy {
private final List<String> showStatements =
private static final List<String> SHOW_STATEMENTS =
List.of("functions", "topics", "streams", "tables", "queries", "properties");
private final List<String> listStatements =
private static final List<String> LIST_STATEMENTS =
List.of("functions", "topics", "streams", "tables");
private String responseValueKey = "";
@ -21,7 +21,7 @@ public class ShowStrategy extends BaseStrategy {
@Override
public boolean test(String sql) {
Optional<String> statement = showStatements.stream()
Optional<String> statement = SHOW_STATEMENTS.stream()
.filter(s -> testSql(sql, getShowRegExp(s)) || testSql(sql, getListRegExp(s)))
.findFirst();
if (statement.isPresent()) {
@ -31,11 +31,6 @@ public class ShowStrategy extends BaseStrategy {
return false;
}
@Override
protected String getRequestPath() {
return BaseStrategy.ksqlRequestPath;
}
@Override
protected String getTestRegExp() {
return "";
@ -46,7 +41,7 @@ public class ShowStrategy extends BaseStrategy {
}
protected String getListRegExp(String key) {
if (listStatements.contains(key)) {
if (LIST_STATEMENTS.contains(key)) {
return "list " + key + ";";
}
return "";

9
kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/TerminateStrategy.java
View file

@ -6,16 +6,11 @@ import org.springframework.stereotype.Component;
@Component
public class TerminateStrategy extends BaseStrategy {
private static final String responseValueKey = "commandStatus";
private static final String RESPONSE_VALUE_KEY = "commandStatus";
@Override
public KsqlCommandResponse serializeResponse(JsonNode response) {
return serializeMessageResponse(response, responseValueKey);
}
@Override
protected String getRequestPath() {
return BaseStrategy.ksqlRequestPath;
return serializeMessageResponse(response, RESPONSE_VALUE_KEY);
}
@Override