[ISSUE #359] Split handler from controller (#359) (#360)

* [ISSUE #359] Split handler from controller (#359)

* add license header

* add ut

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/EventMeshMsgDownStreamHandler.java

@@ -0,0 +1,166 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.runtime.admin.handler;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class EventMeshMsgDownStreamHandler implements HttpHandler {
+
+    private static final Logger logger = LoggerFactory.getLogger(EventMeshMsgDownStreamHandler.class);
+
+    private final EventMeshTCPServer eventMeshTCPServer;
+
+    public EventMeshMsgDownStreamHandler(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+    }
+
+    @Override
+    public void handle(HttpExchange httpExchange) throws IOException {
+        String result = "false";
+        OutputStream out = httpExchange.getResponseBody();
+        try {
+//                Map<String, Object> queryStringInfo =  parsePostParameters(httpExchange);
+//                String msgStr = (String)queryStringInfo.get("msg");
+//                String groupName = (String)queryStringInfo.get("group");
+//                logger.info("recieve msg from other eventMesh, group:{}, msg:{}", groupName, msgStr);
+//                if (StringUtils.isBlank(msgStr) || StringUtils.isBlank(groupName)) {
+//                    logger.warn("msg or groupName is null");
+//                    httpExchange.sendResponseHeaders(200, 0);
+//                    out.write(result.getBytes());
+//                    return;
+//                }
+//                MessageExt messageExt = JSON.parseObject(msgStr, MessageExt.class);
+//                String topic = messageExt.getTopic();
+//
+//                if (!EventMeshUtil.isValidRMBTopic(topic)) {
+//                    logger.warn("msg topic is illegal");
+//                    httpExchange.sendResponseHeaders(200, 0);
+//                    out.write(result.getBytes());
+//                    return;
+//                }
+//
+//                DownstreamDispatchStrategy downstreamDispatchStrategy = eventMeshTCPServer.getClientSessionGroupMapping().getClientGroupWrapper(groupName).getDownstreamDispatchStrategy();
+//                Set<Session> groupConsumerSessions = eventMeshTCPServer.getClientSessionGroupMapping().getClientGroupWrapper(groupName).getGroupConsumerSessions();
+//                Session session = downstreamDispatchStrategy.select(groupName, topic, groupConsumerSessions);
+//
+//                if(session == null){
+//                    logger.error("DownStream msg,retry other eventMesh found no session again");
+//                    httpExchange.sendResponseHeaders(200, 0);
+//                    out.write(result.getBytes());
+//                    return;
+//                }
+//
+//                DownStreamMsgContext downStreamMsgContext =
+//                        new DownStreamMsgContext(messageExt, session, eventMeshTCPServer.getClientSessionGroupMapping().getClientGroupWrapper(groupName).getPersistentMsgConsumer(), null, true);
+//                eventMeshTCPServer.getClientSessionGroupMapping().getClientGroupWrapper(groupName).getDownstreamMap().putIfAbsent(downStreamMsgContext.seq, downStreamMsgContext);
+//
+//                if (session.isCanDownStream()) {
+//                    session.downstreamMsg(downStreamMsgContext);
+//                    httpExchange.sendResponseHeaders(200, 0);
+//                    result = "true";
+//                    out.write(result.getBytes());
+//                    return;
+//                }
+//
+//                logger.warn("EventMeshMsgDownStreamHandler|dispatch retry, seq[{}]", downStreamMsgContext.seq);
+//                long delayTime = EventMeshUtil.isService(downStreamMsgContext.msgExt.getTopic()) ? 0 : eventMeshTCPServer.getAccessConfiguration().eventMeshTcpMsgRetryDelayInMills;
+//                downStreamMsgContext.delay(delayTime);
+//                eventMeshTCPServer.getEventMeshTcpRetryer().pushRetry(downStreamMsgContext);
+//                result = "true";
+//                httpExchange.sendResponseHeaders(200, 0);
+//                out.write(result.getBytes());
+
+        } catch (Exception e) {
+            logger.error("EventMeshMsgDownStreamHandler handle fail...", e);
+        } finally {
+            if (out != null) {
+                try {
+                    out.close();
+                } catch (IOException e) {
+                    logger.warn("out close failed...", e);
+                }
+            }
+        }
+    }
+
+    private Map<String, Object> parsePostParameters(HttpExchange exchange)
+            throws IOException {
+        Map<String, Object> parameters = new HashMap<>();
+        if ("post".equalsIgnoreCase(exchange.getRequestMethod())) {
+            InputStreamReader isr =
+                    new InputStreamReader(exchange.getRequestBody(), "utf-8");
+            BufferedReader br = new BufferedReader(isr);
+            String query = br.readLine();
+            parseQuery(query, parameters);
+        }
+        return parameters;
+    }
+
+    @SuppressWarnings("unchecked")
+    private void parseQuery(String query, Map<String, Object> parameters)
+            throws UnsupportedEncodingException {
+
+        if (query != null) {
+            String pairs[] = query.split("&");
+
+            for (String pair : pairs) {
+                String param[] = pair.split("=");
+
+                String key = null;
+                String value = null;
+                if (param.length > 0) {
+                    key = URLDecoder.decode(param[0], "UTF-8");
+                }
+
+                if (param.length > 1) {
+                    value = URLDecoder.decode(param[1], "UTF-8");
+                }
+
+                if (parameters.containsKey(key)) {
+                    Object obj = parameters.get(key);
+                    if (obj instanceof List<?>) {
+                        List<String> values = (List<String>) obj;
+                        values.add(value);
+                    } else if (obj instanceof String) {
+                        List<String> values = new ArrayList<String>();
+                        values.add((String) obj);
+                        values.add(value);
+                        parameters.put(key, values);
+                    }
+                } else {
+                    parameters.put(key, value);
+                }
+            }
+        }
+    }
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java

@@ -0,0 +1,111 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.runtime.admin.handler;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcp2Client;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientSessionGroupMapping;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
+import org.apache.eventmesh.runtime.util.NetUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class RedirectClientByIpPortHandler implements HttpHandler {
+
+    private static final Logger logger = LoggerFactory.getLogger(RedirectClientByIpPortHandler.class);
+
+    private final EventMeshTCPServer eventMeshTCPServer;
+
+    public RedirectClientByIpPortHandler(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+    }
+
+    @Override
+    public void handle(HttpExchange httpExchange) throws IOException {
+        String result = "";
+        OutputStream out = httpExchange.getResponseBody();
+        try {
+            String queryString = httpExchange.getRequestURI().getQuery();
+            Map<String, String> queryStringInfo = NetUtils.formData2Dic(queryString);
+            String ip = queryStringInfo.get(EventMeshConstants.MANAGE_IP);
+            String port = queryStringInfo.get(EventMeshConstants.MANAGE_PORT);
+            String destEventMeshIp = queryStringInfo.get(EventMeshConstants.MANAGE_DEST_IP);
+            String destEventMeshPort = queryStringInfo.get(EventMeshConstants.MANAGE_DEST_PORT);
+
+            if (StringUtils.isBlank(ip) || !StringUtils.isNumeric(port)
+                    || StringUtils.isBlank(destEventMeshIp) || StringUtils.isBlank(destEventMeshPort)
+                    || !StringUtils.isNumeric(destEventMeshPort)) {
+                httpExchange.sendResponseHeaders(200, 0);
+                result = "params illegal!";
+                out.write(result.getBytes());
+                return;
+            }
+            logger.info("redirectClientByIpPort in admin,ip:{},port:{},destIp:{},destPort:{}====================", ip, port, destEventMeshIp, destEventMeshPort);
+            ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
+            ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+            String redirectResult = "";
+            try {
+                if (!sessionMap.isEmpty()) {
+                    for (Session session : sessionMap.values()) {
+                        if (session.getClient().getHost().equals(ip) && String.valueOf(session.getClient().getPort()).equals(port)) {
+                            redirectResult += "|";
+                            redirectResult += EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer, destEventMeshIp, Integer.parseInt(destEventMeshPort),
+                                    session, clientSessionGroupMapping);
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                logger.error("clientManage|redirectClientByIpPort|fail|ip={}|port={}|destEventMeshIp" +
+                        "={}|destEventMeshPort={},errMsg={}", ip, port, destEventMeshIp, destEventMeshPort, e);
+                result = String.format("redirectClientByIpPort fail! sessionMap size {%d}, {clientIp=%s clientPort=%s " +
+                                "destEventMeshIp=%s destEventMeshPort=%s}, result {%s}, errorMsg : %s",
+                        sessionMap.size(), ip, port, destEventMeshIp, destEventMeshPort, redirectResult, e
+                                .getMessage());
+                httpExchange.sendResponseHeaders(200, 0);
+                out.write(result.getBytes());
+                return;
+            }
+            result = String.format("redirectClientByIpPort success! sessionMap size {%d}, {ip=%s port=%s " +
+                            "destEventMeshIp=%s destEventMeshPort=%s}, result {%s} ",
+                    sessionMap.size(), ip, port, destEventMeshIp, destEventMeshPort, redirectResult);
+            httpExchange.sendResponseHeaders(200, 0);
+            out.write(result.getBytes());
+        } catch (Exception e) {
+            logger.error("redirectClientByIpPort fail...", e);
+        } finally {
+            if (out != null) {
+                try {
+                    out.close();
+                } catch (IOException e) {
+                    logger.warn("out close failed...", e);
+                }
+            }
+        }
+
+    }
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java

@@ -0,0 +1,112 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.runtime.admin.handler;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcp2Client;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientSessionGroupMapping;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
+import org.apache.eventmesh.runtime.util.NetUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * redirect subsystem for path
+ */
+public class RedirectClientByPathHandler implements HttpHandler {
+
+    private static final Logger logger = LoggerFactory.getLogger(RedirectClientByPathHandler.class);
+
+    private EventMeshTCPServer eventMeshTCPServer;
+
+    public RedirectClientByPathHandler(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+    }
+
+    @Override
+    public void handle(HttpExchange httpExchange) throws IOException {
+        String result = "";
+        OutputStream out = httpExchange.getResponseBody();
+        try {
+            String queryString = httpExchange.getRequestURI().getQuery();
+            Map<String, String> queryStringInfo = NetUtils.formData2Dic(queryString);
+            String path = queryStringInfo.get(EventMeshConstants.MANAGE_PATH);
+            String destEventMeshIp = queryStringInfo.get(EventMeshConstants.MANAGE_DEST_IP);
+            String destEventMeshPort = queryStringInfo.get(EventMeshConstants.MANAGE_DEST_PORT);
+
+            if (StringUtils.isBlank(path) || StringUtils.isBlank(destEventMeshIp) || StringUtils.isBlank(destEventMeshPort) ||
+                    !StringUtils.isNumeric(destEventMeshPort)) {
+                httpExchange.sendResponseHeaders(200, 0);
+                result = "params illegal!";
+                out.write(result.getBytes());
+                return;
+            }
+            logger.info("redirectClientByPath in admin,path:{},destIp:{},destPort:{}====================", path, destEventMeshIp, destEventMeshPort);
+            ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
+            ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+            String redirectResult = "";
+            try {
+                if (!sessionMap.isEmpty()) {
+                    for (Session session : sessionMap.values()) {
+                        if (session.getClient().getPath().contains(path)) {
+                            redirectResult += "|";
+                            redirectResult += EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer, destEventMeshIp, Integer.parseInt(destEventMeshPort),
+                                    session, clientSessionGroupMapping);
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                logger.error("clientManage|redirectClientByPath|fail|path={}|destEventMeshIp" +
+                        "={}|destEventMeshPort={},errMsg={}", path, destEventMeshIp, destEventMeshPort, e);
+                result = String.format("redirectClientByPath fail! sessionMap size {%d}, {path=%s " +
+                                "destEventMeshIp=%s destEventMeshPort=%s}, result {%s}, errorMsg : %s",
+                        sessionMap.size(), path, destEventMeshIp, destEventMeshPort, redirectResult, e
+                                .getMessage());
+                httpExchange.sendResponseHeaders(200, 0);
+                out.write(result.getBytes());
+                return;
+            }
+            result = String.format("redirectClientByPath success! sessionMap size {%d}, {path=%s " +
+                            "destEventMeshIp=%s destEventMeshPort=%s}, result {%s} ",
+                    sessionMap.size(), path, destEventMeshIp, destEventMeshPort, redirectResult);
+            httpExchange.sendResponseHeaders(200, 0);
+            out.write(result.getBytes());
+        } catch (Exception e) {
+            logger.error("redirectClientByPath fail...", e);
+        } finally {
+            if (out != null) {
+                try {
+                    out.close();
+                } catch (IOException e) {
+                    logger.warn("out close failed...", e);
+                }
+            }
+        }
+
+    }
+}