@OnMessage
    public void onMessage(final String message, final Session session) {
        if (!Objects.equals(message, DataEventTypeEnum.MYSELF.name())
                && !Objects.equals(message, DataEventTypeEnum.RUNNING_MODE.name())) {
            return;
        }
        
        if (Objects.equals(message, DataEventTypeEnum.RUNNING_MODE.name())) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("websocket fetching running mode info...");
            }
            // check if this node is master
            boolean isMaster = true;
            String runningMode = RunningModeEnum.STANDALONE.name();
            String masterUrl = StringUtils.EMPTY;
            ClusterProperties clusterProperties = SpringBeanUtils.getInstance().getBean(ClusterProperties.class);
            if (clusterProperties.isEnabled()) {
                ClusterSelectMasterService clusterSelectMasterService = SpringBeanUtils.getInstance().getBean(ClusterSelectMasterService.class);
                runningMode = RunningModeEnum.CLUSTER.name();
                isMaster = clusterSelectMasterService.isMaster();
                masterUrl = clusterSelectMasterService.getMasterUrl();
            }
            Map<String, Object> map = Maps.newHashMap();
            map.put(RunningModeConstants.EVENT_TYPE, DataEventTypeEnum.RUNNING_MODE.name());
            map.put(RunningModeConstants.IS_MASTER, isMaster);
            map.put(RunningModeConstants.RUNNING_MODE, runningMode);
            map.put(RunningModeConstants.MASTER_URL, masterUrl
                    .replace("http", "ws")
                    .replace("https", "ws")
                    .concat("/websocket"));
            if (isMaster) {
                ThreadLocalUtils.put(SESSION_KEY, session);
            }

            sendMessageBySession(session, JsonUtils.toJson(map));
            return;
        }
        
        if (Objects.equals(message, DataEventTypeEnum.MYSELF.name())) {
            try {
                ThreadLocalUtils.put(SESSION_KEY, session);
                SpringBeanUtils.getInstance().getBean(SyncDataService.class).syncAll(DataEventTypeEnum.MYSELF);
            } finally {
                ThreadLocalUtils.clear();
            }
        }
        
    }
@Test
    public void testOnMessage() {
        ConfigurableApplicationContext context = mock(ConfigurableApplicationContext.class);
        SpringBeanUtils.getInstance().setApplicationContext(context);
        when(SpringBeanUtils.getInstance().getBean(SyncDataService.class)).thenReturn(syncDataService);
        when(syncDataService.syncAll(DataEventTypeEnum.MYSELF)).thenReturn(true);
        websocketCollector.onOpen(session);
        websocketCollector.onMessage(DataEventTypeEnum.MYSELF.name(), session);
        assertEquals(1L, getSessionSetSize());
        verify(syncDataService, times(1)).syncAll(DataEventTypeEnum.MYSELF);
        doNothing().when(loggerSpy).warn(anyString(), anyString());
        websocketCollector.onClose(session);
    }