diff --git a/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java b/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java index 9dcd90a77..a08dc1dbf 100644 --- a/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java +++ b/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java @@ -19,13 +19,13 @@ package com.navercorp.pinpoint.web.websocket; +import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; -import com.navercorp.pinpoint.common.util.PinpointThreadFactory; -import com.navercorp.pinpoint.rpc.util.TimerFactory; import com.navercorp.pinpoint.web.service.AgentService; import com.navercorp.pinpoint.web.vo.AgentActiveThreadCountList; import com.navercorp.pinpoint.web.vo.AgentInfo; import org.apache.http.NameValuePair; +import org.apache.thrift.TException; import org.jboss.netty.util.Timeout; import org.jboss.netty.util.Timer; import org.jboss.netty.util.TimerTask; @@ -36,15 +36,11 @@ import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.io.IOException; +import java.util.*; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReadWriteLock; -import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.atomic.AtomicBoolean; /** * @Author Taejin Koo @@ -57,20 +53,20 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi private final String requestMapping; private final AgentService agentSerivce; + private final Timer timer; + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + private final Object lock = new Object(); + // it will be changed. private final long time = 1000; - - private final PinpointThreadFactory threadFactory = new PinpointThreadFactory("ActiveThread Handler", true); - private final TimerFactory timerFactory = new TimerFactory(); + private final AtomicBoolean onTimerTask = new AtomicBoolean(false); private final List sessionRepository = new CopyOnWriteArrayList(); private final ObjectMapper jsonConverter = new ObjectMapper(); - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - - private Timer timer; - public ActiveThreadCountHandler(WebSocketHandlerRegister register, AgentService agentSerivce) { this(register, DEFAULT_REQUEST_MAPPING, agentSerivce); } @@ -81,7 +77,6 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi this.timer = register.getTimer(); register.register(this); - Timeout timeout = timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS); } @Override @@ -93,7 +88,15 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi public void afterConnectionEstablished(WebSocketSession newSession) throws Exception { logger.info("ConnectionEstablished : {}", newSession); - sessionRepository.add(newSession); + synchronized (lock) { + sessionRepository.add(newSession); + Timeout timeout = timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS); + + boolean turnOn = onTimerTask.compareAndSet(false, true); + if (turnOn) { + timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS); + } + } super.afterConnectionEstablished(newSession); } @@ -102,7 +105,12 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi public void afterConnectionClosed(WebSocketSession closeSession, CloseStatus status) throws Exception { logger.info("ConnectionClosed : {}, caused : {}", closeSession, status); - sessionRepository.remove(closeSession); + synchronized (lock) { + sessionRepository.remove(closeSession); + if (sessionRepository.size() == 0) { + boolean turnOff = onTimerTask.compareAndSet(true, false); + } + } super.afterConnectionClosed(closeSession, status); } @@ -121,58 +129,108 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi super.handleTextMessage(session, message); } - private String getValue(List params, String key) { - for (NameValuePair nv : params) { - if (key.equals(nv.getName())) { - return nv.getValue(); - } - } - - return null; - } - private class ActiveThreadTimerTask implements TimerTask { @Override public void run(Timeout timeout) throws Exception { - if (sessionRepository.size() != 0) { + try { logger.info("ActiveThreadTimerTask started."); - Map> applicationGroup = new HashMap>(); - for (WebSocketSession session : sessionRepository) { - String applicationName = (String) session.getAttributes().get(APPLICATION_NAME_KEY); - - if (applicationName == null || applicationName.length() == 0) { - continue; - } - - if (!applicationGroup.containsKey(applicationName)) { - applicationGroup.put(applicationName, new ArrayList()); - } - - applicationGroup.get(applicationName).add(session); - } + Map> applicationGroup = createApplicationGroup(sessionRepository); for (Map.Entry> applicationEntry : applicationGroup.entrySet()) { String applicationName = applicationEntry.getKey(); - List agentInfoList = agentSerivce.getAgentInfoList(applicationName); - AgentActiveThreadCountList agentActiveThreadCountList = agentSerivce.getActiveThreadCount(agentInfoList); - - Map response = new HashMap(); - response.put(applicationName, agentActiveThreadCountList); - - String textMessage = jsonConverter.writeValueAsString(response); - for (WebSocketSession session : applicationEntry.getValue()) { - session.sendMessage(new TextMessage(textMessage)); - } + List agentInfoList = getAgentInfoList(applicationName); + AgentActiveThreadCountList agentActiveThreadCountList = getAgentActiveThreadCount(agentInfoList); + doResponse(applicationEntry.getValue(), applicationName, agentActiveThreadCountList); + } + } finally { + if (timer != null && onTimerTask.get()) { + timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS); } - } - - if (timer != null) { - timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS); } } } + private Map> createApplicationGroup(List sessionRepository) { + Map> applicationGroup = new HashMap>(); + for (WebSocketSession session : sessionRepository) { + String applicationName = (String) session.getAttributes().get(APPLICATION_NAME_KEY); + + if (applicationName == null || applicationName.length() == 0) { + continue; + } + + if (!applicationGroup.containsKey(applicationName)) { + applicationGroup.put(applicationName, new ArrayList()); + } + + applicationGroup.get(applicationName).add(session); + } + + return applicationGroup; + } + + private List getAgentInfoList(String applicationName) { + try { + List agentInfoList = agentSerivce.getAgentInfoList(applicationName); + return agentInfoList; + } catch (Exception e) { + logger.warn(e.getMessage(), e); + } + return Collections.emptyList(); + } + + private AgentActiveThreadCountList getAgentActiveThreadCount(List agentInfoList) { + try { + AgentActiveThreadCountList agentActiveThreadCountList = agentSerivce.getActiveThreadCount(agentInfoList); + return agentActiveThreadCountList; + } catch (TException e) { + logger.warn(e.getMessage(), e); + } + + return new AgentActiveThreadCountList(0); + } + + private void doResponse(List webSocketSessions, String applicationName, AgentActiveThreadCountList activeThreadCount) { + if (webSocketSessions == null) { + return; + } + + String textMessage = makeResponseMessage(applicationName, activeThreadCount); + + for (WebSocketSession session : webSocketSessions) { + try { + session.sendMessage(new TextMessage(textMessage)); + } catch (IOException e) { + logger.warn(e.getMessage(), e); + } + } + } + + private String makeResponseMessage(String applicationName, AgentActiveThreadCountList activeThreadCount) { + Map response = new HashMap(); + response.put(applicationName, activeThreadCount); + + try { + return jsonConverter.writeValueAsString(response); + } catch (JsonProcessingException e) { + logger.warn(e.getMessage(), e); + } + + return createEmptyJsonMessage(applicationName); + } + + private String createEmptyJsonMessage(String applicationName) { + StringBuilder emptyJsonMessage = new StringBuilder(); + emptyJsonMessage.append("{"); + emptyJsonMessage.append("\"").append(applicationName).append("\""); + emptyJsonMessage.append(":"); + emptyJsonMessage.append("{}"); + emptyJsonMessage.append("}"); + + return emptyJsonMessage.toString(); + } + }