@Test public void scanNotActiveChannel() throws Exception { producerManager.registerProducer(group, clientInfo); assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNotNull(); assertThat(producerManager.findChannel("clientId")).isNotNull(); Field field = ProducerManager.class.getDeclaredField("CHANNEL_EXPIRED_TIMEOUT"); field.setAccessible(true); long CHANNEL_EXPIRED_TIMEOUT = field.getLong(producerManager); clientInfo.setLastUpdateTimestamp(System.currentTimeMillis() - CHANNEL_EXPIRED_TIMEOUT - 10); when(channel.close()).thenReturn(mock(ChannelFuture.class)); producerManager.scanNotActiveChannel(); assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNull(); assertThat(producerManager.findChannel("clientId")).isNull(); }
public void scanNotActiveChannel() { for (final Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>> entry : this.groupChannelTable .entrySet()) { final String group = entry.getKey(); final ConcurrentHashMap<Channel, ClientChannelInfo> chlMap = entry.getValue(); Iterator<Entry<Channel, ClientChannelInfo>> it = chlMap.entrySet().iterator(); while (it.hasNext()) { Entry<Channel, ClientChannelInfo> item = it.next(); final ClientChannelInfo info = item.getValue(); long diff = System.currentTimeMillis() - info.getLastUpdateTimestamp(); if (diff > CHANNEL_EXPIRED_TIMEOUT) { it.remove(); clientChannelTable.remove(info.getClientId()); log.warn( "SCAN: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}", RemotingHelper.parseChannelRemoteAddr(info.getChannel()), group); RemotingUtil.closeChannel(info.getChannel()); } } } }
ProducerManager { public void scanNotActiveChannel() { for (final Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>> entry : this.groupChannelTable .entrySet()) { final String group = entry.getKey(); final ConcurrentHashMap<Channel, ClientChannelInfo> chlMap = entry.getValue(); Iterator<Entry<Channel, ClientChannelInfo>> it = chlMap.entrySet().iterator(); while (it.hasNext()) { Entry<Channel, ClientChannelInfo> item = it.next(); final ClientChannelInfo info = item.getValue(); long diff = System.currentTimeMillis() - info.getLastUpdateTimestamp(); if (diff > CHANNEL_EXPIRED_TIMEOUT) { it.remove(); clientChannelTable.remove(info.getClientId()); log.warn( "SCAN: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}", RemotingHelper.parseChannelRemoteAddr(info.getChannel()), group); RemotingUtil.closeChannel(info.getChannel()); } } } } }
ProducerManager { public void scanNotActiveChannel() { for (final Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>> entry : this.groupChannelTable .entrySet()) { final String group = entry.getKey(); final ConcurrentHashMap<Channel, ClientChannelInfo> chlMap = entry.getValue(); Iterator<Entry<Channel, ClientChannelInfo>> it = chlMap.entrySet().iterator(); while (it.hasNext()) { Entry<Channel, ClientChannelInfo> item = it.next(); final ClientChannelInfo info = item.getValue(); long diff = System.currentTimeMillis() - info.getLastUpdateTimestamp(); if (diff > CHANNEL_EXPIRED_TIMEOUT) { it.remove(); clientChannelTable.remove(info.getClientId()); log.warn( "SCAN: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}", RemotingHelper.parseChannelRemoteAddr(info.getChannel()), group); RemotingUtil.closeChannel(info.getChannel()); } } } } ProducerManager(); }
ProducerManager { public void scanNotActiveChannel() { for (final Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>> entry : this.groupChannelTable .entrySet()) { final String group = entry.getKey(); final ConcurrentHashMap<Channel, ClientChannelInfo> chlMap = entry.getValue(); Iterator<Entry<Channel, ClientChannelInfo>> it = chlMap.entrySet().iterator(); while (it.hasNext()) { Entry<Channel, ClientChannelInfo> item = it.next(); final ClientChannelInfo info = item.getValue(); long diff = System.currentTimeMillis() - info.getLastUpdateTimestamp(); if (diff > CHANNEL_EXPIRED_TIMEOUT) { it.remove(); clientChannelTable.remove(info.getClientId()); log.warn( "SCAN: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}", RemotingHelper.parseChannelRemoteAddr(info.getChannel()), group); RemotingUtil.closeChannel(info.getChannel()); } } } } ProducerManager(); ConcurrentHashMap<String, ConcurrentHashMap<Channel, ClientChannelInfo>> getGroupChannelTable(); void scanNotActiveChannel(); synchronized void doChannelCloseEvent(final String remoteAddr, final Channel channel); synchronized void registerProducer(final String group, final ClientChannelInfo clientChannelInfo); synchronized void unregisterProducer(final String group, final ClientChannelInfo clientChannelInfo); Channel getAvaliableChannel(String groupId); Channel findChannel(String clientId); }
ProducerManager { public void scanNotActiveChannel() { for (final Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>> entry : this.groupChannelTable .entrySet()) { final String group = entry.getKey(); final ConcurrentHashMap<Channel, ClientChannelInfo> chlMap = entry.getValue(); Iterator<Entry<Channel, ClientChannelInfo>> it = chlMap.entrySet().iterator(); while (it.hasNext()) { Entry<Channel, ClientChannelInfo> item = it.next(); final ClientChannelInfo info = item.getValue(); long diff = System.currentTimeMillis() - info.getLastUpdateTimestamp(); if (diff > CHANNEL_EXPIRED_TIMEOUT) { it.remove(); clientChannelTable.remove(info.getClientId()); log.warn( "SCAN: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}", RemotingHelper.parseChannelRemoteAddr(info.getChannel()), group); RemotingUtil.closeChannel(info.getChannel()); } } } } ProducerManager(); ConcurrentHashMap<String, ConcurrentHashMap<Channel, ClientChannelInfo>> getGroupChannelTable(); void scanNotActiveChannel(); synchronized void doChannelCloseEvent(final String remoteAddr, final Channel channel); synchronized void registerProducer(final String group, final ClientChannelInfo clientChannelInfo); synchronized void unregisterProducer(final String group, final ClientChannelInfo clientChannelInfo); Channel getAvaliableChannel(String groupId); Channel findChannel(String clientId);  }