저장된 ReceiveMessage 처리 진행중

src/docs/settings-examples/dfxagent.json

@@ -1,14 +1,16 @@
 {
   "host-id": "first-agent",
   "listen-port": 16801,
-  "known-agent": [
+  "known-agent-list": [
     {
       "host-id": "second-agent",
+      "host-name": "mkami.foxsoft.kr",
       "listen-port": 16801,
       "drop-box-id-list": ["drop1", "drop2"]
     },
     {
       "host-id": "third-agent",
+      "host-name": "defree.co.kr",
       "listen-port": 16801,
       "drop-box-id-list": ["drop3", "drop4"]
     }

src/main/java/com/bsmlab/dfx/agent/config/AgentConfigDto.java

@@ -8,7 +8,7 @@ import java.util.List;
 public class AgentConfigDto {
     private String hostId;
     private int listenPort;
-    private List<KnownAgent> knownAgent;
+    private List<KnownAgent> knownAgentList;
     private List<DataSourceConfig> datasourceConfig;
     private List<String> sqlMapperLocations;
     private DropBoxConfig dropBox;
@@ -18,6 +18,7 @@ public class AgentConfigDto {
     @Data
     public static class KnownAgent {
         private String hostId;
+        private String hostName;
         private int listenPort;
         private List<String> dropBoxIdList;
     }

src/main/java/com/bsmlab/dfx/agent/config/AgentConfigReader.java

@@ -38,4 +38,15 @@ public class AgentConfigReader {
         }
         return found;
     }
+
+    public AgentConfigDto.KnownAgent getKnownAgent(String hostId) {
+        AgentConfigDto.KnownAgent found = null;
+        for(AgentConfigDto.KnownAgent knownAgent : this.agentConfigDto.getKnownAgentList()) {
+            if(knownAgent.getHostId().equals(hostId)) {
+                found = knownAgent;
+                break;
+            }
+        }
+        return found;
+    }
 }

src/main/java/com/bsmlab/dfx/agent/listener/ListenerController.java

@@ -17,6 +17,11 @@ import java.io.IOException;
 public class ListenerController {
     private final ListenerService listenerService;
 
+    /**
+     * 다른 Agent 에서 수신한 메시지를 저장하고 공유메모리에 등록한다. 메시지 처리는 TaskExecutorService 가 수행한다.
+     * @param request
+     * @return
+     */
     @PostMapping(value = "/listen")
     public AckDto listener(HttpServletRequest request) {
         AckDto ackDto = AckDto.builder().build();
@@ -24,9 +29,18 @@ public class ListenerController {
             String bodyString = ServletUtils.getBodyString(request);
             ackDto = listenerService.receiveMessage(bodyString);
         } catch (IOException e) {
-            ackDto.setResult(AckDto.RECEIVE_FAIL);
+            ackDto.setResult(AckDto.ResultType.RECEIVE_FAIL);
-            ackDto.setMessage(e.getMessage());
+            ackDto.setResultText(e.getMessage());
         }
         return ackDto;
     }
+
+    /**
+     * 처리 결과 수신
+     */
+    @PostMapping(value = "/telegram")
+    public AckDto telegram(HttpServletRequest request) {
+        AckDto ackDto = AckDto.builder().build();
+        return ackDto;
+    }
 }

src/main/java/com/bsmlab/dfx/agent/listener/dto/AckDto.java

@@ -18,10 +18,14 @@ import lombok.Setter;
  * 4.2 Worker 동작 실패시 sender 에게 처리오류 Ack 전달 (result : PROCESS_FAIL)
  */
 public class AckDto {
-    public static final String RECEIVE_SUCCESS = "RECEIVE_SUCCESS";
+    private ResultType result;
-    public static final String RECEIVE_FAIL = "RECEIVE_FAIL";
+    private String resultText;
-    public static final String PROCESS_SUCCESS = "PROCESS_SUCCESS";
+    private String messageUuid;
-    public static final String PROCESS_FAIL = "PROCESS_FAIL";
+
-    private String result;
+    public static enum ResultType {
-    private String message;
+        RECEIVE_SUCCESS,
+        RECEIVE_FAIL,
+        PROCESS_SUCCESS,
+        PROCESS_FAIL
+    }
 }

src/main/java/com/bsmlab/dfx/agent/listener/service/ListenerService.java

@@ -1,9 +1,6 @@
 package com.bsmlab.dfx.agent.listener.service;
 
-import com.bsmlab.dfx.agent.config.Settings;
-import com.bsmlab.dfx.agent.config.datasource.SqlExecuteService;
 import com.bsmlab.dfx.agent.listener.dto.AckDto;
-import com.bsmlab.dfx.agent.listener.dto.ListenerMapper;
 import com.bsmlab.dfx.agent.listener.dto.ReceiveMessageDto;
 import com.bsmlab.dfx.agent.support.MessageUtils;
 import com.bsmlab.dfx.agent.support.exception.IllegalMessageException;
@@ -25,15 +22,15 @@ public class ListenerService {
         try {
             ReceiveMessageDto receiveMessageDto = MessageUtils.toReceiveMessageDto(messageJsonString);
             if(dropBoxService.isExistToday(receiveMessageDto)) {
-                ackDto = AckDto.builder().result(AckDto.RECEIVE_FAIL).message("금일 전송한 메시지 중 중복된 UUID가 존재합니다.").build();
+                ackDto = AckDto.builder().result(AckDto.ResultType.RECEIVE_FAIL).resultText("금일 전송한 메시지 중 중복된 UUID가 존재합니다.").build();
             }
             else {
                 dropBoxService.add(receiveMessageDto);
-                ackDto = AckDto.builder().result(AckDto.RECEIVE_SUCCESS).build();
+                ackDto = AckDto.builder().result(AckDto.ResultType.RECEIVE_SUCCESS).build();
             }
         } catch (IllegalMessageException | NullMessageException | InCompleteMessageException e) {
             log.error("{}", e, e);
-            ackDto = AckDto.builder().result(AckDto.RECEIVE_FAIL).message(e.getLocalizedMessage()).build();
+            ackDto = AckDto.builder().result(AckDto.ResultType.RECEIVE_FAIL).resultText(e.getLocalizedMessage()).build();
         }
         return ackDto;
     }

src/main/java/com/bsmlab/dfx/agent/task/TaskExecutorService.java

@@ -3,12 +3,18 @@ package com.bsmlab.dfx.agent.task;
 import com.bsmlab.dfx.agent.config.AgentConfigDto;
 import com.bsmlab.dfx.agent.config.AgentConfigReader;
 import com.bsmlab.dfx.agent.config.datasource.SqlExecuteService;
+import com.bsmlab.dfx.agent.listener.dto.AckDto;
 import com.bsmlab.dfx.agent.listener.dto.ReceiveMessageDto;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
+import org.springframework.web.client.RestTemplate;
 
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
@@ -41,10 +47,13 @@ public class TaskExecutorService {
 
             }
         } catch (FileNotFoundException e) {
+            //TODO 수신한 메시지 파일을 찾지 못했으니 메시지 처리 불가. 로그 남기고 Ack.PROCESS_FAIL 전달
             throw new RuntimeException(e);
         } catch (IOException e) {
+            //TODO 수신한 메시지를 로드하지 못했으니 메시지 처리 불가. 로그 남기고 파일을 미처리 상태 경로로 옮기고 Ack.PROCESS_FAIL 전달
             throw new RuntimeException(e);
         } catch (ClassNotFoundException e) {
+            //TODO ReceiveMessageDto 변환 실패. 로그 남기고 파일을 미처리 상태 경로로 옮기고 Ack.PROCESS_FAIL 전달
             throw new RuntimeException(e);
         }
     }
@@ -53,4 +62,24 @@ public class TaskExecutorService {
     public void processPostman() {
 
     }
+
+    private void ackDropBoxProcessResult(ReceiveMessageDto receiveMessageDto) {
+        AckDto ackDto = AckDto.builder().build();
+        HttpHeaders httpHeaders = new HttpHeaders();
+        httpHeaders.setContentType(MediaType.APPLICATION_JSON);
+        HttpEntity<AckDto> bodyEntity = new HttpEntity<>(ackDto, httpHeaders);
+        RestTemplate restTemplate = new RestTemplate();
+        //TODO sender 정보를 찾아야 함.
+        AgentConfigDto.KnownAgent knownAgent = agentConfigReader.getKnownAgent(receiveMessageDto.getSenderHostId());
+        String url = "https://" + knownAgent.getHostName() + "/telegram";
+        String response = restTemplate.postForObject("hostname", bodyEntity, String.class);
+        ObjectMapper objectMapper = new ObjectMapper();
+        try {
+            AckDto responseAckDto = objectMapper.readValue(response, AckDto.class);
+            // 수신 메시지 처리 완료. 메시지 삭제
+        } catch (JsonProcessingException e) {
+            //TODO 처리 결과 Ack 파싱 실패. 로그 남기고 메시지 삭제
+            throw new RuntimeException(e);
+        }
+    }
 }

src/main/java/com/bsmlab/dfx/agent/task/TaskExecutorStarter.java

@@ -29,7 +29,6 @@ public class TaskExecutorStarter {
                 log.error("{}", e, e);
                 Thread.currentThread().interrupt();
             }
-
         }
     }
 }