Socket Client 적용

2024-04-12 11:12:50 +09:00
parent dc3c42a341
commit 95924af31a
13 changed files with 389 additions and 104 deletions
@@ -2,23 +2,23 @@ package com.eactive.testmaster.client.controller;

 import com.eactive.testmaster.client.dto.ApiRequestDTO;
 import com.eactive.testmaster.client.dto.ApiResponse;
-import com.eactive.testmaster.client.service.NettyApiClient;
+import com.eactive.testmaster.client.service.ApiClient;
+import com.eactive.testmaster.client.service.ApiClientFactory;
+import java.util.concurrent.ExecutionException;
 import org.springframework.validation.BindingResult;
 import org.springframework.validation.Validator;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;

-import java.util.concurrent.ExecutionException;
-
@RestController
 public class ApiClientController {

-    private final NettyApiClient nettyApiClient;
+    private final ApiClientFactory apiClientFactory;
    private final Validator validator;

-    public ApiClientController(NettyApiClient nettyApiClient, Validator validator) {
-        this.nettyApiClient = nettyApiClient;
+    public ApiClientController(ApiClientFactory apiClientFactory, Validator validator) {
+        this.apiClientFactory = apiClientFactory;
        this.validator = validator;
    }

@@ -31,6 +31,8 @@ public class ApiClientController {
            throw new IllegalArgumentException(bindingResult.getAllErrors().get(0).getDefaultMessage());
        }

-        return nettyApiClient.handleRequest(request).get();
+        ApiClient client = apiClientFactory.getClient(request);
+        return client.handleRequest(request).get();
    }
+
 }
@@ -16,6 +16,8 @@ public class ApiRequestDTO implements Serializable {
    private String path;
    private String method;

+    private String type;
+
    @NotNull(message = "서버를 선택해주세요.")
    private Long server;

@@ -69,6 +71,14 @@ public class ApiRequestDTO implements Serializable {
        return method;
    }

+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
    public void setMethod(String method) {
        this.method = method;
    }
@@ -121,11 +131,11 @@ public class ApiRequestDTO implements Serializable {
        this.postRequestScript = postRequestScript;
    }

-    public long getSentBytes() {
-        return sentBytes;
-    }
-
-    public void setSentBytes(long sentBytes) {
-        this.sentBytes = sentBytes;
-    }
+//    public long getSentBytes() {
+//        return sentBytes;
+//    }
+//
+//    public void setSentBytes(long sentBytes) {
+//        this.sentBytes = sentBytes;
+//    }
 }
