add websocket interface

src/main/java/com/its/api/config/JobConfig.java

@@ -0,0 +1,26 @@
+package com.its.api.config;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.PostConstruct;
+
+@Slf4j
+@Getter
+@Setter
+@ToString
+@Configuration
+@ConfigurationProperties(prefix = "application.job")
+public class JobConfig {
+
+    private boolean cctvPreset = false;
+
+    @PostConstruct
+    private void init() {
+        log.info("{}", this);
+    }
+}

src/main/java/com/its/api/config/ThreadPoolTaskExecutorConfig.java

@@ -57,10 +57,10 @@ public class ThreadPoolTaskExecutorConfig extends AsyncConfigurerSupport {
         return threadPoolTaskExecutor;
     }
 
-    @Bean(name="dbmsJobExecutor")
-    public Executor getDbmsJobExecutor() {
+    @Bean(name="schJobExecutor")
+    public Executor getSchJobExecutor() {
         ThreadPoolTaskExecutor threadPoolTaskExecutor = getDefaultExecutor(this.poolCore);
-        threadPoolTaskExecutor.setThreadNamePrefix("dbms-pool-");
+        threadPoolTaskExecutor.setThreadNamePrefix("sch-job-pool-");
         threadPoolTaskExecutor.initialize();
         return threadPoolTaskExecutor;
     }

src/main/java/com/its/api/its/model/dto/incident/TbIncdOcrrDto.java

@@ -7,7 +7,7 @@ import io.swagger.annotations.ApiModelProperty;
 import lombok.*;
 
 import javax.validation.constraints.Pattern;
-import javax.validation.constraints.Positive;
+import javax.validation.constraints.PositiveOrZero;
 import javax.validation.constraints.Size;
 import java.io.Serializable;
 
@@ -341,12 +341,12 @@ public class TbIncdOcrrDto implements Serializable {
 
         @ApiModelProperty("지체 길이, Nullable = Y, NUMBER(5)")  // Y NUMBER(5)
         @JsonProperty("dely_lngt")
-        @Positive
+        @PositiveOrZero
         private Integer delyLngt;
 
         @ApiModelProperty("돌발상황 길이, Nullable = Y, NUMBER(5)")  // Y NUMBER(5)
         @JsonProperty("incd_lngt")
-        @Positive
+        @PositiveOrZero
         private Integer incdLngt;
 
         @ApiModelProperty("돌발상황 발생 차로, Nullable = Y, VARCHAR(6)")  // Y VARCHAR(6)
@@ -390,22 +390,22 @@ public class TbIncdOcrrDto implements Serializable {
 
         @ApiModelProperty("소형 차량 대수, Nullable = Y, NUMBER(5)")  // Y NUMBER(5)
         @JsonProperty("smlt_vhcl_unum")
-        @Positive
+        @PositiveOrZero
         private Integer smltVhclUnum;
 
         @ApiModelProperty("중대형 차량 대수, Nullable = Y, NUMBER(5)")  // Y NUMBER(5)
         @JsonProperty("mdlt_vhcl_unum")
-        @Positive
+        @PositiveOrZero
         private Integer mdltVhclUnum;
 
         @ApiModelProperty("부상자 인원수, Nullable = Y, NUMBER(4)")  // Y NUMBER(4)
         @JsonProperty("injr_pnum")
-        @Positive
+        @PositiveOrZero
         private Integer injrPnum;
 
         @ApiModelProperty("사망자 인원수, Nullable = Y, NUMBER(4)")  // Y NUMBER(4)
         @JsonProperty("ftlt_pnum")
-        @Positive
+        @PositiveOrZero
         private Integer ftltPnum;
 
         @ApiModelProperty("기상 코드, Nullable = Y, VARCHAR(7)")  // Y VARCHAR(7)
@@ -446,7 +446,7 @@ public class TbIncdOcrrDto implements Serializable {
 
         @ApiModelProperty("출동 인원수, Nullable = Y, NUMBER(5)")  // Y NUMBER(5)
         @JsonProperty("mblz_pnum")
-        @Positive
+        @PositiveOrZero
         private Integer mblzPnum;
 
         @ApiModelProperty("복구 내용, Nullable = Y, VARCHAR(400)")  // Y VARCHAR(400)
@@ -456,12 +456,12 @@ public class TbIncdOcrrDto implements Serializable {
 
         @ApiModelProperty("복구 기간, Nullable = Y, NUMBER(5)")  // Y NUMBER(5)
         @JsonProperty("recr_perd")
-        @Positive
+        @PositiveOrZero
         private Integer recrPerd;
 
         @ApiModelProperty("링크ID, Nullable = Y, NUMBER(10)")  // Y NUMBER(10)
         @JsonProperty("link_id")
-        @Positive
+        @PositiveOrZero
         private Long linkId;
 
         @Builder

src/main/java/com/its/api/scheduler/ItsApiScheduler.java

@@ -1,6 +1,8 @@
 package com.its.api.scheduler;
 
-import com.its.api.scheduler.job.CctvPsetScnrThread;
+import com.its.api.config.JobConfig;
+import com.its.api.scheduler.job.CctvPsetScnrJobThread;
+import com.its.api.scheduler.job.FcltSttsJobThread;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Async;
@@ -16,7 +18,9 @@ import javax.annotation.PreDestroy;
 @Component
 public class ItsApiScheduler {
 
-    private final CctvPsetScnrThread cctvPsetScnrThread;
+    private final JobConfig jobConfig;
+    private final CctvPsetScnrJobThread cctvPsetScnrThread;
+    private final FcltSttsJobThread fcltSttsJobThread;
 
     @PreDestroy
     public void onShutDown() {
@@ -28,8 +32,17 @@ public class ItsApiScheduler {
     @Async
     @Scheduled(cron = "0 * * * * *")  // 1분 주기 작업 실행
     public void jobCctvPsetScnr() {
-        log.info("START jobCctvPsetScnr");
-        this.cctvPsetScnrThread.run();
+        if (this.jobConfig.isCctvPreset()) {
+            log.info("START CctvPsetScnrJob");
+            this.cctvPsetScnrThread.run();
+        }
+    }
+
+    @Async
+    @Scheduled(cron = "0/30 * * * * *")  // 30초 주기 작업 실행
+    public void pollingFcltStts() {
+        log.info("START FcltSttsJob");
+        fcltSttsJobThread.run();
     }
 
     @Async

src/main/java/com/its/api/scheduler/job/CctvPsetScnrThread.java → src/main/java/com/its/api/scheduler/job/CctvPsetScnrJobThread.java

@@ -16,7 +16,7 @@ import java.util.List;
 @Slf4j
 @AllArgsConstructor
 @Service
-public class CctvPsetScnrThread {
+public class CctvPsetScnrJobThread {
 
     private final TbCctvScnrRepository cctvScnrRepo;
 

src/main/java/com/its/api/scheduler/job/FcltSttsJobThread.java

@@ -0,0 +1,67 @@
+package com.its.api.scheduler.job;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.its.api.bis.service.BitService;
+import com.its.api.its.model.dto.common.FcltSttsListDto;
+import com.its.api.its.service.common.CommonSttsService;
+import com.its.api.websocket.ItsWebSocketMessage;
+import com.its.api.websocket.ItsWebSocketSessionManager;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StopWatch;
+import org.springframework.web.socket.TextMessage;
+
+import javax.annotation.PostConstruct;
+
+@Slf4j
+@AllArgsConstructor
+@Service
+public class FcltSttsJobThread {
+
+    private final CommonSttsService itsService;
+    private final BitService bitService;
+    private ObjectMapper mapper;
+
+    @PostConstruct
+    private void init() {
+        this.mapper = new ObjectMapper();
+        log.info("{}", this);
+    }
+
+    @Async("schJobExecutor")
+    public void run() {
+
+        log.info("START: FcltSttsJobThread.run: {}", Thread.currentThread().getName());
+
+        StopWatch stopWatch = new StopWatch();
+        stopWatch.start();
+
+        // ITS Fclt Stts
+        FcltSttsListDto itsSttsDto = this.itsService.findAllFcltSttsTotal();
+        ItsWebSocketMessage itsMessage = new ItsWebSocketMessage("itsFcltStts", itsSttsDto);
+        try {
+            String itsSttsJsonData = this.mapper.writeValueAsString(itsMessage);
+            ItsWebSocketSessionManager.getInstance().sendBroadcastMessage(itsMessage.getCommand(), new TextMessage(itsSttsJsonData));
+        }
+        catch(JsonProcessingException e){
+            log.error("FcltSttsJobThread ItsFcltSttsDto Json parsing Exception: {}, {}", itsSttsDto, e.getMessage());
+        }
+
+        // BIS BIT Stts
+        FcltSttsListDto bitSttsDto = this.bitService.findAllListSttsTotal();
+        ItsWebSocketMessage bisMessage = new ItsWebSocketMessage("bisFcltStts", bitSttsDto);
+        try {
+            String bitSttsJsonData = this.mapper.writeValueAsString(bisMessage);
+            ItsWebSocketSessionManager.getInstance().sendBroadcastMessage(bisMessage.getCommand(), new TextMessage(bitSttsJsonData));
+        }
+        catch(JsonProcessingException e){
+            log.error("FcltSttsJobThread BisFcltSttsDto Json parsing Exception: {}, {}", itsSttsDto, e.getMessage());
+        }
+
+        stopWatch.stop();
+        log.info("--END: FcltSttsJobThread.run: {}, {} ms.", Thread.currentThread().getName(), stopWatch.getTotalTimeMillis());
+    }
+}

src/main/java/com/its/api/websocket/ItsWebSocketConfig.java

@@ -0,0 +1,29 @@
+package com.its.api.websocket;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.config.annotation.EnableWebSocket;
+import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
+
+@Slf4j
+@RequiredArgsConstructor
+@Configuration
+@EnableWebSocket
+public class ItsWebSocketConfig implements WebSocketConfigurer {
+
+    private final ItsWebSocketHandler itsWebsocketHandler;
+
+    @Override
+    public void registerWebSocketHandlers(WebSocketHandlerRegistry webSocketHandlerRegistry) {
+
+        webSocketHandlerRegistry.addHandler(this.itsWebsocketHandler, "/ws/messages.do")
+                .setAllowedOriginPatterns("*")
+                .withSockJS();  // sockjs
+
+       webSocketHandlerRegistry.addHandler(this.itsWebsocketHandler, "/ws/messages.do")
+                .setAllowedOriginPatterns("*"); // 그냥 websocket 지원
+    }
+
+}

src/main/java/com/its/api/websocket/ItsWebSocketHandler.java

@@ -0,0 +1,69 @@
+package com.its.api.websocket;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+
+import java.util.Map;
+
+@Slf4j
+@Controller
+@RequestMapping("/ws/messages.do")
+public class ItsWebSocketHandler extends TextWebSocketHandler {
+
+    public ItsWebSocketHandler() {
+
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            for (Map.Entry<org.springframework.web.socket.WebSocketSession, ItsWebSocketSession> obj : ItsWebSocketSessionManager.getInstance().getMap().entrySet()) {
+                if (obj.getValue() != null) {
+                }
+            }
+        }));
+
+        log.info("WebsocketHandler() START");
+    }
+
+    @Override
+    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
+        // 클라이언트가 연결되었을때 실행
+        log.info("afterConnectionEstablished: " + session.getRemoteAddress() + ",  URI: " + session.getUri() + ", UUID: " + session.getId());
+        super.afterConnectionEstablished(session);
+
+        ItsWebSocketSession vo = new ItsWebSocketSession(this, session);
+        ItsWebSocketSessionManager.getInstance().addSession(session, vo);
+    }
+
+    //클라이언트가 웹소켓 서버로 메시지를 전송했을 때 실행
+    @Override
+    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
+        log.info("handleTextMessage: " + session.getRemoteAddress() + ",  Uri[" + session.getUri() + "], UUID[" + session.getId() + "], message[" + message.getPayload() + "]");
+        /*List<String> requests = ItsUtils.split(message.getPayload(), ":");
+        if (requests.size() != 2) {
+            log.error("Request data parsing error: {}", requests);
+            session.close();
+            return;
+        }
+*/
+        ItsWebSocketSession sessionClient = ItsWebSocketSessionManager.getInstance().getSession(session);
+        if (sessionClient == null) {
+            log.error("Request session not found: {}", session);
+            session.close();
+            return;
+        }
+        log.info("Payload: {}", message.getPayload());
+    }
+
+    //클라이언트 연결을 끊었을 때 실행
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
+        log.info("afterConnectionClosed: " + session.getRemoteAddress() + ",  URI: " + session.getUri() + ", UUID: " + session.getId());
+        super.afterConnectionClosed(session, status);
+
+        ItsWebSocketSessionManager.getInstance().removeSession(session);
+    }
+
+}

src/main/java/com/its/api/websocket/ItsWebSocketMessage.java

@@ -0,0 +1,18 @@
+package com.its.api.websocket;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Data
+public class ItsWebSocketMessage {
+
+    private String command;
+    private Object data;
+
+    public ItsWebSocketMessage(String command, Object data) {
+        this.command = command;
+        this.data = data;
+    }
+
+}

src/main/java/com/its/api/websocket/ItsWebSocketSession.java

@@ -0,0 +1,33 @@
+package com.its.api.websocket;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+
+@Slf4j
+@Data
+public class ItsWebSocketSession {
+
+    private ItsWebSocketHandler itsWebsocketHandler;
+    private WebSocketSession session;
+    private String groupId;
+
+    public ItsWebSocketSession(ItsWebSocketHandler itsWebsocketHandler, org.springframework.web.socket.WebSocketSession session) {
+        this.itsWebsocketHandler = itsWebsocketHandler;
+        this.session = session;
+    }
+
+    public void sendMessage(String command, TextMessage message) {
+
+        if (this.session != null && this.session.isOpen()) {
+            synchronized (this.session) {
+                try {
+                    this.session.sendMessage(message);
+                } catch (Exception e) {
+                    log.error("sendMessage: nodeId: {}, session: {}, {}", command, session, e.getMessage());
+                }
+            }
+        }
+    }
+}

src/main/java/com/its/api/websocket/WebsocketSessionManager.java → src/main/java/com/its/api/websocket/ItsWebSocketSessionManager.java

@@ -2,35 +2,36 @@ package com.its.api.websocket;
 
 import com.its.api.utils.Counter;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
 
 import java.util.concurrent.ConcurrentHashMap;
 
 @Slf4j
-public class WebsocketSessionManager {
+public class ItsWebSocketSessionManager {
 
-    private static WebsocketSessionManager _instance = null;
+    private static ItsWebSocketSessionManager _instance = null;
 
-    public static WebsocketSessionManager getInstance() {
+    private volatile boolean serverRun;
+    private Counter sessions;
+    private ConcurrentHashMap<WebSocketSession, ItsWebSocketSession> sessionMap;
+
+    public static ItsWebSocketSessionManager getInstance() {
         if (_instance == null) {
-            synchronized (WebsocketSessionManager.class) {
+            synchronized (ItsWebSocketSessionManager.class) {
                 if (_instance == null)
-                    _instance = new WebsocketSessionManager();
+                    _instance = new ItsWebSocketSessionManager();
             }
         }
         return _instance;
     }
 
-    private WebsocketSessionManager() {
+    private ItsWebSocketSessionManager() {
         this.sessions = new Counter();
         this.sessionMap = new ConcurrentHashMap<>();
     }
 
-    private volatile boolean serverRun;
-    private Counter sessions;
-    private ConcurrentHashMap<WebSocketSession, WebSocketSessionClient> sessionMap;
-
-    public ConcurrentHashMap<WebSocketSession, WebSocketSessionClient> getMap() {
+    public ConcurrentHashMap<WebSocketSession, ItsWebSocketSession> getMap() {
         return this.sessionMap;
     }
 
@@ -44,28 +45,24 @@ public class WebsocketSessionManager {
         return (int) this.sessions.get();
     }
 
-    public void addSession(WebSocketSession session, WebSocketSessionClient vo) {
+    public void addSession(WebSocketSession session, ItsWebSocketSession vo) {
         this.sessionMap.put(session, vo);
     }
     public void removeSession(WebSocketSession session) {
         this.sessionMap.remove(session);
     }
-
-    public WebSocketSessionClient getSessionVo(WebSocketSession session) {
+    public ItsWebSocketSession getSession(WebSocketSession session) {
         return this.sessionMap.get(session);
     }
 
-    public void start() {
-        this.serverRun = true;
-    }
-    public void stop() {
-        this.serverRun = false;
-    }
-    public boolean isServerRun() {
-        return this.serverRun;
-    }
-
     public void reportSession() {
         log.info("Sessions: {}", this.sessionMap.toString());
     }
+
+    public void sendBroadcastMessage(String command, TextMessage message) {
+        this.sessionMap.forEach((key, value) -> {
+            log.info("{}, {}", command, message);
+            value.sendMessage(command, message);
+        });
+    }
 }

src/main/java/com/its/api/websocket/SessionConsumerThread.java

@@ -1,145 +0,0 @@
-package com.its.api.websocket;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.web.socket.WebSocketSession;
-
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-@Slf4j
-@Getter
-@Setter
-public class SessionConsumerThread implements Runnable {
-
-    private AtomicBoolean stopFlag = new AtomicBoolean(false);
-    private AtomicBoolean updateTopics = new AtomicBoolean(false);
-
-    private final WebsocketHandler websocketHandler;
-    private final WebSocketSession session;
-    private String groupId;
-    private List<String> topics;
-    //private KafkaConsumer<String, byte[]> consumer;
-    private ObjectMapper mapper;
-
-    public SessionConsumerThread(WebsocketHandler websocketHandler, WebSocketSession session) {
-        this.websocketHandler = websocketHandler;
-        this.session = session;
-
-        this.topics = new CopyOnWriteArrayList();
-        this.mapper = new ObjectMapper();
-
-        this.groupId = this.session.getId();
-    }
-
-    /*public List<String> formatPartitions(Collection<TopicPartition> partitions) {
-        return partitions.stream().map(topicPartition ->
-                        String.format("\ntopic: %s, partition: %s", topicPartition.topic(), topicPartition.partition()))
-                .collect(Collectors.toList());
-    }*/
-
-    @Override
-    public void run() {
-
-        log.info("SessionConsumerThread start: {}", this.groupId);
-
-        try {
-            boolean unsubscribe = false;
-            while (!this.stopFlag.get() && (!Thread.currentThread().isInterrupted())) {
-/*
-                if (this.topics.size() == 0) {
-                    try {
-                        if (unsubscribe) {
-                            this.consumer.unsubscribe();
-                            unsubscribe = false;
-                        }
-                        Thread.sleep(100);
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    }
-                    continue;
-                }
-
-                if (this.consumer == null) {
-                    TsiKafkaConsumerConfig config = (TsiKafkaConsumerConfig) AppUtils.getBean(TsiKafkaConsumerConfig.class);
-                    this.consumer = new KafkaConsumer<>(config.getConsumerProperties(this.groupId));
-                    log.info("Kafka Consumer Create: {}", this.consumer);
-                }
-
-                if (this.updateTopics.get()) {
-                    this.updateTopics.set(false);
-
-                    log.info("SessionConsumerThread Subscribe Start: {}, {}", this.groupId, this.topics);
-                    *//*this.consumer.unsubscribe();
-
-                    this.consumer.subscribe(this.topics);
-                    Set<TopicPartition> assignment = new HashSet<>();
-                    while (assignment.size() == 0) {
-                        consumer.poll(Duration.ofMillis(100));
-                        assignment = consumer.assignment();
-                    }
-                    log.info("SessionConsumerThread Subscribe Partitions: {} EA", assignment.size());
-                    consumer.seekToEnd(assignment);
-*//*
-                    this.consumer.subscribe(this.topics,
-                            new ConsumerRebalanceListener() {
-                                @Override
-                                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
-                                    log.info("onPartitionsRevoked - consumerName: {}, partitions: {}", topics.toString(), formatPartitions(partitions));
-                                }
-
-                                @Override
-                                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
-                                    log.info("onPartitionsAssigned - consumerName: {}, partitions: {}", topics.toString(), formatPartitions(partitions));
-                                    consumer.seekToEnd(partitions);
-                                }
-                            });
-                    log.info("SessionConsumerThread Subscribe ..End: {}, {}", this.groupId, this.topics);
-                    unsubscribe = true;
-                }
-
-                ConsumerRecords<String, byte[]> records = this.consumer.poll(Duration.ofMillis(100));
-                for (ConsumerRecord<String, byte[]> record : records) {
-                    try {
-                        CpuNodeStatusDTO status = TsiCpuNodePacket.getNodeStatusDTO(Long.parseLong(record.key()), record.value());
-                        if (status != null) {
-                            try {
-                                String jsonInString = this.mapper.writeValueAsString(status);
-                                 try {
-                                    this.websocketHandler.sendMessage(this.session, record.key(), new TextMessage(jsonInString));
-                                    log.info("Send to: {}, {}, {} bytes.", this.session.getRemoteAddress().getAddress(), record.key(), jsonInString.length());
-                                }
-                                catch(Exception e) {
-                                    log.error("Send Failed: {}, {}, {} bytes.", this.session, record.key(), jsonInString.length());
-                                }
-                            }
-                            catch(JsonProcessingException e) {
-                                log.error("ConsumerThread Json parsing Exception: {}, {}", this.topics.toString(), e.getMessage());
-                            }
-                        }
-                    } catch (Exception e) {
-                        log.error("SessionConsumerThread Exception: {}, {}, {}", this.topics.toString(), this.session.toString(), e.getMessage());
-                    }
-                }*/
-            }
-
-            log.info("SessionConsumerThread: {}, {}, {}, stopped.", this.groupId, this.topics.toString(), this.session.toString());
-        }
-        catch(Exception e) {
-            log.error("SessionConsumerThread Wakeup Exception: {}, {}", this.groupId, e.getMessage());
-        }
-        finally {
-        }
-    }
-
-    public void stop() {
-        this.stopFlag.set(true);
-    }
-
-    public void shutdown() {
-        //this.consumer.wakeup();
-    }
-}

src/main/java/com/its/api/websocket/WebSocketConfig.java

@@ -1,30 +0,0 @@
-package com.its.api.websocket;
-
-import org.springframework.context.annotation.Configuration;
-import org.springframework.web.socket.config.annotation.EnableWebSocket;
-import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
-import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
-
-@Configuration
-@EnableWebSocket
-public class WebSocketConfig implements WebSocketConfigurer {
-
-    private final WebsocketHandler websocketHandler;
-
-    public WebSocketConfig(WebsocketHandler websocketHandler) {
-        this.websocketHandler = websocketHandler;
-    }
-
-    @Override
-    public void registerWebSocketHandlers(WebSocketHandlerRegistry webSocketHandlerRegistry) {
-        webSocketHandlerRegistry.addHandler(this.websocketHandler, "/op/socket")
-                .setAllowedOrigins("*")
-                .withSockJS();  // sockjs
-        webSocketHandlerRegistry.addHandler(this.websocketHandler, "/op/socket")
-                .setAllowedOrigins("*"); // 그냥 websocket 지원
-
-        /*webSocketHandlerRegistry.addHandler(echoHandler, "/echo").setAllowedOrigins("*").withSockJS()
-                .setInterceptors(new HttpSessionHandshakeInterceptor())
-                .setClientLibraryUrl("http://localhost:8080/resources/sockjs.min.js");*/
-    }
-}

src/main/java/com/its/api/websocket/WebSocketSessionClient.java

@@ -1,154 +0,0 @@
-package com.its.api.websocket;
-
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.web.socket.WebSocketSession;
-
-import java.util.ArrayList;
-import java.util.List;
-
-@Slf4j
-@Data
-public class WebSocketSessionClient {
-
-    private WebsocketHandler websocketHandler;
-    private WebSocketSession session;
-    private String groupId;
-    private List<String> topics = new ArrayList<>();
-    private List<String> threadTopics = new ArrayList<>();
-
-    private SessionConsumerThread consumerThread = null;
-    private Thread thread;
-
-    public WebSocketSessionClient(WebsocketHandler websocketHandler, WebSocketSession session) {
-        this.websocketHandler = websocketHandler;
-        this.session = session;
-    }
-
-    public void delSubscription(List<String> topics, boolean isExit) {
-        if (topics == null) {
-            return;
-        }
-/*
-        this.threadTopics = new ArrayList<>();
-        for (String topic: topics) {
-            boolean remain = true;
-            TsiKafkaVo kafkaVo = TsiKafkaConsumerManager.getInstance().get(Long.parseLong(topic));
-            if (kafkaVo != null) {
-                if (kafkaVo.getNodeConsumerThread() != null) {
-                    kafkaVo.getNodeConsumerThread().removeSession(this.session);
-                    remain = false;
-                    log.info("delSubscription: {}, consumer: {}, session: {}", topic, kafkaVo.getNodeConsumerThread(), session.toString());
-                }
-            }
-            if (remain) {
-                this.threadTopics.add(topic);
-            }
-        }*/
-
-        if (this.consumerThread != null && this.threadTopics.size() > 0) {
-            if (isExit) {
-                this.threadTopics = new ArrayList<>();
-            }
-            this.consumerThread.setTopics(this.threadTopics);
-            this.consumerThread.getUpdateTopics().set(true);
-        }
-    }
-
-    public void addSubscription(List<String> topics) {
-        if (topics == null) {
-            return;
-        }
-
-        long currMilliseconds = System.currentTimeMillis();
-        this.threadTopics = new ArrayList<>();
-        /*for (String topic: topics) {
-            boolean remain = true;
-            TsiKafkaVo kafkaVo = TsiKafkaConsumerManager.getInstance().get(Long.parseLong(topic));
-            if (kafkaVo != null) {
-                // 등록된 교차로인 경우 일단 요청해 놓는다.
-                if (kafkaVo.getNodeConsumerThread() != null) {
-                    kafkaVo.getNodeConsumerThread().addSession(this.session);
-
-                    log.info("addSubscription: {}, recvTm: {}, consumer: {}, session: {}", topic, currMilliseconds - kafkaVo.getNodeConsumerThread().getRecvTime(), kafkaVo.getNodeConsumerThread(), session.toString());
-
-                    // 통신이 OffLine 인 경우 offline 메시지 전송
-                    if (currMilliseconds - kafkaVo.getNodeConsumerThread().getRecvTime() > 5000) {
-                        try {
-                            this.websocketHandler.sendMessage(this.session, topic, new TextMessage(topic + ":offline"));
-                        } catch (Exception e) {
-                            //
-                        }
-                    }
-                }
-            }
-            if (remain) {
-                this.threadTopics.add(topic);
-            }
-        }*/
-
-        if (this.threadTopics.size() > 0) {
-            if (this.consumerThread == null) {
-                this.consumerThread = new SessionConsumerThread(this.websocketHandler, this.session);
-                this.thread = new Thread(this.consumerThread);
-                this.thread.setUncaughtExceptionHandler(
-                        new Thread.UncaughtExceptionHandler() {
-                            @Override
-                            public void uncaughtException(Thread th, Throwable ex) {
-                                log.error("Uncaught exception: {}, {}", groupId, ex.getMessage());
-                            }
-                        }
-                );
-                this.thread.start();
-            }
-            this.consumerThread.setTopics(this.threadTopics);
-            this.consumerThread.getUpdateTopics().set(true);
-        }
-    }
-
-    public void subscribe(List<String> topics) {
-        if (topics.size() == 1 && topics.get(0).equals("x")) {
-            // 세션 종료 메시지 안 경우임. 모든 요청 토픽 삭제
-            delSubscription(this.topics, true);
-            this.topics = new ArrayList<>();
-        }
-        else {
-            List<String> delTopics = new ArrayList<>();
-            List<String> newTopics = new ArrayList<>();
-
-            // 지금 요청중인 토픽 중에서 이번 요청에 포함되지 않은 토픽은 삭제한다.
-            for (String topic: this.topics) {
-                if (!topics.contains(topic)) {
-                    delTopics.add(topic);
-                }
-            }
-            if (delTopics.size() > 0) {
-                delSubscription(delTopics, false);
-            }
-
-            // 이번 요청 토픽중에 지금 요청중이 아닌 경우에만 새롭게 요청한다.
-            for (String topic: topics) {
-                if (!this.topics.contains(topic)) {
-                    newTopics.add(topic);
-                }
-            }
-            if (newTopics.size() > 0) {
-                addSubscription(newTopics);
-            }
-
-            this.topics = topics;
-        }
-    }
-
-    public void start(String groupId) {
-        this.groupId = groupId;
-    }
-
-    public void stop() {
-        delSubscription(this.topics, true);
-
-        if (this.consumerThread != null) {
-            this.consumerThread.stop();
-        }
-    }
-}

src/main/java/com/its/api/websocket/WebsocketHandler.java

@@ -1,135 +0,0 @@
-package com.its.api.websocket;
-
-import com.its.api.utils.ItsUtils;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Controller;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.socket.CloseStatus;
-import org.springframework.web.socket.TextMessage;
-import org.springframework.web.socket.WebSocketSession;
-import org.springframework.web.socket.handler.TextWebSocketHandler;
-
-import java.util.List;
-import java.util.Map;
-
-@Slf4j
-@Controller
-@RequestMapping("/op/socket")
-public class WebsocketHandler extends TextWebSocketHandler {
-
-    public WebsocketHandler() {
-
-        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-            for (Map.Entry<WebSocketSession, WebSocketSessionClient> obj : WebsocketSessionManager.getInstance().getMap().entrySet()) {
-                if (obj.getValue() != null) {
-                    if (obj.getValue().getConsumerThread() != null) {
-                        obj.getValue().getConsumerThread().shutdown();
-                    }
-                }
-            }
-        }));
-
-        log.info("WebsocketHandler() START");
-    }
-
-    @Override
-    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
-        // 클라이언트가 연결되었을때 실행
-        log.info("afterConnectionEstablished: " + session.getRemoteAddress() + ",  Uri: " + session.getUri() + ", UUID: " + session.getId());
-        super.afterConnectionEstablished(session);
-
-        WebSocketSessionClient vo = new WebSocketSessionClient(this, session);
-        WebsocketSessionManager.getInstance().addSession(session, vo);
-    }
-
-    //클라이언트가 웹소켓 서버로 메시지를 전송했을 때 실행
-    @Override
-    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
-        log.info("handleTextMessage: " + session.getRemoteAddress() + ",  Uri[" + session.getUri() + "], UUID[" + session.getId() + "], message[" + message.getPayload() + "]");
-        List<String> requests = ItsUtils.split(message.getPayload(), ":");
-        if (requests.size() != 2) {
-            log.error("Request data parsing error: {}", requests.toString());
-            session.close();
-            return;
-        }
-
-        WebSocketSessionClient sessionClient = WebsocketSessionManager.getInstance().getSessionVo(session);
-        if (sessionClient == null) {
-            log.error("Request session not found: {}", session.toString());
-            session.close();
-            return;
-        }
-        log.info("Payload: {}", message.getPayload());
-
-        String command = requests.get(0).trim();
-        String data = requests.get(1).trim();
-        if (command.equals("group")) {
-            String groupId = data;
-            sessionClient.start(groupId);
-            log.info("Request group: {}", groupId);
-        }
-        else if (command.equals("topics")) {
-
-            String topics = data.replaceAll("[ ]", "");
-            topics = topics.replaceAll(" ", "");
-            List<String> oldTopics = sessionClient.getTopics();
-            List<String> newTopics = ItsUtils.split(topics, ",");
-
-            log.info("Request topic: old: {}, new: {}", oldTopics, newTopics);
-
-            // 현재 요청 토픽목록을 새롭게 할당한다.
-            sessionClient.subscribe(newTopics);
-            log.info("Subscribe topic ok: {}", sessionClient.toString());
-        }
-    }
-
-    //클라이언트 연결을 끊었을 때 실행
-    @Override
-    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
-        log.info("afterConnectionClosed: " + session.getRemoteAddress() + ",  Uri: " + session.getUri() + ", UUID: " + session.getId());
-        super.afterConnectionClosed(session, status);
-
-        removeSession(session, status);
-    }
-
-    public void removeSession(WebSocketSession session, CloseStatus status) {
-        WebSocketSessionClient sessionClient = WebsocketSessionManager.getInstance().getSessionVo(session);
-        if (sessionClient != null) {
-            sessionClient.stop();
-        }
-        WebsocketSessionManager.getInstance().removeSession(session);
-
-    }
-
-    // 세션에 해당 하는 토픽스레드 삭제
-    /*public void removeConsumerThread(WebSocketSession session, ConsumerThread consumerThread) {
-        try {
-            if (consumerThread == null) {
-                return;
-            }
-            int remainSessions = consumerThread.removeSession(session);
-            if (remainSessions == 0) {
-                String topicName = consumerThread.getTopicName();
-                consumerThread.stop();
-                this.consumerMap.remove(topicName);
-                log.info("topic: {}, consumer terminated.", topicName);
-            }
-        }
-        catch(Exception e) {
-            log.error("removeConsumerThread: Exception: {}", e.getMessage());
-        }
-    }*/
-
-    public void sendMessage(WebSocketSession session, String nodeId, TextMessage message) {
-
-        if (session != null && session.isOpen()) {
-            synchronized (session) {
-                try {
-                    session.sendMessage(message);
-                } catch (Exception e) {
-                    log.error("sendMessage: nodeId: {}, session: {}, {}", nodeId, session, e.getMessage());
-                }
-            }
-        }
-    }
-}

src/main/resources/application.yml

@@ -7,6 +7,8 @@ application:
     region-name: 용인시
   center-comm:
     binding-port: 4602
+  job:
+    cctv-preset: false
 
 server:
   port: 8999