[ISSUE #4767] Refactor Admin server http handler.

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/HttpCommand.java

@@ -27,6 +27,7 @@
 import org.apache.eventmesh.common.protocol.http.header.Header;
 import org.apache.eventmesh.common.utils.JsonUtils;
 
+import java.net.URI;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;
@@ -41,6 +42,7 @@
 
 import lombok.Data;
 
+@Deprecated
 @Data
 public class HttpCommand implements ProtocolTransportObject {
 
@@ -66,6 +68,8 @@ public class HttpCommand implements ProtocolTransportObject {
     // Command response time
     public long resTime;
 
+    public URI requestURI;
+
     public CmdType cmdType = CmdType.REQ;
 
     public HttpCommand() {

eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/JsonUtils.java

@@ -25,6 +25,7 @@
 
 import java.io.IOException;
 import java.lang.reflect.Type;
+import java.util.Map;
 import java.util.Objects;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -52,6 +53,15 @@ public class JsonUtils {
         OBJECT_MAPPER.registerModule(new JavaTimeModule());
     }
 
+    public static <T> T mapToObject(Map<String, Object> map, Class<T> beanClass) {
+        if (map == null) {
+            return null;
+        }
+        Object obj = OBJECT_MAPPER.convertValue(map, beanClass);
+        return (T) obj;
+    }
+
+
     /**
      * Serialize object to json string.
      *

eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/NetUtils.java

@@ -17,15 +17,10 @@
 
 package org.apache.eventmesh.common.utils;
 
-import static org.apache.eventmesh.common.Constants.SUCCESS_CODE;
-
 import org.apache.eventmesh.common.Constants;
-import org.apache.eventmesh.common.enums.HttpMethod;
 
 import org.apache.commons.lang3.StringUtils;
 
-import java.io.IOException;
-import java.io.InputStreamReader;
 import java.io.UnsupportedEncodingException;
 import java.net.InetSocketAddress;
 import java.net.URLDecoder;
@@ -34,8 +29,6 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import com.sun.net.httpserver.HttpExchange;
-
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -81,26 +74,4 @@ public static String addressToString(Collection<InetSocketAddress> clients) {
         }
         return sb.toString();
     }
-
-    public static String parsePostBody(HttpExchange exchange)
-        throws IOException {
-
-        if (!HttpMethod.POST.name().equalsIgnoreCase(exchange.getRequestMethod())
-            && !HttpMethod.PUT.name().equalsIgnoreCase(exchange.getRequestMethod())) {
-            return "";
-        }
-        StringBuilder body = new StringBuilder(1024);
-        try (InputStreamReader reader = new InputStreamReader(exchange.getRequestBody(), Constants.DEFAULT_CHARSET.name())) {
-            char[] buffer = new char[256];
-            int readIndex;
-            while ((readIndex = reader.read(buffer)) != -1) {
-                body.append(buffer, 0, readIndex);
-            }
-        }
-        return body.toString();
-    }
-
-    public static void sendSuccessResponseHeaders(HttpExchange httpExchange) throws IOException {
-        httpExchange.sendResponseHeaders(SUCCESS_CODE, 0);
-    }
 }

eventmesh-common/src/test/java/org/apache/eventmesh/common/utils/NetUtilsTest.java

@@ -17,21 +17,13 @@
 
 package org.apache.eventmesh.common.utils;
 
-import org.apache.eventmesh.common.enums.HttpMethod;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-import com.sun.net.httpserver.HttpExchange;
 
 public class NetUtilsTest {
 
@@ -58,25 +50,4 @@ public void testAddressToString() {
         Assertions.assertEquals(localAddress + "|", result);
     }
 
-    @Test
-    public void testParsePostBody() throws Exception {
-
-        HttpExchange exchange = Mockito.mock(HttpExchange.class);
-        String expected = "mxsm";
-        ByteArrayInputStream inputStream = new ByteArrayInputStream(expected.getBytes(StandardCharsets.UTF_8));
-        Mockito.when(exchange.getRequestMethod()).thenReturn(HttpMethod.POST.name());
-        Mockito.when(exchange.getRequestBody()).thenReturn(inputStream);
-
-        String actual = NetUtils.parsePostBody(exchange);
-        Assertions.assertEquals(expected, actual);
-
-    }
-
-    @Test
-    public void testSendSuccessResponseHeaders() throws IOException {
-        HttpExchange exchange = Mockito.mock(HttpExchange.class);
-        NetUtils.sendSuccessResponseHeaders(exchange);
-        Mockito.verify(exchange, Mockito.times(1))
-            .sendResponseHeaders(Mockito.anyInt(), Mockito.anyLong());
-    }
 }

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

@@ -17,18 +17,161 @@
 
 package org.apache.eventmesh.runtime.admin.handler;
 
-import com.sun.net.httpserver.HttpHandler;
+import org.apache.eventmesh.common.enums.HttpMethod;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.util.HttpResponseUtils;
 
-import lombok.Data;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 
-/**
- * An abstract class that implements the {@link HttpHandler} interface
- * and provides basic functionality for HTTP request handling.
- * <p>
- * Subclasses should extend this class to implement specific HTTP request handling logic.
- */
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.codec.http.multipart.Attribute;
+import io.netty.handler.codec.http.multipart.DefaultHttpDataFactory;
+import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
+import io.netty.handler.codec.http.multipart.InterfaceHttpData;
+import io.netty.util.AsciiString;
+
+import lombok.Data;
 
 @Data
-public abstract class AbstractHttpHandler implements HttpHandler {
+public abstract class AbstractHttpHandler implements org.apache.eventmesh.runtime.admin.handler.HttpHandler {
+
+    private  static final DefaultHttpDataFactory DEFAULT_HTTP_DATA_FACTORY = new DefaultHttpDataFactory(false);
+
+    protected void write(ChannelHandlerContext ctx, byte[] result, AsciiString headerValue) {
+        ctx.writeAndFlush(HttpResponseUtils.getHttpResponse(result, ctx, headerValue)).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    protected void write(ChannelHandlerContext ctx, HttpResponse response) {
+        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    protected void write(ChannelHandlerContext ctx, byte[] result, AsciiString headerValue, HttpResponseStatus httpResponseStatus) {
+        ctx.writeAndFlush(HttpResponseUtils.getHttpResponse(result, ctx, headerValue, httpResponseStatus)).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    protected void write401(ChannelHandlerContext ctx) {
+        DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.UNAUTHORIZED);
+        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    protected void writeSuccess(ChannelHandlerContext ctx) {
+        HttpHeaders responseHeaders = new DefaultHttpHeaders();
+        responseHeaders.add(EventMeshConstants.HANDLER_ORIGIN, "*");
+        DefaultFullHttpResponse response =
+            new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.EMPTY_BUFFER, responseHeaders, responseHeaders);
+        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    protected void writeSuccess(ChannelHandlerContext ctx, DefaultHttpHeaders responseHeaders) {
+        DefaultFullHttpResponse response =
+            new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.EMPTY_BUFFER, responseHeaders, responseHeaders);
+        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    protected void preflight(ChannelHandlerContext ctx) {
+        HttpHeaders responseHeaders = new DefaultHttpHeaders();
+        responseHeaders.add(EventMeshConstants.HANDLER_ORIGIN, "*");
+        responseHeaders.add(EventMeshConstants.HANDLER_METHODS, "*");
+        responseHeaders.add(EventMeshConstants.HANDLER_HEADERS, "*");
+        responseHeaders.add(EventMeshConstants.HANDLER_AGE, EventMeshConstants.MAX_AGE);
+        DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.EMPTY_BUFFER,
+            responseHeaders, responseHeaders);
+        write(ctx, response);
+    }
+
+    /**
+     * Converts a query string to a map of key-value pairs.
+     * <p>
+     * This method takes a query string and parses it to create a map of key-value pairs, where each key and value are extracted from the query string
+     * separated by '='.
+     * <p>
+     * If the query string is null, an empty map is returned.
+     *
+     * @param query the query string to convert to a map
+     * @return a map containing the key-value pairs from the query string
+     */
+    protected Map<String, String> queryToMap(String query) {
+        if (query == null) {
+            return new HashMap<>();
+        }
+        Map<String, String> result = new HashMap<>();
+        for (String param : query.split("&")) {
+            String[] entry = param.split("=");
+            if (entry.length > 1) {
+                result.put(entry[0], entry[1]);
+            } else {
+                result.put(entry[0], "");
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public void handle(HttpRequest httpRequest, ChannelHandlerContext ctx) throws Exception {
+        switch (HttpMethod.valueOf(httpRequest.method().name())) {
+            case OPTIONS:
+                preflight(ctx);
+                break;
+            case GET:
+                get(httpRequest, ctx);
+                break;
+            case POST:
+                post(httpRequest, ctx);
+                break;
+            case DELETE:
+                delete(httpRequest, ctx);
+                break;
+            default: // do nothing
+                break;
+        }
+    }
+
+    protected void post(HttpRequest httpRequest, ChannelHandlerContext ctx) throws Exception{
+    }
+
+    protected void delete(HttpRequest httpRequest, ChannelHandlerContext ctx) throws Exception{
+    }
+
+    protected void get(HttpRequest httpRequest, ChannelHandlerContext ctx) throws Exception{
+    }
+
+    protected Map<String, Object> parseHttpRequestBody(final HttpRequest httpRequest) throws IOException {
+        final long bodyDecodeStart = System.currentTimeMillis();
+        final Map<String, Object> httpRequestBody = new HashMap<>();
+
+        if (io.netty.handler.codec.http.HttpMethod.GET.equals(httpRequest.method())) {
+            new QueryStringDecoder(httpRequest.uri())
+                .parameters()
+                .forEach((key, value) -> httpRequestBody.put(key, value.get(0)));
+        } else if (io.netty.handler.codec.http.HttpMethod.POST.equals(httpRequest.method())) {
+            decodeHttpRequestBody(httpRequest, httpRequestBody);
+        }
+        return httpRequestBody;
+    }
+
+    private void decodeHttpRequestBody(HttpRequest httpRequest, Map<String, Object> httpRequestBody) throws IOException {
+        final HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(DEFAULT_HTTP_DATA_FACTORY, httpRequest);
+        for (final InterfaceHttpData param : decoder.getBodyHttpDatas()) {
+            if (InterfaceHttpData.HttpDataType.Attribute == param.getHttpDataType()) {
+                final Attribute data = (Attribute) param;
+                httpRequestBody.put(data.getName(), data.getValue());
+            }
+        }
+        decoder.destroy();
+    }
+
 
 }
+

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

@@ -31,7 +31,6 @@
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
-import com.sun.net.httpserver.HttpHandler;
 
 public class AdminHandlerManager {