|
|
|
|
@ -5,10 +5,13 @@ 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.bsmlab.dfx.agent.task.dropbox.DropBoxService;
|
|
|
|
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
|
|
import io.micrometer.common.util.StringUtils;
|
|
|
|
|
import lombok.RequiredArgsConstructor;
|
|
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
|
|
import org.apache.commons.lang3.EnumUtils;
|
|
|
|
|
import org.springframework.http.HttpEntity;
|
|
|
|
|
import org.springframework.http.HttpHeaders;
|
|
|
|
|
import org.springframework.http.MediaType;
|
|
|
|
|
@ -16,10 +19,7 @@ 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;
|
|
|
|
|
import java.io.IOException;
|
|
|
|
|
import java.io.ObjectInputStream;
|
|
|
|
|
import java.io.*;
|
|
|
|
|
import java.util.List;
|
|
|
|
|
import java.util.Map;
|
|
|
|
|
|
|
|
|
|
@ -29,12 +29,20 @@ import java.util.Map;
|
|
|
|
|
public class TaskExecutorService {
|
|
|
|
|
private final AgentConfigReader agentConfigReader;
|
|
|
|
|
private final SqlExecuteService sqlExecuteService;
|
|
|
|
|
private final DropBoxService dropBoxService;
|
|
|
|
|
|
|
|
|
|
@Async("threadPoolTaskExecutor")
|
|
|
|
|
public void processDropBox(String messageFilePath) {
|
|
|
|
|
//TODO 1. 메시지 처리
|
|
|
|
|
ReceiveMessageDto receiveMessageDto = ReceiveMessageDto.builder().build();
|
|
|
|
|
ReceiveMessageDto.ProcessStatus processStatus = ReceiveMessageDto.ProcessStatus.PROCESS_DONE;
|
|
|
|
|
String processMessage = "";
|
|
|
|
|
String messageUuid = "";
|
|
|
|
|
try (ObjectInputStream objectInputStream = new ObjectInputStream(new FileInputStream(messageFilePath))) {
|
|
|
|
|
ReceiveMessageDto receiveMessageDto = (ReceiveMessageDto)objectInputStream.readObject();
|
|
|
|
|
receiveMessageDto = (ReceiveMessageDto)objectInputStream.readObject();
|
|
|
|
|
messageUuid = receiveMessageDto.getMessageUuid();
|
|
|
|
|
AgentConfigDto.DropBox dropBox = agentConfigReader.getDropBox(receiveMessageDto.getRecipientDropBoxId());
|
|
|
|
|
log.info("process messageUuid:{} dropBoxId: {}", receiveMessageDto.getMessageUuid(), dropBox.getDropBoxId());
|
|
|
|
|
if(dropBox.getTaskType() == AgentConfigDto.TaskType.SAVE_DB_TABLE) {
|
|
|
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
|
List<Map<String, Object>> dataMapList = null;
|
|
|
|
|
@ -44,17 +52,47 @@ public class TaskExecutorService {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
else if(dropBox.getTaskType() == AgentConfigDto.TaskType.RECEIVE_FILE) {
|
|
|
|
|
|
|
|
|
|
//TODO 1.2 파일 수신 처리
|
|
|
|
|
}
|
|
|
|
|
} catch (FileNotFoundException e) {
|
|
|
|
|
//TODO 수신한 메시지 파일을 찾지 못했으니 메시지 처리 불가. 로그 남기고 Ack.PROCESS_FAIL 전달
|
|
|
|
|
throw new RuntimeException(e);
|
|
|
|
|
//수신한 메시지 파일을 알 수 없으므로 로그 남기고 종료
|
|
|
|
|
dropBoxService.clearMessageFile(messageFilePath);
|
|
|
|
|
processStatus = ReceiveMessageDto.ProcessStatus.PROCESS_NOT_POSSIBLE;
|
|
|
|
|
processMessage = e.getMessage();
|
|
|
|
|
log.error("process not possible - messageFilePath: ");
|
|
|
|
|
log.error("{}", e, e);
|
|
|
|
|
} catch (IOException e) {
|
|
|
|
|
//TODO 수신한 메시지를 로드하지 못했으니 메시지 처리 불가. 로그 남기고 파일을 미처리 상태 경로로 옮기고 Ack.PROCESS_FAIL 전달
|
|
|
|
|
throw new RuntimeException(e);
|
|
|
|
|
//수신한 메시지를 로드하지 못했으니 메시지 처리 불가. ReceiveMessageDto 를 알 수 없으므로 로그 남기고 종료
|
|
|
|
|
processStatus = ReceiveMessageDto.ProcessStatus.PROCESS_FAIL;
|
|
|
|
|
processMessage = e.getMessage();
|
|
|
|
|
log.error("cannot load a message file - messageFilePath: ");
|
|
|
|
|
log.error("{}", e, e);
|
|
|
|
|
} catch (ClassNotFoundException e) {
|
|
|
|
|
//TODO ReceiveMessageDto 변환 실패. 로그 남기고 파일을 미처리 상태 경로로 옮기고 Ack.PROCESS_FAIL 전달
|
|
|
|
|
throw new RuntimeException(e);
|
|
|
|
|
//ReceiveMessageDto 변환 실패. 로그 남기고 파일을 미처리 상태 경로로 옮기고 ReceiveMessageDto 를 알 수 없으므로 종료
|
|
|
|
|
processStatus = ReceiveMessageDto.ProcessStatus.PROCESS_FAIL;
|
|
|
|
|
processMessage = e.getMessage();
|
|
|
|
|
log.error("cannot parse a message file - messageFilePath: ");
|
|
|
|
|
log.error("{}", e, e);
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
//기타 메시지 처리 중 오류
|
|
|
|
|
processStatus = ReceiveMessageDto.ProcessStatus.PROCESS_FAIL;
|
|
|
|
|
processMessage = e.getMessage();
|
|
|
|
|
log.error("cannot process a message file - messageFilePath: ");
|
|
|
|
|
log.error("{}", e, e);
|
|
|
|
|
}
|
|
|
|
|
//2. 결과 Ack 전송
|
|
|
|
|
if(StringUtils.isNotBlank(receiveMessageDto.getMessageUuid()) && StringUtils.isNotBlank(receiveMessageDto.getSenderHostId())) {
|
|
|
|
|
receiveMessageDto.setProcessStatus(processStatus);
|
|
|
|
|
this.ackDropBoxProcessResult(receiveMessageDto, processMessage);
|
|
|
|
|
}
|
|
|
|
|
//3. 처리한 메시지 이동
|
|
|
|
|
try {
|
|
|
|
|
boolean isSuccess = ReceiveMessageDto.ProcessStatus.PROCESS_DONE == processStatus; // PROCESS_DONE 이면 success 경로로 옮기고 아니면 failure 경로로 옮김
|
|
|
|
|
if(ReceiveMessageDto.ProcessStatus.PROCESS_NOT_POSSIBLE != processStatus) { // PROCESS_NOT_POSSIBLE 인 경우는 메시지 파일을 찾을 수 없으므로 옮길 수 없음
|
|
|
|
|
dropBoxService.moveMessageFile(messageFilePath, isSuccess);
|
|
|
|
|
}
|
|
|
|
|
} catch (IOException ex) {
|
|
|
|
|
log.error("{}", ex, ex);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -63,23 +101,25 @@ public class TaskExecutorService {
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private void ackDropBoxProcessResult(ReceiveMessageDto receiveMessageDto) {
|
|
|
|
|
AckDto ackDto = AckDto.builder().build();
|
|
|
|
|
private void ackDropBoxProcessResult(ReceiveMessageDto receiveMessageDto, String processMessage) {
|
|
|
|
|
AckDto.ResultType resultType = EnumUtils.getEnum(AckDto.ResultType.class, receiveMessageDto.getProcessStatus().toString());
|
|
|
|
|
AckDto ackDto = AckDto.builder().result(resultType).messageUuid(receiveMessageDto.getMessageUuid()).resultText(processMessage).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);
|
|
|
|
|
String url = "https://" + knownAgent.getHostName() + ":" + knownAgent.getListenPort() + "/telegram";
|
|
|
|
|
String response = restTemplate.postForObject(url, bodyEntity, String.class);
|
|
|
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
|
AckDto responseAckDto = null;
|
|
|
|
|
try {
|
|
|
|
|
AckDto responseAckDto = objectMapper.readValue(response, AckDto.class);
|
|
|
|
|
// 수신 메시지 처리 완료. 메시지 삭제
|
|
|
|
|
responseAckDto = objectMapper.readValue(response, AckDto.class);
|
|
|
|
|
log.info("message processing is done. {}", receiveMessageDto.getMessageUuid());
|
|
|
|
|
log.info("response {}", responseAckDto);
|
|
|
|
|
} catch (JsonProcessingException e) {
|
|
|
|
|
//TODO 처리 결과 Ack 파싱 실패. 로그 남기고 메시지 삭제
|
|
|
|
|
throw new RuntimeException(e);
|
|
|
|
|
log.error("Exception after sending ACK. messageUuid: {} ", receiveMessageDto.getMessageUuid());
|
|
|
|
|
log.error("{}", e, e);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
