ruanwenjun commented on a change in pull request #434:
URL:
https://github.com/apache/incubator-eventmesh/pull/434#discussion_r669568396
##########
File path: eventmesh-runtime/build.gradle
##########
@@ -29,8 +29,12 @@ List open_message = [
"io.openmessaging:openmessaging-api:2.2.1-pubsub"
]
+List h2_database = [
Review comment:
This dependency should move to `eventmesh-store-h2`.
##########
File path:
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/CreateSubjectHandler.java
##########
@@ -0,0 +1,94 @@
+package org.apache.eventmesh.runtime.admin.handler;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ServiceLoader;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.runtime.util.NetUtils;
+import org.apache.eventmesh.store.api.openschema.common.ServiceException;
+import org.apache.eventmesh.store.api.openschema.request.SubjectCreateRequest;
+import org.apache.eventmesh.store.api.openschema.response.SubjectResponse;
+import org.apache.eventmesh.store.api.openschema.service.SchemaService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+public class CreateSubjectHandler implements HttpHandler {
+
+ private Logger logger =
LoggerFactory.getLogger(CreateSubjectHandler.class);
+
+ private static ObjectMapper jsonMapper;
Review comment:
It might be better use `JsonUtils` to serialize or deserialize JSON.
##########
File path:
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/controller/ClientManageController.java
##########
@@ -57,6 +60,10 @@ public void start() throws IOException {
server.createContext("/clientManage/redirectClientByPath", new
RedirectClientByPathHandler(eventMeshTCPServer));
server.createContext("/clientManage/redirectClientByIpPort", new
RedirectClientByIpPortHandler(eventMeshTCPServer));
server.createContext("/clientManage/showListenClientByTopic", new
ShowListenClientByTopicHandler(eventMeshTCPServer));
+
+ server.createContext("/schemaregistry/createSubject", new
CreateSubjectHandler());
Review comment:
`schemaregistry` should rename to `schemaRegistry`
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/common/ServiceError.java
##########
@@ -0,0 +1,72 @@
+package org.apache.eventmesh.store.api.openschema.common;
+
+public enum ServiceError {
+ /**
+ *
+ */
+ ERR_SCHEMA_EMPTY("442", 442, "schema info cannot be empty"),
+
+ /**
+ *
+ */
+ ERR_PERM_NO_AUTH("401", 401, "40101 - no permission to access"),
+
+ /**
+ *
+ */
+ ERR_SCHEMA_INVALID("404", 404, "40401 - schema information does not exist
in our record"),
+
+ /**
+ *
+ */
+ ERR_SCHEMA_VERSION_INVALID("404", 404, "40402 - schema version does not
exist in our record"),
+
+ /**
+ *
+ */
+ ERR_SUBJECT_INVALID("404", 404, "40401 - schema version does not exist in
this subject"),
+
+ /**
+ *
+ */
+ ERR_SCHEMA_FORMAT_INVALID("422", 422, "40401 - schema format is invalid"),
+
+ /**
+ *
+ */
+ ERR_SCHEMA_VERSION_FORMAT_INVALID("422", 422, "40402 - schema version
format is invalid"),
+
+ /**
+ *
+ */
+ ERR_SERVER_ERROR("500", 500, "I50001 - Internal Server Error"),
+
+ /**
+ * resource level error definitions
+ */
+ ERR_RES_NOT_FOUND("500", 500, "50002 - Request Timeout");
+
+ private String status;
Review comment:
The `status` seems to be redundant.
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/request/BaseRequest.java
##########
@@ -0,0 +1,19 @@
+package org.apache.eventmesh.store.api.openschema.request;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class BaseRequest {
+
+ private String id;
+
+ @JsonProperty("id")
Review comment:
`@JsonProperty("id")` can remove
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/common/ServiceException.java
##########
@@ -0,0 +1,50 @@
+package org.apache.eventmesh.store.api.openschema.common;
+
+
+import org.apache.eventmesh.store.api.openschema.response.ErrorResponse;
+
+import com.alibaba.fastjson.JSONObject;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class ServiceException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ private int errorCode;
+ private String errorMessage;
+
+ //public ErrorResponse(@JsonProperty("error_code") int errorCode,
+ // @JsonProperty("error_message") String
errorMessage) {
+ public ServiceException(ServiceError serviceError) {
+ super(serviceError.getMessage());
+ this.errorCode = serviceError.getErrorCode();
+ this.errorMessage = serviceError.getMessage();
+ }
+
+ @JsonProperty("error_code")
+ public int getErrorCode() {
+ return errorCode;
+ }
+
+ @JsonProperty("error_code")
+ public void setErrorCode(int error_code) {
Review comment:
Use Camel-Case, rename to errorCode
##########
File path:
eventmesh-store-h2/src/main/java/org/apache/eventmesh/store/h2/schema/util/DBDataSource.java
##########
@@ -0,0 +1,42 @@
+package org.apache.eventmesh.store.h2.schema.util;
+
+import org.apache.eventmesh.store.h2.schema.configuration.DBConfiguration;
+import org.apache.commons.dbcp2.BasicDataSource;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+public class DBDataSource {
+ private static BasicDataSource ds = null;
+
+ private DBDataSource() {
+
+ }
+
+ public static synchronized DBDataSource createDataSource(DBConfiguration
dbConfig) {
Review comment:
The `DBDataSource` is singleton?
##########
File path:
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/CreateSubjectHandler.java
##########
@@ -0,0 +1,94 @@
+package org.apache.eventmesh.runtime.admin.handler;
Review comment:
Add license header here.
##########
File path:
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/CreateSubjectHandler.java
##########
@@ -0,0 +1,94 @@
+package org.apache.eventmesh.runtime.admin.handler;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ServiceLoader;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.runtime.util.NetUtils;
+import org.apache.eventmesh.store.api.openschema.common.ServiceException;
+import org.apache.eventmesh.store.api.openschema.request.SubjectCreateRequest;
+import org.apache.eventmesh.store.api.openschema.response.SubjectResponse;
+import org.apache.eventmesh.store.api.openschema.service.SchemaService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+public class CreateSubjectHandler implements HttpHandler {
+
+ private Logger logger =
LoggerFactory.getLogger(CreateSubjectHandler.class);
+
+ private static ObjectMapper jsonMapper;
+
+ static {
+ jsonMapper = new ObjectMapper();
+ jsonMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+ jsonMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+ jsonMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
+ }
+
+ @Override
+ public void handle(HttpExchange httpExchange) throws IOException {
+ String result = "";
+ OutputStream out = httpExchange.getResponseBody();
+ try {
+ String params = NetUtils.parsePostBody(httpExchange);
+ SubjectCreateRequest subjectCreateRequest =
jsonMapper.readValue(params, SubjectCreateRequest.class);
+ String subject = subjectCreateRequest.getSubject();
+
+ if (StringUtils.isBlank(subject)) {
+ result = "Create subject failed. Parameter subject not found.";
+ logger.error(result);
+ out.write(result.getBytes());
+ return;
+ }
+ SchemaService schemaService = getSchemaService();
+ SubjectResponse subjectResponse =
schemaService.createSubject(subjectCreateRequest);
+ if (subjectResponse != null) {
+ logger.info("createTopic subject: {}", subject);
+ httpExchange.getResponseHeaders().add("Content-Type",
"appication/json");
+ httpExchange.sendResponseHeaders(200, 0);
+ result = jsonMapper.writeValueAsString(subjectResponse);
+ logger.info(result);
+ out.write(result.getBytes());
+ return;
+ } else {
+ httpExchange.sendResponseHeaders(500, 0);
+ result = String.format("create subject failed! Server side
error");
+ logger.error(result);
+ out.write(result.getBytes());
+ return;
+ }
+ } catch (ServiceException e) {
+ httpExchange.getResponseHeaders().add("Content-Type",
"appication/json");
+ httpExchange.sendResponseHeaders(500, 0);
+ result = jsonMapper.writeValueAsString(e.getErrorResponse());
+ logger.error(result);
+ out.write(result.getBytes());
+ return;
+ } finally {
+ if (out != null) {
+ try {
+ out.close();
+ } catch (IOException e) {
+ logger.warn("out close failed...", e);
+ }
+ }
+ }
+ }
+
+ private SchemaService getSchemaService() {
Review comment:
Use EventMeshExtensionFactory to load plugin, by the way, there may
cause NPE.
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/common/CompatibilityLevel.java
##########
@@ -0,0 +1,42 @@
+package org.apache.eventmesh.store.api.openschema.common;
+
+public enum CompatibilityLevel {
+ NONE,
+ BACKWARD,
+ BACKWARD_TRANSITIVE,
+ FORWARD,
+ FORWARD_TRANSITIVE,
+ FULL,
+ FULL_TRANSITIVE;
+
+ public final String name;
Review comment:
This attribute seems to be redundant if the name just represent name().
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/request/ConfigUpdateRequest.java
##########
@@ -0,0 +1,26 @@
+package org.apache.eventmesh.store.api.openschema.request;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ConfigUpdateRequest {
+
+ private String compatibilityLevel;
+
+ /* @ApiModelProperty(value = "Compatability Level",
Review comment:
Remove unused code.
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/request/SchemaCreateRequest.java
##########
@@ -0,0 +1,124 @@
+package org.apache.eventmesh.store.api.openschema.request;
+
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class SchemaCreateRequest extends BaseRequest{
+
+ private String name;
+ private String comment;
+ private String serialization;
+ private String schemaType;
+ private String schemaDefinition;
+ private String validator;
+ private String version;
+
+ @JsonCreator
+ public SchemaCreateRequest(@JsonProperty("name") String name,
Review comment:
Remove `@JsonCreator` and `@JsonProperty`
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/request/CompatibilityCheckRequest.java
##########
@@ -0,0 +1,23 @@
+package org.apache.eventmesh.store.api.openschema.request;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CompatibilityCheckRequest {
+
+ private String schema;
+
+ @JsonProperty("schema")
Review comment:
`@JsonProperty("id")` can remove
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/request/ConfigUpdateRequest.java
##########
@@ -0,0 +1,26 @@
+package org.apache.eventmesh.store.api.openschema.request;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ConfigUpdateRequest {
+
+ private String compatibilityLevel;
+
+ /* @ApiModelProperty(value = "Compatability Level",
+ allowableValues = "BACKWARD, BACKWARD_TRANSITIVE, FORWARD,
FORWARD_TRANSITIVE, FULL, "
+ + "FULL_TRANSITIVE, NONE")*/
+ @JsonProperty("compatibility")
Review comment:
It might make confusion why the returned attribute don't match.
##########
File path: eventmesh-store-api/build.gradle
##########
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
Review comment:
Keep only one empty line.
##########
File path: eventmesh-store-api/gradle.properties
##########
@@ -0,0 +1,21 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+group=org.apache.eventmesh
+version=1.2.0-SNAPSHOT
+jdk=1.7
Review comment:
Use jdk 1.8
##########
File path:
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/CreateSubjectHandler.java
##########
@@ -0,0 +1,94 @@
+package org.apache.eventmesh.runtime.admin.handler;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ServiceLoader;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.runtime.util.NetUtils;
+import org.apache.eventmesh.store.api.openschema.common.ServiceException;
+import org.apache.eventmesh.store.api.openschema.request.SubjectCreateRequest;
+import org.apache.eventmesh.store.api.openschema.response.SubjectResponse;
+import org.apache.eventmesh.store.api.openschema.service.SchemaService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+public class CreateSubjectHandler implements HttpHandler {
+
+ private Logger logger =
LoggerFactory.getLogger(CreateSubjectHandler.class);
+
+ private static ObjectMapper jsonMapper;
+
+ static {
+ jsonMapper = new ObjectMapper();
+ jsonMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+ jsonMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+ jsonMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
+ }
+
+ @Override
+ public void handle(HttpExchange httpExchange) throws IOException {
+ String result = "";
+ OutputStream out = httpExchange.getResponseBody();
+ try {
+ String params = NetUtils.parsePostBody(httpExchange);
+ SubjectCreateRequest subjectCreateRequest =
jsonMapper.readValue(params, SubjectCreateRequest.class);
+ String subject = subjectCreateRequest.getSubject();
+
+ if (StringUtils.isBlank(subject)) {
+ result = "Create subject failed. Parameter subject not found.";
+ logger.error(result);
+ out.write(result.getBytes());
+ return;
+ }
+ SchemaService schemaService = getSchemaService();
+ SubjectResponse subjectResponse =
schemaService.createSubject(subjectCreateRequest);
+ if (subjectResponse != null) {
+ logger.info("createTopic subject: {}", subject);
+ httpExchange.getResponseHeaders().add("Content-Type",
"appication/json");
+ httpExchange.sendResponseHeaders(200, 0);
+ result = jsonMapper.writeValueAsString(subjectResponse);
+ logger.info(result);
+ out.write(result.getBytes());
+ return;
+ } else {
+ httpExchange.sendResponseHeaders(500, 0);
+ result = String.format("create subject failed! Server side
error");
+ logger.error(result);
+ out.write(result.getBytes());
+ return;
+ }
+ } catch (ServiceException e) {
+ httpExchange.getResponseHeaders().add("Content-Type",
"appication/json");
+ httpExchange.sendResponseHeaders(500, 0);
+ result = jsonMapper.writeValueAsString(e.getErrorResponse());
+ logger.error(result);
+ out.write(result.getBytes());
+ return;
+ } finally {
Review comment:
Use try with resource instead of try finally
##########
File path:
eventmesh-store-api/src/main/java/org/apache/eventmesh/store/api/openschema/request/BaseRequest.java
##########
@@ -0,0 +1,19 @@
+package org.apache.eventmesh.store.api.openschema.request;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class BaseRequest {
Review comment:
The request might implement `Serializable`
##########
File path:
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
##########
@@ -112,5 +112,7 @@
public static final String LEAVE_TIME = "LEAVE_TIME";
//leaveBrokerTime
public static final String ARRIVE_TIME = "ARRIVE_TIME";
public static final String STORE_TIME = "STORE_TIME";
-
+
+ public static final String MANAGE_SCHEMA_SUBJECT = "subject";
Review comment:
The meaning of `subject` seems unclear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
