Commit @229f464ec3ae15ad121aa0a0010bddf62d7ba779

장동신 02-17

누락건 로직 수정, IP체크, 중복건 로직 추가

@229f464ec3ae15ad121aa0a0010bddf62d7ba779

9ea4d76

229f464

src/main/java/com/munjaon/server/config/RunnerConfiguration.java

--- src/main/java/com/munjaon/server/config/RunnerConfiguration.java

+++ src/main/java/com/munjaon/server/config/RunnerConfiguration.java


         try {
             String serviceName = "SMS_COLLECTOR";
             String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
+            String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
             int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
+            int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
 //            CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
-            CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
+            CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
 
             ShutdownService shutdownService = new ShutdownService(collectServer);
             Runtime.getRuntime().addShutdownHook(shutdownService);

         try {
             String serviceName = "LMS_COLLECTOR";
             String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
+            String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
             int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
+            int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
 //            CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
-            CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
+            CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
 
             ShutdownService shutdownService = new ShutdownService(collectServer);
             Runtime.getRuntime().addShutdownHook(shutdownService);

         try {
             String serviceName = "MMS_COLLECTOR";
             String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
+            String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
             int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
+            int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
 //            CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
-            CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
+            CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
 
             ShutdownService shutdownService = new ShutdownService(collectServer);
             Runtime.getRuntime().addShutdownHook(shutdownService);

         try {
             String serviceName = "KAT_COLLECTOR";
             String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
+            String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
             int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
+            int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
 //            CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
-            CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
+            CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
 
             ShutdownService shutdownService = new ShutdownService(collectServer);
             Runtime.getRuntime().addShutdownHook(shutdownService);

         try {
             String serviceName = "KFT_COLLECTOR";
             String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
+            String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
             int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
+            int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
 //            CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
-            CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
+            CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
 
             ShutdownService shutdownService = new ShutdownService(collectServer);
             Runtime.getRuntime().addShutdownHook(shutdownService);

229f464

src/main/java/com/munjaon/server/queue/pool/KeyWriteQueue.java (added)

+++ src/main/java/com/munjaon/server/queue/pool/KeyWriteQueue.java

...	...	@@ -0,0 +1,216 @@
	1	+package com.munjaon.server.queue.pool;
	2	+
	3	+import com.munjaon.server.queue.config.QueueConstants;
	4	+import com.munjaon.server.server.packet.common.Packet;
	5	+import com.munjaon.server.util.FileUtil;
	6	+import lombok.Getter;
	7	+
	8	+import java.io.File;
	9	+import java.io.IOException;
	10	+import java.io.RandomAccessFile;
	11	+import java.nio.ByteBuffer;
	12	+import java.nio.channels.FileChannel;
	13	+
	14	+@Getter
	15	+public class KeyWriteQueue {
	16	+ private final static int HEAD_SIZE = 40;
	17	+ private final static int HEAD_TOTAL_COUNT_LOCATION = 0;
	18	+ private final static int HEAD_TOTAL_COUNT_SIZE = 20;
	19	+ private final static int HEAD_WRITE_POSITION_LOCATION = 20;
	20	+ private final static int HEAD_WRITE_POSITION_SIZE = 20;
	21	+ private final static int BODY_WRITE_KEY_SIZE = 20;
	22	+ /* Queue 변수 */
	23	+ private final String queuePath;
	24	+ private final String queueName;
	25	+ private final int MAX_QUEUE_SIZE;
	26	+
	27	+ private int writePosition = -1;
	28	+ private int totalCount = -1;
	29	+
	30	+ /** Queue Header Buffer */
	31	+ protected ByteBuffer headerBuffer = null;
	32	+ /** Header 에서 사용하는 변수 */
	33	+ protected byte[] headerArray = null;
	34	+ /** Queue File Channel */
	35	+ protected FileChannel channel = null;
	36	+ /** pushBuffer() 함수에서 사용하는 변수 */
	37	+ protected ByteBuffer dataBuffer = null;
	38	+ /** isExist() 함수에서 사용하는 변수 */
	39	+ protected ByteBuffer searchBuffer = null;
	40	+
	41	+ public KeyWriteQueue(String queuePath, String queueName, int maxQueueSize) {
	42	+ this.queuePath = System.getProperty("ROOTPATH") + File.separator + queuePath;
	43	+ this.queueName = queueName;
	44	+ MAX_QUEUE_SIZE = maxQueueSize;
	45	+ /* 큐초기화 */
	46	+ try {
	47	+ initQueue();
	48	+ } catch (Exception e) {}
	49	+ }
	50	+
	51	+ public boolean isExist(String key) throws Exception {
	52	+ if (this.totalCount == 0) {
	53	+ return false;
	54	+ }
	55	+
	56	+ boolean exists = false;
	57	+ for (int i = 0; i < this.totalCount; i++) {
	58	+ int position = this.HEAD_SIZE + (i * BODY_WRITE_KEY_SIZE);
	59	+ this.channel.position(position);
	60	+ /* 읽기용 데이터 버퍼 초기화 */
	61	+ initSearchBuffer();
	62	+ this.channel.read(searchBuffer);
	63	+
	64	+ byte[] byteArray = new byte[BODY_WRITE_KEY_SIZE];
	65	+ searchBuffer.position(0);
	66	+ searchBuffer.get(byteArray);
	67	+ String queueKey = QueueConstants.getString(byteArray);
	68	+
	69	+ if (queueKey.equals(key)) {
	70	+ exists = true;
	71	+ break;
	72	+ }
	73	+ }
	74	+
	75	+ return exists;
	76	+ }
	77	+
	78	+ public void pushKeyToBuffer(String key) throws Exception {
	79	+ if (key == null \|\| key.isEmpty()) {
	80	+ return;
	81	+ }
	82	+
	83	+ /* 1. dataBuffer 초기화 */
	84	+ initDataBuffer();
	85	+ /* 2. messageDto >> dataBuffer */
	86	+ this.dataBuffer.position(0);
	87	+ this.dataBuffer.put(key.getBytes(Packet.AGENT_CHARACTER_SET));
	88	+
	89	+ /* 3.1 Header 정보 다시 일기 */
	90	+ readHeader();
	91	+ if (this.dataBuffer != null) {
	92	+ int position = HEAD_SIZE + (BODY_WRITE_KEY_SIZE * this.writePosition);
	93	+ this.channel.position(position);
	94	+ this.dataBuffer.flip();
	95	+ this.channel.write(this.dataBuffer);
	96	+ /* 3.2 Push 카운터 증가 */
	97	+ this.writePosition = this.writePosition + 1;
	98	+ if (this.writePosition > this.MAX_QUEUE_SIZE) {
	99	+ this.writePosition = 0;
	100	+ }
	101	+ this.totalCount = this.totalCount + 1;
	102	+ if (this.totalCount > this.MAX_QUEUE_SIZE) {
	103	+ this.totalCount = this.MAX_QUEUE_SIZE;
	104	+ }
	105	+ /* 3.3 Header 정보 변경 */
	106	+ writeHeader();
	107	+ }
	108	+ }
	109	+
	110	+ public void initDataBuffer() {
	111	+ if (this.dataBuffer == null) {
	112	+ this.dataBuffer = ByteBuffer.allocateDirect(BODY_WRITE_KEY_SIZE);
	113	+ }
	114	+ this.dataBuffer.clear();
	115	+ for(int loopCnt = 0; loopCnt < BODY_WRITE_KEY_SIZE; loopCnt++){
	116	+ this.dataBuffer.put(QueueConstants.SET_DEFAULT_BYTE);
	117	+ }
	118	+ this.dataBuffer.position(0);
	119	+ }
	120	+
	121	+ public void initSearchBuffer() {
	122	+ if (this.searchBuffer == null) {
	123	+ this.searchBuffer = ByteBuffer.allocateDirect(BODY_WRITE_KEY_SIZE);
	124	+ }
	125	+ this.searchBuffer.clear();
	126	+ for(int loopCnt = 0; loopCnt < BODY_WRITE_KEY_SIZE; loopCnt++){
	127	+ this.searchBuffer.put(QueueConstants.SET_DEFAULT_BYTE);
	128	+ }
	129	+ this.searchBuffer.position(0);
	130	+ }
	131	+
	132	+ private void initQueue() throws Exception {
	133	+ this.headerBuffer = ByteBuffer.allocateDirect(HEAD_SIZE);
	134	+ try{
	135	+ /* 1. 경로 체크 및 생성 */
	136	+ FileUtil.mkdirs(this.queuePath);
	137	+ /* 2. 파일생성 및 읽기 */
	138	+ File file = new File(this.queuePath + File.separator + queueName + ".queue");
	139	+ this.channel = new RandomAccessFile(file, "rw").getChannel();
	140	+
	141	+ if (file.length() == 0) {
	142	+ // Push 및 Pop 카운트 초기화
	143	+ this.writePosition = 0;
	144	+ this.totalCount = 0;
	145	+ // 헤더 초기화
	146	+ writeHeader();
	147	+ } else {
	148	+ readHeader();
	149	+ }
	150	+// backupQueue();
	151	+ } catch(Exception e) {
	152	+ throw e;
	153	+ } finally {
	154	+ //lock.release();
	155	+ }
	156	+ }
	157	+ protected void readHeader() throws Exception {
	158	+ try {
	159	+ initHeaderBuffer();
	160	+ this.channel.position(0);
	161	+ this.channel.read(this.headerBuffer);
	162	+ this.headerArray = new byte[HEAD_TOTAL_COUNT_SIZE];
	163	+ // 생성날짜 가져오기 - 생성날짜(10) / 읽은카운트(10) / 쓴카운트(10)
	164	+ this.headerBuffer.position(HEAD_TOTAL_COUNT_LOCATION);
	165	+ this.headerBuffer.get(this.headerArray);
	166	+ this.totalCount = Integer.parseInt(QueueConstants.getString(this.headerArray));
	167	+ // 쓴 카운트 가져오기
	168	+ this.headerArray = new byte[HEAD_WRITE_POSITION_SIZE];
	169	+ this.headerBuffer.position(HEAD_WRITE_POSITION_LOCATION);
	170	+ this.headerBuffer.get(this.headerArray);
	171	+ this.writePosition = Integer.parseInt(QueueConstants.getString(this.headerArray));
	172	+ } catch(Exception e) {
	173	+ throw e;
	174	+ }
	175	+ }
	176	+
	177	+ protected void writeHeader() throws Exception {
	178	+ try {
	179	+ initHeaderBuffer();
	180	+ this.channel.position(0);
	181	+ this.headerBuffer.put(Integer.toString(this.totalCount).getBytes(Packet.AGENT_CHARACTER_SET));
	182	+ this.headerBuffer.position(HEAD_WRITE_POSITION_LOCATION);
	183	+ this.headerBuffer.put(Integer.toString(this.writePosition).getBytes(Packet.AGENT_CHARACTER_SET));
	184	+ this.headerBuffer.flip();
	185	+ this.channel.write(this.headerBuffer);
	186	+ } catch(Exception e) {
	187	+ throw e;
	188	+ }
	189	+ }
	190	+
	191	+ protected void initHeaderBuffer(){
	192	+ this.headerBuffer.clear();
	193	+ for(int loopCnt = 0; loopCnt < HEAD_SIZE; loopCnt++){
	194	+ this.headerBuffer.put(QueueConstants.SET_DEFAULT_BYTE);
	195	+ }
	196	+ this.headerBuffer.position(0);
	197	+ }
	198	+
	199	+ public void close() throws IOException {
	200	+ try {
	201	+ if (isOpen()) {
	202	+ channel.close();
	203	+ }
	204	+ } catch(IOException e) {
	205	+ throw e;
	206	+ }
	207	+ }
	208	+
	209	+ protected boolean isOpen() {
	210	+ if (this.channel == null) {
	211	+ return false;
	212	+ }
	213	+
	214	+ return this.channel.isOpen();
	215	+ }
	216	+}

9ea4d76

229f464

src/main/java/com/munjaon/server/queue/pool/QueuePool.java

--- src/main/java/com/munjaon/server/queue/pool/QueuePool.java

+++ src/main/java/com/munjaon/server/queue/pool/QueuePool.java


                 queue = queuePool.get(loopCnt);
                 if(name.equals(queue.getQueueName())){
                     queuePool.remove(loopCnt);
-                    System.out.println("[" + queue.getQueueInfo().getServiceType() + " Queue] [" + queue.getQueueName() + " is Removed]");
+                    log.info("[" + queue.getQueueInfo().getServiceType() + " Queue] [" + queue.getQueueName() + " is Removed]");
                     break;
                 }
             }

         synchronized(lockMonitor){
             if (queue != null){
                 queuePool.addLast(queue);
+                log.info("[" + queue.getQueueInfo().getServiceType() + " Queue] [" + queue.getQueueName() + " is Added]");
                 lockMonitor.notifyAll();
             }
         }

9ea4d76

229f464

src/main/java/com/munjaon/server/server/dto/ConnectUserDto.java

--- src/main/java/com/munjaon/server/server/dto/ConnectUserDto.java

+++ src/main/java/com/munjaon/server/server/dto/ConnectUserDto.java


 package com.munjaon.server.server.dto;
 
 import com.munjaon.server.cache.dto.MemberDto;
+import com.munjaon.server.queue.pool.KeyWriteQueue;
 import com.munjaon.server.server.config.ServerConfig;
 import lombok.Builder;
 import lombok.Getter;

     private int command;
     /* 요청을 처리중인지 여부 */
     private boolean isRunningMode;
+    /* MSGID 중복큐 경로 */
+    private String msgKeyPath;
+    /* MSGID 중복큐 사이즈 */
+    private int msgIdQueueSize;
+    /* MSG ID 중복체크 큐 */
+    private KeyWriteQueue keyWriteQueue;
+
 
     private MemberDto memberDto;
 

9ea4d76

229f464

src/main/java/com/munjaon/server/server/service/CollectServer.java

--- src/main/java/com/munjaon/server/server/service/CollectServer.java

+++ src/main/java/com/munjaon/server/server/service/CollectServer.java


 package com.munjaon.server.server.service;
 
 import com.munjaon.server.queue.enums.QueueTypeWorker;
+import com.munjaon.server.queue.pool.KeyWriteQueue;
 import com.munjaon.server.server.config.ServerConfig;
 import com.munjaon.server.server.dto.ConnectUserDto;
 import com.munjaon.server.server.queue.CollectUserQueue;

 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.net.SocketAddress;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.ServerSocketChannel;

 
     private Selector selector;
     private final String serviceType;
+    private final String msgKeyPath;
+    private final int msgIdQueueSize;
 
     private long USER_STATUS_LAST_CHECK_TIME = System.currentTimeMillis();
 
-    public CollectServer(String serviceName, String serviceType, int port) {
+    public CollectServer(String serviceName, String serviceType, int port, String msgKeyPath, int msgIdQueueSize) {
         super(serviceName);
         this.listenAddress = new InetSocketAddress(port);
         this.serviceType = serviceType;
+        this.msgKeyPath = msgKeyPath;
+        this.msgIdQueueSize = msgIdQueueSize;
     }
 
     @Override

             /* 소켓 취득 */
             Socket socket = channel.socket();
             InetAddress inetAddress = socket.getInetAddress();
+            SocketAddress socketAddress = socket.getRemoteSocketAddress();
+
             saveLog("[CLIENT CONNECTION IP : " + inetAddress.getHostAddress() + "]");
+            saveLog("[CLIENT CONNECTION IP : " + socketAddress.toString() + "]");
             // Socket 채널을 channel에 수신 등록한다
-            channel.register(selector, SelectionKey.OP_READ, ConnectUserDto.builder().serviceType(this.serviceType).lastTrafficTime(System.currentTimeMillis()).remoteIP(inetAddress.getHostAddress()).build());
+            channel.register(selector, SelectionKey.OP_READ, ConnectUserDto.builder().serviceType(this.serviceType).msgKeyPath(this.msgKeyPath).msgIdQueueSize(this.msgIdQueueSize).lastTrafficTime(System.currentTimeMillis()).remoteIP(inetAddress.getHostAddress()).build());
         } catch (Exception e) {
             saveLog(e.toString());
             throw new RuntimeException(e);

             SocketChannel channel = (SocketChannel) key.channel();
             ConnectUserDto userDto = (ConnectUserDto) key.attachment();
             if (userDto != null && userDto.getUserId() != null) {
+                /* MSG ID 중복체크 큐 해제 */
+                KeyWriteQueue keyWriteQueue = userDto.getKeyWriteQueue();
+                if (keyWriteQueue != null) {
+                    keyWriteQueue.close();
+                }
                 saveLog("[CLIENT USER IS DISCONNECT : " + userDto.toString() + "]");
                 collectUserQueue.removeUser(this.serviceType, userDto.getUserId());
                 key.attach(null);

9ea4d76

229f464

src/main/java/com/munjaon/server/server/service/ReportServer.java

--- src/main/java/com/munjaon/server/server/service/ReportServer.java

+++ src/main/java/com/munjaon/server/server/service/ReportServer.java


 import org.json.simple.JSONObject;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketAddress;

             channel.configureBlocking(false);
             /* 소켓 취득 */
             Socket socket = channel.socket();
-            SocketAddress remoteAddr = socket.getRemoteSocketAddress();
-            saveLog("Connected to: " + remoteAddr);
+            InetAddress inetAddress = socket.getInetAddress();
+            SocketAddress socketAddress = socket.getRemoteSocketAddress();
+
+            saveLog("[CLIENT CONNECTION IP : " + inetAddress.getHostAddress() + "]");
+            saveLog("[CLIENT CONNECTION IP : " + socketAddress.toString() + "]");
             // Socket 채널을 channel에 수신 등록한다
-            channel.register(selector, SelectionKey.OP_READ, ReportUserDto.builder().lastTrafficTime(System.currentTimeMillis()).remoteIP(remoteAddr.toString()).queuePath(System.getProperty("ROOTPATH") + getProp("QUEUE_PATH")).build());
+            channel.register(selector, SelectionKey.OP_READ, ReportUserDto.builder().lastTrafficTime(System.currentTimeMillis()).remoteIP(inetAddress.getHostAddress()).queuePath(System.getProperty("ROOTPATH") + getProp("QUEUE_PATH")).build());
         } catch (Exception e) {
             saveLog(e.toString());
             throw new RuntimeException(e);

9ea4d76

229f464

src/main/java/com/munjaon/server/server/task/CollectServerTask.java

--- src/main/java/com/munjaon/server/server/task/CollectServerTask.java

+++ src/main/java/com/munjaon/server/server/task/CollectServerTask.java


 import com.munjaon.server.cache.service.MemberService;
 import com.munjaon.server.queue.dto.BasicMessageDto;
 import com.munjaon.server.queue.enums.QueueTypeWorker;
+import com.munjaon.server.queue.pool.KeyWriteQueue;
 import com.munjaon.server.server.config.ServerConfig;
 import com.munjaon.server.server.dto.ConnectUserDto;
 import com.munjaon.server.server.packet.common.*;

             saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
             QueueTypeWorker worker = QueueTypeWorker.find("SMS");
             if (worker != null) {
-                worker.pushQueue(messageDto);
-                channel.write(SmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
-                connectUserDto.updateLastTrafficTime();
+                boolean isQueueYn = true;
+                KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
+                try {
+                    if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
+                        saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " | DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
+                    } else {
+                        worker.pushQueue(messageDto);
+                    }
+                } catch (Exception e) {
+                    isQueueYn = false;
+                    saveSystemLog(e);
+                }
+                if (isQueueYn) {
+                    channel.write(SmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
+                    connectUserDto.updateLastTrafficTime();
+                    /* MSG_ID 큐에 저장 */
+                    keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
+                }
             }
         } catch (Exception e) {
             saveLog(e);

             saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
             QueueTypeWorker worker = QueueTypeWorker.find("LMS");
             if (worker != null) {
-                worker.pushQueue(messageDto);
-                channel.write(LmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
-                connectUserDto.updateLastTrafficTime();
+                boolean isQueueYn = true;
+                KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
+                try {
+                    if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
+                        saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " | DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
+                    } else {
+                        worker.pushQueue(messageDto);
+                    }
+
+                } catch (Exception e) {
+                    isQueueYn = false;
+                    saveSystemLog(e);
+                }
+                if (isQueueYn) {
+                    channel.write(LmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
+                    connectUserDto.updateLastTrafficTime();
+                    /* MSG_ID 큐에 저장 */
+                    keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
+                }
             }
         } catch (Exception e) {
             saveLog(e);

             saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
             QueueTypeWorker worker = QueueTypeWorker.find("MMS");
             if (worker != null) {
-                worker.pushQueue(messageDto);
-                channel.write(MmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
-                connectUserDto.updateLastTrafficTime();
+                boolean isQueueYn = true;
+                KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
+                try {
+                    if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
+                        saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " | DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
+                    } else {
+                        worker.pushQueue(messageDto);
+                    }
+
+                } catch (Exception e) {
+                    isQueueYn = false;
+                    saveSystemLog(e);
+                }
+                if (isQueueYn) {
+                    channel.write(MmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
+                    connectUserDto.updateLastTrafficTime();
+                    /* MSG_ID 큐에 저장 */
+                    keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
+                }
             }
         } catch (Exception e) {
             saveLog(e);

             saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
             QueueTypeWorker worker = QueueTypeWorker.find("KAT");
             if (worker != null) {
-                worker.pushQueue(messageDto);
-                channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
-                connectUserDto.updateLastTrafficTime();
+                boolean isQueueYn = true;
+                KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
+                try {
+                    if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
+                        saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " | DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
+                    } else {
+                        worker.pushQueue(messageDto);
+                    }
+
+                } catch (Exception e) {
+                    isQueueYn = false;
+                    saveSystemLog(e);
+                }
+                if (isQueueYn) {
+                    channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
+                    connectUserDto.updateLastTrafficTime();
+                    /* MSG_ID 큐에 저장 */
+                    keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
+                }
             } else {
                 saveSystemLog("worker is null");
             }
         } catch (Exception e) {
             saveLog(e);
             throw e;
-        }
-    }
-
-    private void recvKatDeliver_bak(SocketChannel channel, ByteBuffer headBuffer) throws IOException {
-        try {
-            ByteBuffer bodyBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
-            channel.read(bodyBuffer);
-            ByteBuffer deliverBuffer = ByteBuffer.allocate(Header.HEADER_LENGTH + KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
-            Packet.mergeBuffers(deliverBuffer, headBuffer, bodyBuffer);
-
-            BasicMessageDto messageDto = recvCommonMessage(deliverBuffer);
-            messageDto.setUserSubject(KakaoMessage.getSubjectForDeliver(deliverBuffer));
-            messageDto.setUserMessage(KakaoMessage.getMessageForDeliver(deliverBuffer));
-            messageDto.setKakaoSenderKey(KakaoMessage.getKakaoSenderKeyForDeliver(deliverBuffer));
-            messageDto.setKakaoTemplateCode(KakaoMessage.getKakaoTemplateCodeForDeliver(deliverBuffer));
-
-            String jsonPath = System.getProperty("ROOTPATH") + File.separator + "kakaofile";
-            jsonPath = jsonPath + File.separator + MessageUtil.getDate() + File.separator + SerialNoUtil.getSerialNo();
-            FileUtil.mkdirs(jsonPath);
-
-            ByteBuffer fileHeadBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_JSON_FILENAME_LENGTH + KakaoMessage.DELIVER_JSON_FILESIZE_LENGTH);
-            channel.read(fileHeadBuffer);
-
-            String fileName = KakaoMessage.getFileNameForDeliver(fileHeadBuffer);
-            String fileSize = KakaoMessage.getFileSizeForDeliver(fileHeadBuffer);
-
-            ByteBuffer fileBuffer = ByteBuffer.allocate(Integer.parseInt(fileSize));
-            channel.read(fileBuffer);
-            fileBuffer.flip();
-            JobFileFactory.saveFileForByteBuffer(jsonPath, fileName, fileBuffer);
-
-            messageDto.setKakaoJsonFile(jsonPath + File.separator + fileName);
-            saveLog(printTaskLog() + "[KAT JSON] [File : " + fileName + ", Size : " + fileSize + "]");
-
-            /* 사용자 단가, 발송망 설정 */
-            MemberDto savedMemberDto = null;
-            if (this.connectUserDto != null) {
-                savedMemberDto = this.connectUserDto.getMemberDto();
-            }
-            if (savedMemberDto != null) {
-                messageDto.setRouterSeq(savedMemberDto.getKakaoAtAgentCode());
-                messageDto.setUnitCost(String.valueOf(savedMemberDto.getKakaoAtPrice()));
-            }
-
-            saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
-            QueueTypeWorker worker = QueueTypeWorker.find("KAT");
-            if (worker != null) {
-                worker.pushQueue(messageDto);
-                channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
-                connectUserDto.updateLastTrafficTime();
-            } else {
-                saveSystemLog("worker is null");
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
         }
     }
 

             saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
             QueueTypeWorker worker = QueueTypeWorker.find("KFT");
             if (worker != null) {
-                worker.pushQueue(messageDto);
-                channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
-                connectUserDto.updateLastTrafficTime();
+                boolean isQueueYn = true;
+                KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
+                try {
+                    if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
+                        saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " | DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
+                    } else {
+                        worker.pushQueue(messageDto);
+                    }
+                } catch (Exception e) {
+                    isQueueYn = false;
+                    saveSystemLog(e);
+                }
+                if (isQueueYn) {
+                    channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
+                    connectUserDto.updateLastTrafficTime();
+                    /* MSG_ID 큐에 저장 */
+                    keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
+                }
             }
         } catch (Exception e) {
             saveLog(e);
             throw e;
-        }
-    }
-
-    private void recvKftDeliver_bak(SocketChannel channel, ByteBuffer headBuffer) throws IOException {
-        try {
-            ByteBuffer bodyBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
-            channel.read(bodyBuffer);
-            ByteBuffer deliverBuffer = ByteBuffer.allocate(Header.HEADER_LENGTH + KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
-            Packet.mergeBuffers(deliverBuffer, headBuffer, bodyBuffer);
-
-            BasicMessageDto messageDto = recvCommonMessage(deliverBuffer);
-            messageDto.setUserSubject(KakaoMessage.getSubjectForDeliver(deliverBuffer));
-            messageDto.setUserMessage(KakaoMessage.getMessageForDeliver(deliverBuffer));
-            messageDto.setKakaoSenderKey(KakaoMessage.getKakaoSenderKeyForDeliver(deliverBuffer));
-            messageDto.setKakaoTemplateCode(KakaoMessage.getKakaoTemplateCodeForDeliver(deliverBuffer));
-
-            String jsonPath = System.getProperty("ROOTPATH") + File.separator + "kakaofile";
-            jsonPath = jsonPath + File.separator + MessageUtil.getDate() + File.separator + SerialNoUtil.getSerialNo();
-            FileUtil.mkdirs(jsonPath);
-
-            ByteBuffer fileHeadBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_JSON_FILENAME_LENGTH + KakaoMessage.DELIVER_JSON_FILESIZE_LENGTH);
-            channel.read(fileHeadBuffer);
-
-            String fileName = KakaoMessage.getFileNameForDeliver(fileHeadBuffer);
-            String fileSize = KakaoMessage.getFileSizeForDeliver(fileHeadBuffer);
-
-            ByteBuffer fileBuffer = ByteBuffer.allocate(Integer.parseInt(fileSize));
-            channel.read(fileBuffer);
-            fileBuffer.flip();
-            JobFileFactory.saveFileForByteBuffer(jsonPath, fileName, fileBuffer);
-
-            messageDto.setKakaoJsonFile(jsonPath + File.separator + fileName);
-            saveLog(printTaskLog() + "[KFT JSON] [File : " + fileName + ", Size : " + fileSize + "]");
-
-            /* 사용자 단가, 발송망 설정 */
-            MemberDto savedMemberDto = null;
-            if (this.connectUserDto != null) {
-                savedMemberDto = this.connectUserDto.getMemberDto();
-            }
-            if (savedMemberDto != null) {
-                messageDto.setRouterSeq(savedMemberDto.getKakaoFtAgentCode());
-                messageDto.setUnitCost(String.valueOf(savedMemberDto.getKakaoFtPrice()));
-            }
-
-            saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
-            QueueTypeWorker worker = QueueTypeWorker.find("KFT");
-            if (worker != null) {
-                worker.pushQueue(messageDto);
-                channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
-                connectUserDto.updateLastTrafficTime();
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
         }
     }
 

                     saveLog(printTaskLog() + "[REMOTE IP : " + connectUserDto.getRemoteIP() + "]");
                     saveLog(printTaskLog() + "[ALLOW IP BASIC : " + memberDto.getAllowIpBasic() + "]");
                     saveLog(printTaskLog() + "[ALLOW IP EXTEND : " + memberDto.getAllowIpExtend() + "]");
-                    if (memberDto.getAllowIpBasic() != null && connectUserDto.getRemoteIP().indexOf(memberDto.getAllowIpBasic()) >= 0) {
+                    if (memberDto.getAllowIpBasic() != null && connectUserDto.getRemoteIP().trim().equals(memberDto.getAllowIpBasic())) {
                         isPermit = true;
                     }
-                    if (memberDto.getAllowIpExtend() != null && connectUserDto.getRemoteIP().indexOf(memberDto.getAllowIpExtend()) >= 0) {
+                    if (memberDto.getAllowIpExtend() != null && connectUserDto.getRemoteIP().trim().equals(memberDto.getAllowIpExtend())) {
                         isPermit = true;
                     }
                 } else {

                 connectUserDto.setMemberDto(memberDto);
                 /* 세션통신 시간 업데이트 */
                 connectUserDto.updateLastTrafficTime();
+                /* MSGID 중복체크큐 초기화 */
+                KeyWriteQueue keyWriteQueue = new KeyWriteQueue(connectUserDto.getMsgKeyPath(), connectUserDto.getUserId(), connectUserDto.getMsgIdQueueSize());
+                connectUserDto.setKeyWriteQueue(keyWriteQueue);
             }
         } catch (Exception e) {
             resultCode = "10";

             if (connectUserDto != null) {
                 if (connectUserDto.getUserId() != null) {
                     collectUserQueue.removeUser(connectUserDto.getServiceType(), connectUserDto.getUserId());
+                    /* MSG ID 중복체크 큐 해제 */
+                    KeyWriteQueue keyWriteQueue = connectUserDto.getKeyWriteQueue();
+                    if (keyWriteQueue != null) {
+                        keyWriteQueue.close();
+                    }
                     /* 모니터링 로그 */
                     HealthCheckServer.saveMonitorLog("[COLLECT SERVER][SERVICE TYPE : " + this.serviceType + "][ID : " + connectUserDto.getUserId() + "][EXPIRE CONNECT USER]");
                 }

9ea4d76

229f464

src/main/resources/sqlmap/sms_sql.xml

--- src/main/resources/sqlmap/sms_sql.xml

+++ src/main/resources/sqlmap/sms_sql.xml


         <foreach separator="," item="item" collection="list">
             (#{item.id}, #{item.msgGroupID}, #{item.userId}, #{item.userMsgID}, #{item.routerSeq}, 0, NOW(), #{item.userReceiver}, #{item.userSender}, NULL, #{item.userMessage}, '4' )
         </foreach>
+        ON DUPLICATE KEY UPDATE REQ_DATE = VALUES(REQ_DATE)
     </insert>
 
     <insert id="insertGroupForList" parameterType="java.util.List">

Add a comment

Open 0
Closed 0

List

...	...	@@ -197,9 +197,11 @@
197	197	try {
198	198	String serviceName = "SMS_COLLECTOR";
199	199	String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
	200	+ String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
200	201	int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
	202	+ int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
201	203	// CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
202		- CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
	204	+ CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
203	205
204	206	ShutdownService shutdownService = new ShutdownService(collectServer);
205	207	Runtime.getRuntime().addShutdownHook(shutdownService);
...	...	@@ -217,9 +219,11 @@
217	219	try {
218	220	String serviceName = "LMS_COLLECTOR";
219	221	String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
	222	+ String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
220	223	int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
	224	+ int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
221	225	// CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
222		- CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
	226	+ CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
223	227
224	228	ShutdownService shutdownService = new ShutdownService(collectServer);
225	229	Runtime.getRuntime().addShutdownHook(shutdownService);
...	...	@@ -237,9 +241,11 @@
237	241	try {
238	242	String serviceName = "MMS_COLLECTOR";
239	243	String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
	244	+ String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
240	245	int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
	246	+ int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
241	247	// CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
242		- CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
	248	+ CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
243	249
244	250	ShutdownService shutdownService = new ShutdownService(collectServer);
245	251	Runtime.getRuntime().addShutdownHook(shutdownService);
...	...	@@ -257,9 +263,11 @@
257	263	try {
258	264	String serviceName = "KAT_COLLECTOR";
259	265	String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
	266	+ String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
260	267	int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
	268	+ int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
261	269	// CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
262		- CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
	270	+ CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
263	271
264	272	ShutdownService shutdownService = new ShutdownService(collectServer);
265	273	Runtime.getRuntime().addShutdownHook(shutdownService);
...	...	@@ -277,9 +285,11 @@
277	285	try {
278	286	String serviceName = "KFT_COLLECTOR";
279	287	String serviceType = serverConfig.getString(serviceName + ".SERVICE_TYPE");
	288	+ String msgKeyPath = serverConfig.getString(serviceName + ".MSGID_QUEUE");
280	289	int port = serverConfig.getInt(serviceName + ".SERVICE_PORT");
	290	+ int msgIdQueueSize = serverConfig.getInt(serviceName + ".MSGID_QUEUE_SIZE");
281	291	// CollectBackServerService collectServerService = new CollectBackServerService(serviceName, serviceType, port);
282		- CollectServer collectServer = new CollectServer(serviceName, serviceType, port);
	292	+ CollectServer collectServer = new CollectServer(serviceName, serviceType, port, msgKeyPath, msgIdQueueSize);
283	293
284	294	ShutdownService shutdownService = new ShutdownService(collectServer);
285	295	Runtime.getRuntime().addShutdownHook(shutdownService);

...	...	@@ -43,7 +43,7 @@
43	43	queue = queuePool.get(loopCnt);
44	44	if(name.equals(queue.getQueueName())){
45	45	queuePool.remove(loopCnt);
46		- System.out.println("[" + queue.getQueueInfo().getServiceType() + " Queue] [" + queue.getQueueName() + " is Removed]");
	46	+ log.info("[" + queue.getQueueInfo().getServiceType() + " Queue] [" + queue.getQueueName() + " is Removed]");
47	47	break;
48	48	}
49	49	}
...	...	@@ -55,6 +55,7 @@
55	55	synchronized(lockMonitor){
56	56	if (queue != null){
57	57	queuePool.addLast(queue);
	58	+ log.info("[" + queue.getQueueInfo().getServiceType() + " Queue] [" + queue.getQueueName() + " is Added]");
58	59	lockMonitor.notifyAll();
59	60	}
60	61	}

...	...	@@ -1,6 +1,7 @@
1	1	package com.munjaon.server.server.dto;
2	2
3	3	import com.munjaon.server.cache.dto.MemberDto;
	4	+import com.munjaon.server.queue.pool.KeyWriteQueue;
4	5	import com.munjaon.server.server.config.ServerConfig;
5	6	import lombok.Builder;
6	7	import lombok.Getter;
...	...	@@ -28,6 +29,13 @@
28	29	private int command;
29	30	/* 요청을 처리중인지 여부 */
30	31	private boolean isRunningMode;
	32	+ /* MSGID 중복큐 경로 */
	33	+ private String msgKeyPath;
	34	+ /* MSGID 중복큐 사이즈 */
	35	+ private int msgIdQueueSize;
	36	+ /* MSG ID 중복체크 큐 */
	37	+ private KeyWriteQueue keyWriteQueue;
	38	+
31	39
32	40	private MemberDto memberDto;
33	41

...	...	@@ -1,6 +1,7 @@
1	1	package com.munjaon.server.server.service;
2	2
3	3	import com.munjaon.server.queue.enums.QueueTypeWorker;
	4	+import com.munjaon.server.queue.pool.KeyWriteQueue;
4	5	import com.munjaon.server.server.config.ServerConfig;
5	6	import com.munjaon.server.server.dto.ConnectUserDto;
6	7	import com.munjaon.server.server.queue.CollectUserQueue;
...	...	@@ -12,6 +13,7 @@
12	13	import java.net.InetAddress;
13	14	import java.net.InetSocketAddress;
14	15	import java.net.Socket;
	16	+import java.net.SocketAddress;
15	17	import java.nio.channels.SelectionKey;
16	18	import java.nio.channels.Selector;
17	19	import java.nio.channels.ServerSocketChannel;
...	...	@@ -24,13 +26,17 @@
24	26
25	27	private Selector selector;
26	28	private final String serviceType;
	29	+ private final String msgKeyPath;
	30	+ private final int msgIdQueueSize;
27	31
28	32	private long USER_STATUS_LAST_CHECK_TIME = System.currentTimeMillis();
29	33
30		- public CollectServer(String serviceName, String serviceType, int port) {
	34	+ public CollectServer(String serviceName, String serviceType, int port, String msgKeyPath, int msgIdQueueSize) {
31	35	super(serviceName);
32	36	this.listenAddress = new InetSocketAddress(port);
33	37	this.serviceType = serviceType;
	38	+ this.msgKeyPath = msgKeyPath;
	39	+ this.msgIdQueueSize = msgIdQueueSize;
34	40	}
35	41
36	42	@Override
...	...	@@ -174,9 +180,12 @@
174	180	/* 소켓 취득 */
175	181	Socket socket = channel.socket();
176	182	InetAddress inetAddress = socket.getInetAddress();
	183	+ SocketAddress socketAddress = socket.getRemoteSocketAddress();
	184	+
177	185	saveLog("[CLIENT CONNECTION IP : " + inetAddress.getHostAddress() + "]");
	186	+ saveLog("[CLIENT CONNECTION IP : " + socketAddress.toString() + "]");
178	187	// Socket 채널을 channel에 수신 등록한다
179		- channel.register(selector, SelectionKey.OP_READ, ConnectUserDto.builder().serviceType(this.serviceType).lastTrafficTime(System.currentTimeMillis()).remoteIP(inetAddress.getHostAddress()).build());
	188	+ channel.register(selector, SelectionKey.OP_READ, ConnectUserDto.builder().serviceType(this.serviceType).msgKeyPath(this.msgKeyPath).msgIdQueueSize(this.msgIdQueueSize).lastTrafficTime(System.currentTimeMillis()).remoteIP(inetAddress.getHostAddress()).build());
180	189	} catch (Exception e) {
181	190	saveLog(e.toString());
182	191	throw new RuntimeException(e);
...	...	@@ -191,6 +200,11 @@
191	200	SocketChannel channel = (SocketChannel) key.channel();
192	201	ConnectUserDto userDto = (ConnectUserDto) key.attachment();
193	202	if (userDto != null && userDto.getUserId() != null) {
	203	+ /* MSG ID 중복체크 큐 해제 */
	204	+ KeyWriteQueue keyWriteQueue = userDto.getKeyWriteQueue();
	205	+ if (keyWriteQueue != null) {
	206	+ keyWriteQueue.close();
	207	+ }
194	208	saveLog("[CLIENT USER IS DISCONNECT : " + userDto.toString() + "]");
195	209	collectUserQueue.removeUser(this.serviceType, userDto.getUserId());
196	210	key.attach(null);

...	...	@@ -9,6 +9,7 @@
9	9	import org.json.simple.JSONObject;
10	10
11	11	import java.io.IOException;
	12	+import java.net.InetAddress;
12	13	import java.net.InetSocketAddress;
13	14	import java.net.Socket;
14	15	import java.net.SocketAddress;
...	...	@@ -173,10 +174,13 @@
173	174	channel.configureBlocking(false);
174	175	/* 소켓 취득 */
175	176	Socket socket = channel.socket();
176		- SocketAddress remoteAddr = socket.getRemoteSocketAddress();
177		- saveLog("Connected to: " + remoteAddr);
	177	+ InetAddress inetAddress = socket.getInetAddress();
	178	+ SocketAddress socketAddress = socket.getRemoteSocketAddress();
	179	+
	180	+ saveLog("[CLIENT CONNECTION IP : " + inetAddress.getHostAddress() + "]");
	181	+ saveLog("[CLIENT CONNECTION IP : " + socketAddress.toString() + "]");
178	182	// Socket 채널을 channel에 수신 등록한다
179		- channel.register(selector, SelectionKey.OP_READ, ReportUserDto.builder().lastTrafficTime(System.currentTimeMillis()).remoteIP(remoteAddr.toString()).queuePath(System.getProperty("ROOTPATH") + getProp("QUEUE_PATH")).build());
	183	+ channel.register(selector, SelectionKey.OP_READ, ReportUserDto.builder().lastTrafficTime(System.currentTimeMillis()).remoteIP(inetAddress.getHostAddress()).queuePath(System.getProperty("ROOTPATH") + getProp("QUEUE_PATH")).build());
180	184	} catch (Exception e) {
181	185	saveLog(e.toString());
182	186	throw new RuntimeException(e);

...	...	@@ -5,6 +5,7 @@
5	5	import com.munjaon.server.cache.service.MemberService;
6	6	import com.munjaon.server.queue.dto.BasicMessageDto;
7	7	import com.munjaon.server.queue.enums.QueueTypeWorker;
	8	+import com.munjaon.server.queue.pool.KeyWriteQueue;
8	9	import com.munjaon.server.server.config.ServerConfig;
9	10	import com.munjaon.server.server.dto.ConnectUserDto;
10	11	import com.munjaon.server.server.packet.common.*;
...	...	@@ -278,9 +279,24 @@
278	279	saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
279	280	QueueTypeWorker worker = QueueTypeWorker.find("SMS");
280	281	if (worker != null) {
281		- worker.pushQueue(messageDto);
282		- channel.write(SmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
283		- connectUserDto.updateLastTrafficTime();
	282	+ boolean isQueueYn = true;
	283	+ KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
	284	+ try {
	285	+ if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
	286	+ saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " \| DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
	287	+ } else {
	288	+ worker.pushQueue(messageDto);
	289	+ }
	290	+ } catch (Exception e) {
	291	+ isQueueYn = false;
	292	+ saveSystemLog(e);
	293	+ }
	294	+ if (isQueueYn) {
	295	+ channel.write(SmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
	296	+ connectUserDto.updateLastTrafficTime();
	297	+ /* MSG_ID 큐에 저장 */
	298	+ keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
	299	+ }
284	300	}
285	301	} catch (Exception e) {
286	302	saveLog(e);
...	...	@@ -312,9 +328,25 @@
312	328	saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
313	329	QueueTypeWorker worker = QueueTypeWorker.find("LMS");
314	330	if (worker != null) {
315		- worker.pushQueue(messageDto);
316		- channel.write(LmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
317		- connectUserDto.updateLastTrafficTime();
	331	+ boolean isQueueYn = true;
	332	+ KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
	333	+ try {
	334	+ if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
	335	+ saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " \| DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
	336	+ } else {
	337	+ worker.pushQueue(messageDto);
	338	+ }
	339	+
	340	+ } catch (Exception e) {
	341	+ isQueueYn = false;
	342	+ saveSystemLog(e);
	343	+ }
	344	+ if (isQueueYn) {
	345	+ channel.write(LmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
	346	+ connectUserDto.updateLastTrafficTime();
	347	+ /* MSG_ID 큐에 저장 */
	348	+ keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
	349	+ }
318	350	}
319	351	} catch (Exception e) {
320	352	saveLog(e);
...	...	@@ -434,9 +466,25 @@
434	466	saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
435	467	QueueTypeWorker worker = QueueTypeWorker.find("MMS");
436	468	if (worker != null) {
437		- worker.pushQueue(messageDto);
438		- channel.write(MmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
439		- connectUserDto.updateLastTrafficTime();
	469	+ boolean isQueueYn = true;
	470	+ KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
	471	+ try {
	472	+ if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
	473	+ saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " \| DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
	474	+ } else {
	475	+ worker.pushQueue(messageDto);
	476	+ }
	477	+
	478	+ } catch (Exception e) {
	479	+ isQueueYn = false;
	480	+ saveSystemLog(e);
	481	+ }
	482	+ if (isQueueYn) {
	483	+ channel.write(MmsMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
	484	+ connectUserDto.updateLastTrafficTime();
	485	+ /* MSG_ID 큐에 저장 */
	486	+ keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
	487	+ }
440	488	}
441	489	} catch (Exception e) {
442	490	saveLog(e);
...	...	@@ -514,70 +562,31 @@
514	562	saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
515	563	QueueTypeWorker worker = QueueTypeWorker.find("KAT");
516	564	if (worker != null) {
517		- worker.pushQueue(messageDto);
518		- channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
519		- connectUserDto.updateLastTrafficTime();
	565	+ boolean isQueueYn = true;
	566	+ KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
	567	+ try {
	568	+ if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
	569	+ saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " \| DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
	570	+ } else {
	571	+ worker.pushQueue(messageDto);
	572	+ }
	573	+
	574	+ } catch (Exception e) {
	575	+ isQueueYn = false;
	576	+ saveSystemLog(e);
	577	+ }
	578	+ if (isQueueYn) {
	579	+ channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
	580	+ connectUserDto.updateLastTrafficTime();
	581	+ /* MSG_ID 큐에 저장 */
	582	+ keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
	583	+ }
520	584	} else {
521	585	saveSystemLog("worker is null");
522	586	}
523	587	} catch (Exception e) {
524	588	saveLog(e);
525	589	throw e;
526		- }
527		- }
528		-
529		- private void recvKatDeliver_bak(SocketChannel channel, ByteBuffer headBuffer) throws IOException {
530		- try {
531		- ByteBuffer bodyBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
532		- channel.read(bodyBuffer);
533		- ByteBuffer deliverBuffer = ByteBuffer.allocate(Header.HEADER_LENGTH + KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
534		- Packet.mergeBuffers(deliverBuffer, headBuffer, bodyBuffer);
535		-
536		- BasicMessageDto messageDto = recvCommonMessage(deliverBuffer);
537		- messageDto.setUserSubject(KakaoMessage.getSubjectForDeliver(deliverBuffer));
538		- messageDto.setUserMessage(KakaoMessage.getMessageForDeliver(deliverBuffer));
539		- messageDto.setKakaoSenderKey(KakaoMessage.getKakaoSenderKeyForDeliver(deliverBuffer));
540		- messageDto.setKakaoTemplateCode(KakaoMessage.getKakaoTemplateCodeForDeliver(deliverBuffer));
541		-
542		- String jsonPath = System.getProperty("ROOTPATH") + File.separator + "kakaofile";
543		- jsonPath = jsonPath + File.separator + MessageUtil.getDate() + File.separator + SerialNoUtil.getSerialNo();
544		- FileUtil.mkdirs(jsonPath);
545		-
546		- ByteBuffer fileHeadBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_JSON_FILENAME_LENGTH + KakaoMessage.DELIVER_JSON_FILESIZE_LENGTH);
547		- channel.read(fileHeadBuffer);
548		-
549		- String fileName = KakaoMessage.getFileNameForDeliver(fileHeadBuffer);
550		- String fileSize = KakaoMessage.getFileSizeForDeliver(fileHeadBuffer);
551		-
552		- ByteBuffer fileBuffer = ByteBuffer.allocate(Integer.parseInt(fileSize));
553		- channel.read(fileBuffer);
554		- fileBuffer.flip();
555		- JobFileFactory.saveFileForByteBuffer(jsonPath, fileName, fileBuffer);
556		-
557		- messageDto.setKakaoJsonFile(jsonPath + File.separator + fileName);
558		- saveLog(printTaskLog() + "[KAT JSON] [File : " + fileName + ", Size : " + fileSize + "]");
559		-
560		- /* 사용자 단가, 발송망 설정 */
561		- MemberDto savedMemberDto = null;
562		- if (this.connectUserDto != null) {
563		- savedMemberDto = this.connectUserDto.getMemberDto();
564		- }
565		- if (savedMemberDto != null) {
566		- messageDto.setRouterSeq(savedMemberDto.getKakaoAtAgentCode());
567		- messageDto.setUnitCost(String.valueOf(savedMemberDto.getKakaoAtPrice()));
568		- }
569		-
570		- saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
571		- QueueTypeWorker worker = QueueTypeWorker.find("KAT");
572		- if (worker != null) {
573		- worker.pushQueue(messageDto);
574		- channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
575		- connectUserDto.updateLastTrafficTime();
576		- } else {
577		- saveSystemLog("worker is null");
578		- }
579		- } catch (Exception e) {
580		- e.printStackTrace();
581	590	}
582	591	}
583	592
...	...	@@ -651,66 +660,28 @@
651	660	saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
652	661	QueueTypeWorker worker = QueueTypeWorker.find("KFT");
653	662	if (worker != null) {
654		- worker.pushQueue(messageDto);
655		- channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
656		- connectUserDto.updateLastTrafficTime();
	663	+ boolean isQueueYn = true;
	664	+ KeyWriteQueue keyWriteQueue = this.connectUserDto.getKeyWriteQueue();
	665	+ try {
	666	+ if (keyWriteQueue.isExist(messageDto.getUserMsgID())) {
	667	+ saveSystemLog("USER_ID : " + this.connectUserDto.getUserId() + " \| DUPLICATED MSG_ID : " + messageDto.getUserMsgID());
	668	+ } else {
	669	+ worker.pushQueue(messageDto);
	670	+ }
	671	+ } catch (Exception e) {
	672	+ isQueueYn = false;
	673	+ saveSystemLog(e);
	674	+ }
	675	+ if (isQueueYn) {
	676	+ channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
	677	+ connectUserDto.updateLastTrafficTime();
	678	+ /* MSG_ID 큐에 저장 */
	679	+ keyWriteQueue.pushKeyToBuffer(messageDto.getUserMsgID());
	680	+ }
657	681	}
658	682	} catch (Exception e) {
659	683	saveLog(e);
660	684	throw e;
661		- }
662		- }
663		-
664		- private void recvKftDeliver_bak(SocketChannel channel, ByteBuffer headBuffer) throws IOException {
665		- try {
666		- ByteBuffer bodyBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
667		- channel.read(bodyBuffer);
668		- ByteBuffer deliverBuffer = ByteBuffer.allocate(Header.HEADER_LENGTH + KakaoMessage.DELIVER_KAKAO_BODY_LENGTH);
669		- Packet.mergeBuffers(deliverBuffer, headBuffer, bodyBuffer);
670		-
671		- BasicMessageDto messageDto = recvCommonMessage(deliverBuffer);
672		- messageDto.setUserSubject(KakaoMessage.getSubjectForDeliver(deliverBuffer));
673		- messageDto.setUserMessage(KakaoMessage.getMessageForDeliver(deliverBuffer));
674		- messageDto.setKakaoSenderKey(KakaoMessage.getKakaoSenderKeyForDeliver(deliverBuffer));
675		- messageDto.setKakaoTemplateCode(KakaoMessage.getKakaoTemplateCodeForDeliver(deliverBuffer));
676		-
677		- String jsonPath = System.getProperty("ROOTPATH") + File.separator + "kakaofile";
678		- jsonPath = jsonPath + File.separator + MessageUtil.getDate() + File.separator + SerialNoUtil.getSerialNo();
679		- FileUtil.mkdirs(jsonPath);
680		-
681		- ByteBuffer fileHeadBuffer = ByteBuffer.allocate(KakaoMessage.DELIVER_JSON_FILENAME_LENGTH + KakaoMessage.DELIVER_JSON_FILESIZE_LENGTH);
682		- channel.read(fileHeadBuffer);
683		-
684		- String fileName = KakaoMessage.getFileNameForDeliver(fileHeadBuffer);
685		- String fileSize = KakaoMessage.getFileSizeForDeliver(fileHeadBuffer);
686		-
687		- ByteBuffer fileBuffer = ByteBuffer.allocate(Integer.parseInt(fileSize));
688		- channel.read(fileBuffer);
689		- fileBuffer.flip();
690		- JobFileFactory.saveFileForByteBuffer(jsonPath, fileName, fileBuffer);
691		-
692		- messageDto.setKakaoJsonFile(jsonPath + File.separator + fileName);
693		- saveLog(printTaskLog() + "[KFT JSON] [File : " + fileName + ", Size : " + fileSize + "]");
694		-
695		- /* 사용자 단가, 발송망 설정 */
696		- MemberDto savedMemberDto = null;
697		- if (this.connectUserDto != null) {
698		- savedMemberDto = this.connectUserDto.getMemberDto();
699		- }
700		- if (savedMemberDto != null) {
701		- messageDto.setRouterSeq(savedMemberDto.getKakaoFtAgentCode());
702		- messageDto.setUnitCost(String.valueOf(savedMemberDto.getKakaoFtPrice()));
703		- }
704		-
705		- saveLog(printTaskLog() + "[MESSAGE : " + messageDto.toString() + "]");
706		- QueueTypeWorker worker = QueueTypeWorker.find("KFT");
707		- if (worker != null) {
708		- worker.pushQueue(messageDto);
709		- channel.write(KakaoMessage.makeDeliverAckBuffer(messageDto.getUserMsgID(), messageDto.getSendStatus()));
710		- connectUserDto.updateLastTrafficTime();
711		- }
712		- } catch (Exception e) {
713		- e.printStackTrace();
714	685	}
715	686	}
716	687
...	...	@@ -756,10 +727,10 @@
756	727	saveLog(printTaskLog() + "[REMOTE IP : " + connectUserDto.getRemoteIP() + "]");
757	728	saveLog(printTaskLog() + "[ALLOW IP BASIC : " + memberDto.getAllowIpBasic() + "]");
758	729	saveLog(printTaskLog() + "[ALLOW IP EXTEND : " + memberDto.getAllowIpExtend() + "]");
759		- if (memberDto.getAllowIpBasic() != null && connectUserDto.getRemoteIP().indexOf(memberDto.getAllowIpBasic()) >= 0) {
	730	+ if (memberDto.getAllowIpBasic() != null && connectUserDto.getRemoteIP().trim().equals(memberDto.getAllowIpBasic())) {
760	731	isPermit = true;
761	732	}
762		- if (memberDto.getAllowIpExtend() != null && connectUserDto.getRemoteIP().indexOf(memberDto.getAllowIpExtend()) >= 0) {
	733	+ if (memberDto.getAllowIpExtend() != null && connectUserDto.getRemoteIP().trim().equals(memberDto.getAllowIpExtend())) {
763	734	isPermit = true;
764	735	}
765	736	} else {
...	...	@@ -777,6 +748,9 @@
777	748	connectUserDto.setMemberDto(memberDto);
778	749	/* 세션통신 시간 업데이트 */
779	750	connectUserDto.updateLastTrafficTime();
	751	+ /* MSGID 중복체크큐 초기화 */
	752	+ KeyWriteQueue keyWriteQueue = new KeyWriteQueue(connectUserDto.getMsgKeyPath(), connectUserDto.getUserId(), connectUserDto.getMsgIdQueueSize());
	753	+ connectUserDto.setKeyWriteQueue(keyWriteQueue);
780	754	}
781	755	} catch (Exception e) {
782	756	resultCode = "10";
...	...	@@ -864,6 +838,11 @@
864	838	if (connectUserDto != null) {
865	839	if (connectUserDto.getUserId() != null) {
866	840	collectUserQueue.removeUser(connectUserDto.getServiceType(), connectUserDto.getUserId());
	841	+ /* MSG ID 중복체크 큐 해제 */
	842	+ KeyWriteQueue keyWriteQueue = connectUserDto.getKeyWriteQueue();
	843	+ if (keyWriteQueue != null) {
	844	+ keyWriteQueue.close();
	845	+ }
867	846	/* 모니터링 로그 */
868	847	HealthCheckServer.saveMonitorLog("[COLLECT SERVER][SERVICE TYPE : " + this.serviceType + "][ID : " + connectUserDto.getUserId() + "][EXPIRE CONNECT USER]");
869	848	}

...	...	@@ -24,6 +24,7 @@
24	24	<foreach separator="," item="item" collection="list">
25	25	(#{item.id}, #{item.msgGroupID}, #{item.userId}, #{item.userMsgID}, #{item.routerSeq}, 0, NOW(), #{item.userReceiver}, #{item.userSender}, NULL, #{item.userMessage}, '4' )
26	26	</foreach>
	27	+ ON DUPLICATE KEY UPDATE REQ_DATE = VALUES(REQ_DATE)
27	28	</insert>
28	29
29	30	<insert id="insertGroupForList" parameterType="java.util.List">

Delete comment