@@ -28,6 +28,8 @@ public class ApiRequestInfo {

    private String path;

+    private String type; //API TYPE : HTTP, TCP
+
    @Lob
    private String headers;

@@ -143,6 +145,14 @@ public class ApiRequestInfo {
        this.postRequestScript = postRequestScript;
    }

+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
@@ -0,0 +1,11 @@
+package com.eactive.testmaster.client.service;
+
+
+import com.eactive.testmaster.client.dto.ApiRequestDTO;
+import com.eactive.testmaster.client.dto.ApiResponse;
+import java.util.concurrent.CompletableFuture;
+
+public interface ApiClient {
+
+    CompletableFuture<ApiResponse> handleRequest(ApiRequestDTO apiRequest);
+}
@@ -0,0 +1,27 @@
+package com.eactive.testmaster.client.service;
+
+import com.eactive.testmaster.client.dto.ApiRequestDTO;
+import org.springframework.stereotype.Service;
+
+@Service
+public class ApiClientFactory {
+
+    private final NettyHttpApiClient httpApiClient;
+    private final NettyTcpApiClient tcpApiClient;
+
+    public ApiClientFactory(NettyHttpApiClient httpApiClient, NettyTcpApiClient tcpApiClient) {
+        this.httpApiClient = httpApiClient;
+        this.tcpApiClient = tcpApiClient;
+    }
+
+    public ApiClient getClient(ApiRequestDTO request) {
+        switch (request.getType().toUpperCase()) {
+            case "HTTP":
+                return httpApiClient;
+            case "TCP":
+                return tcpApiClient;
+            default:
+                throw new IllegalArgumentException("Unsupported client type: " + request.getType());
+        }
+    }
+}
@@ -46,10 +46,9 @@ public class ApiRequestMigrationService {
        }

        apiRequestRepository.findAll().forEach(apiRequest -> {
-
-            apiRequest.setHeaders(apiRequest.getHeaders().replace("=", "::"));
-            apiRequest.setQueryParams(apiRequest.getQueryParams().replace("=", "::"));
-
+            if(apiRequest.getType() == null){
+                apiRequest.setType("HTTP");
+            }
            apiRequestRepository.save(apiRequest);
        });

@@ -30,8 +30,9 @@ import javax.net.ssl.SSLException;
 import java.util.concurrent.CompletableFuture;

@Service
-public class NettyApiClient {
-    private static final Logger logger = LoggerFactory.getLogger(NettyApiClient.class);
+public class NettyHttpApiClient implements ApiClient {
+
+    private static final Logger logger = LoggerFactory.getLogger(NettyHttpApiClient.class);

    @Autowired
    ServerRepository serverRepository;
@@ -43,29 +44,30 @@ public class NettyApiClient {

        CompletableFuture<ApiResponse> responseFuture = new CompletableFuture<>();
        EventLoopGroup group = new NioEventLoopGroup();
+
        try {
            Bootstrap b = new Bootstrap()
-                    .group(group)
-                    .channel(NioSocketChannel.class)
-                    .option(ChannelOption.TCP_NODELAY, true)
-                    .handler(new ChannelInitializer<Channel>() {
-                        @Override
-                        public void initChannel(Channel ch) throws SSLException {
-                            if (server.getScheme().equalsIgnoreCase("https")) {
-                                final SslContext sslCtx = SslContextBuilder.forClient()
-                                        .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
-                                ch.pipeline().addLast(sslCtx.newHandler(ch.alloc(), host, port));// Add SSL handler
-                            }
-
-                            ch.pipeline().addLast(new HttpClientCodec());
-                            ch.pipeline().addLast(new HttpObjectAggregator(65536));
-                            ch.pipeline().addLast(new HttpContentDecompressor());
-                            ch.pipeline().addLast(new HttpHandler(responseFuture));
+                .group(group)
+                .channel(NioSocketChannel.class)
+                .option(ChannelOption.TCP_NODELAY, true)
+                .handler(new ChannelInitializer<Channel>() {
+                    @Override
+                    public void initChannel(Channel ch) throws SSLException {
+                        if (server.getScheme().equalsIgnoreCase("https")) {
+                            final SslContext sslCtx = SslContextBuilder.forClient()
+                                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
+                            ch.pipeline().addLast(sslCtx.newHandler(ch.alloc(), host, port));// Add SSL handler
                        }
-                    });
+
+                        ch.pipeline().addLast(new HttpClientCodec());
+                        ch.pipeline().addLast(new HttpObjectAggregator(65536));
+                        ch.pipeline().addLast(new HttpContentDecompressor());
+                        ch.pipeline().addLast(new HttpHandler(responseFuture));
+                    }
+                });

            Channel ch = b.connect(host, port).sync().channel();
-            ByteBuf content = null;
+            ByteBuf content;
            if (StringUtils.isNotEmpty(apiRequest.getRequestBody())) {
                content = Unpooled.copiedBuffer(apiRequest.getRequestBody(), CharsetUtil.UTF_8);
            } else {
@@ -94,7 +96,7 @@ public class NettyApiClient {
                request.headers().add(customHeaders);
            }

-            apiRequest.setSentBytes(request.toString().length());
+//            apiRequest.setSentBytes(request.toString().length());

            ch.writeAndFlush(request);
            ch.closeFuture().sync();
@@ -0,0 +1,106 @@
+package com.eactive.testmaster.client.service;
+
+import com.eactive.testmaster.client.dto.ApiRequestDTO;
+import com.eactive.testmaster.client.dto.ApiResponse;
+import com.eactive.testmaster.common.exception.ServerNotFoundException;
+import com.eactive.testmaster.server.entity.Server;
+import com.eactive.testmaster.server.entity.ServerRepository;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+@Service
+public class NettyTcpApiClient implements ApiClient {
+
+    private static final Logger logger = LoggerFactory.getLogger(NettyTcpApiClient.class);
+
+    private final ServerRepository serverRepository;
+
+    public NettyTcpApiClient(ServerRepository serverRepository) {
+        this.serverRepository = serverRepository;
+    }
+
+    public CompletableFuture<ApiResponse> handleRequest(ApiRequestDTO apiRequest) {
+        Server server = serverRepository.findById(apiRequest.getServer()).orElseThrow(() -> new ServerNotFoundException("Server not found"));
+        String host = server.getHostname();
+        int port = server.getPort();
+
+        CompletableFuture<ApiResponse> responseFuture = new CompletableFuture<>();
+        EventLoopGroup group = new NioEventLoopGroup();
+        try {
+            String response = sendMessage(host, port, group, apiRequest.getRequestBody());
+            ApiResponse apiResponse = new ApiResponse();
+            apiResponse.setBody(response);
+            responseFuture.complete(apiResponse);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            ApiResponse response = new ApiResponse();
+            response.setBody("An error occurred: " + e.getMessage());
+            return CompletableFuture.completedFuture(response);
+        } finally {
+            group.shutdownGracefully();
+        }
+        return responseFuture;
+    }
+
+    public String sendMessage(String host, int port, EventLoopGroup group, final String message) throws InterruptedException {
+        final BlockingQueue<String> answer = new ArrayBlockingQueue<>(1);
+        Bootstrap b = new Bootstrap();
+        b.group(group)
+            .channel(NioSocketChannel.class)
+            .handler(new ChannelInitializer<SocketChannel>() {
+                @Override
+                protected void initChannel(SocketChannel ch) {
+                    ch.pipeline().addLast(new StringDecoder(), new StringEncoder(), new SimpleChannelInboundHandler<String>() {
+                        @Override
+                        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
+                            answer.offer(msg);
+                            ctx.close();
+                        }
+
+                        @Override
+                        public void channelActive(ChannelHandlerContext ctx) {
+                            ctx.writeAndFlush(message);
+                            // Send message on activation, consider thread safety
+                        }
+                    });
+                }
+            });
+
+        // Connect to the server
+        ChannelFuture f = b.connect(host, port).sync();
+        f.channel().closeFuture().sync();
+
+        // Wait for the response
+        return answer.take(); // This will block until a message is available
+    }
+
+
+//    public static void main(String[] args) {
+//        NettyTcpApiClient client = new NettyTcpApiClient();
+//        try {
+//            String response = client.sendMessage("localhost", 30913, "00340000BASICTST010134567890ABCDEF");
+//            System.out.println("Server replied: " + response);
+//        } catch (InterruptedException e) {
+//            Thread.currentThread().interrupt();
+//            e.printStackTrace();
+//        } finally {
+//            client.shutdown();
+//        }
+//    }
+
+}
@@ -6,7 +6,7 @@ import com.eactive.testmaster.client.entity.ApiRequestInfo;
 import com.eactive.testmaster.client.entity.ApiScenario;
 import com.eactive.testmaster.client.mapper.ApiRequestMapper;
 import com.eactive.testmaster.client.mapper.ApiScenarioMapper;
-import com.eactive.testmaster.client.service.NettyApiClient;
+import com.eactive.testmaster.client.service.ApiClientFactory;
 import com.eactive.testmaster.loadtest.dto.ApiRequestEvent;
 import com.eactive.testmaster.loadtest.dto.LoadTestDTO;
 import com.eactive.testmaster.loadtest.dto.LoadTestRequestAndResponse;
@@ -42,20 +42,19 @@ public class LoadTestService extends TextWebSocketHandler {

    private LoadTestMonitorMap loadTestMonitorMap;

-    private NettyApiClient nettyApiClient;
-
+    private ApiClientFactory apiClientFactory;
    private ApiScenarioMapper apiScenarioMapper;

    private ApiRequestMapper apiRequestMapper;

    private final RingBuffer<ApiRequestEvent> ringBuffer;

-    public LoadTestService(SimpMessagingTemplate template, LoadTestMonitorMap loadTestMonitorMap, NettyApiClient nettyApiClient, ApiScenarioMapper apiScenarioMapper, ApiRequestMapper apiRequestMapper) {
+    public LoadTestService(SimpMessagingTemplate template, LoadTestMonitorMap loadTestMonitorMap,ApiClientFactory apiClientFactory, ApiScenarioMapper apiScenarioMapper, ApiRequestMapper apiRequestMapper) {
        this.template = template;
        this.loadTestMonitorMap = loadTestMonitorMap;
-        this.nettyApiClient = nettyApiClient;
        this.apiScenarioMapper = apiScenarioMapper;
        this.apiRequestMapper = apiRequestMapper;
+        this.apiClientFactory = apiClientFactory;

        Disruptor<ApiRequestEvent> disruptor = new Disruptor<>(ApiRequestEvent::new, 1024, Executors.defaultThreadFactory());
        disruptor.handleEventsWith(this::processApiRequestEvent);
@@ -137,7 +136,7 @@ public class LoadTestService extends TextWebSocketHandler {
        List<ScheduledFuture<?>> scheduledFutures = new ArrayList<>();
        List<VirtualUser> virtualUsers = new ArrayList<>();
        for (int i = 0; i < numberOfUsers; i++) {
-            final VirtualUser virtualUser = new VirtualUser(loadTestDTO.getId(), maxCount, scenario, this, nettyApiClient, latch);
+            final VirtualUser virtualUser = new VirtualUser(loadTestDTO.getId(), maxCount, scenario, this, apiClientFactory, latch);
            virtualUsers.add(virtualUser);

            Instant startTime = Instant.now().plusMillis(i * rampUpTimeMillis);
@@ -4,7 +4,7 @@ import com.eactive.testmaster.client.dto.ApiRequestDTO;
 import com.eactive.testmaster.client.dto.ApiRequestKeyValueDTO;
 import com.eactive.testmaster.client.dto.ApiResponse;
 import com.eactive.testmaster.client.dto.internal.ScenarioNode;
-import com.eactive.testmaster.client.service.NettyApiClient;
+import com.eactive.testmaster.client.service.ApiClientFactory;
 import com.eactive.testmaster.loadtest.dto.LoadTestRequestAndResponse;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -36,7 +36,7 @@ public class VirtualUser implements Runnable {
    private Thread runningThread;

    private final LoadTestService loadTestService;
-    private final NettyApiClient nettyApiClient;
+    private final ApiClientFactory apiClientFactory;

    private static Map<String, String> globals = new ConcurrentHashMap<>();
    private Map<String, String> variables = new ConcurrentHashMap<>();
@@ -49,12 +49,12 @@ public class VirtualUser implements Runnable {

    private CountDownLatch latch;

-    public VirtualUser(String testId, int maxCount, ScenarioNode scenarioDTO, LoadTestService loadTestService, NettyApiClient nettyApiClient, CountDownLatch latch) {
+    public VirtualUser(String testId, int maxCount, ScenarioNode scenarioDTO, LoadTestService loadTestService, ApiClientFactory apiClientFactory, CountDownLatch latch) {
        this.testId = testId;
        this.scenario = scenarioDTO;
        this.loadTestService = loadTestService;
-        this.nettyApiClient = nettyApiClient;
        this.maxCount = maxCount;
+        this.apiClientFactory = apiClientFactory;
        this.latch = latch;

    }
@@ -151,7 +151,7 @@ public class VirtualUser implements Runnable {
                ApiRequestDTO replaced = replacePlaceHoldersWithVariables(cloned);

                long startTime = System.currentTimeMillis();
-                ApiResponse response = nettyApiClient.handleRequest(replaced).get();
+                ApiResponse response = apiClientFactory.getClient(replaced).handleRequest(replaced).get();
                long endTime = System.currentTimeMillis();
                long responseTime = endTime - startTime;