@Test public void testGetAvaliableChannel() { producerManager.registerProducer(group, clientInfo); when(channel.isActive()).thenReturn(true); when(channel.isWritable()).thenReturn(true); Channel c = producerManager.getAvaliableChannel(group); assertThat(c).isSameAs(channel); when(channel.isWritable()).thenReturn(false); c = producerManager.getAvaliableChannel(group); assertThat(c).isSameAs(channel); when(channel.isActive()).thenReturn(false); c = producerManager.getAvaliableChannel(group); assertThat(c).isNull(); }
public Channel getAvaliableChannel(String groupId) { if (groupId == null) { return null; } List<Channel> channelList = new ArrayList<Channel>(); ConcurrentHashMap<Channel, ClientChannelInfo> channelClientChannelInfoHashMap = groupChannelTable.get(groupId); if (channelClientChannelInfoHashMap != null) { for (Channel channel : channelClientChannelInfoHashMap.keySet()) { channelList.add(channel); } } else { log.warn("Check transaction failed, channel table is empty. groupId={}", groupId); return null; } int size = channelList.size(); if (0 == size) { log.warn("Channel list is empty. groupId={}", groupId); return null; } Channel lastActiveChannel = null; int index = positiveAtomicCounter.incrementAndGet() % size; Channel channel = channelList.get(index); int count = 0; boolean isOk = channel.isActive() && channel.isWritable(); while (count++ < GET_AVALIABLE_CHANNEL_RETRY_COUNT) { if (isOk) { return channel; } if (channel.isActive()) { lastActiveChannel = channel; } index = (++index) % size; channel = channelList.get(index); isOk = channel.isActive() && channel.isWritable(); } return lastActiveChannel; }
ProducerManager { public Channel getAvaliableChannel(String groupId) { if (groupId == null) { return null; } List<Channel> channelList = new ArrayList<Channel>(); ConcurrentHashMap<Channel, ClientChannelInfo> channelClientChannelInfoHashMap = groupChannelTable.get(groupId); if (channelClientChannelInfoHashMap != null) { for (Channel channel : channelClientChannelInfoHashMap.keySet()) { channelList.add(channel); } } else { log.warn("Check transaction failed, channel table is empty. groupId={}", groupId); return null; } int size = channelList.size(); if (0 == size) { log.warn("Channel list is empty. groupId={}", groupId); return null; } Channel lastActiveChannel = null; int index = positiveAtomicCounter.incrementAndGet() % size; Channel channel = channelList.get(index); int count = 0; boolean isOk = channel.isActive() && channel.isWritable(); while (count++ < GET_AVALIABLE_CHANNEL_RETRY_COUNT) { if (isOk) { return channel; } if (channel.isActive()) { lastActiveChannel = channel; } index = (++index) % size; channel = channelList.get(index); isOk = channel.isActive() && channel.isWritable(); } return lastActiveChannel; } }
ProducerManager { public Channel getAvaliableChannel(String groupId) { if (groupId == null) { return null; } List<Channel> channelList = new ArrayList<Channel>(); ConcurrentHashMap<Channel, ClientChannelInfo> channelClientChannelInfoHashMap = groupChannelTable.get(groupId); if (channelClientChannelInfoHashMap != null) { for (Channel channel : channelClientChannelInfoHashMap.keySet()) { channelList.add(channel); } } else { log.warn("Check transaction failed, channel table is empty. groupId={}", groupId); return null; } int size = channelList.size(); if (0 == size) { log.warn("Channel list is empty. groupId={}", groupId); return null; } Channel lastActiveChannel = null; int index = positiveAtomicCounter.incrementAndGet() % size; Channel channel = channelList.get(index); int count = 0; boolean isOk = channel.isActive() && channel.isWritable(); while (count++ < GET_AVALIABLE_CHANNEL_RETRY_COUNT) { if (isOk) { return channel; } if (channel.isActive()) { lastActiveChannel = channel; } index = (++index) % size; channel = channelList.get(index); isOk = channel.isActive() && channel.isWritable(); } return lastActiveChannel; } ProducerManager(); }
ProducerManager { public Channel getAvaliableChannel(String groupId) { if (groupId == null) { return null; } List<Channel> channelList = new ArrayList<Channel>(); ConcurrentHashMap<Channel, ClientChannelInfo> channelClientChannelInfoHashMap = groupChannelTable.get(groupId); if (channelClientChannelInfoHashMap != null) { for (Channel channel : channelClientChannelInfoHashMap.keySet()) { channelList.add(channel); } } else { log.warn("Check transaction failed, channel table is empty. groupId={}", groupId); return null; } int size = channelList.size(); if (0 == size) { log.warn("Channel list is empty. groupId={}", groupId); return null; } Channel lastActiveChannel = null; int index = positiveAtomicCounter.incrementAndGet() % size; Channel channel = channelList.get(index); int count = 0; boolean isOk = channel.isActive() && channel.isWritable(); while (count++ < GET_AVALIABLE_CHANNEL_RETRY_COUNT) { if (isOk) { return channel; } if (channel.isActive()) { lastActiveChannel = channel; } index = (++index) % size; channel = channelList.get(index); isOk = channel.isActive() && channel.isWritable(); } return lastActiveChannel; } 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 Channel getAvaliableChannel(String groupId) { if (groupId == null) { return null; } List<Channel> channelList = new ArrayList<Channel>(); ConcurrentHashMap<Channel, ClientChannelInfo> channelClientChannelInfoHashMap = groupChannelTable.get(groupId); if (channelClientChannelInfoHashMap != null) { for (Channel channel : channelClientChannelInfoHashMap.keySet()) { channelList.add(channel); } } else { log.warn("Check transaction failed, channel table is empty. groupId={}", groupId); return null; } int size = channelList.size(); if (0 == size) { log.warn("Channel list is empty. groupId={}", groupId); return null; } Channel lastActiveChannel = null; int index = positiveAtomicCounter.incrementAndGet() % size; Channel channel = channelList.get(index); int count = 0; boolean isOk = channel.isActive() && channel.isWritable(); while (count++ < GET_AVALIABLE_CHANNEL_RETRY_COUNT) { if (isOk) { return channel; } if (channel.isActive()) { lastActiveChannel = channel; } index = (++index) % size; channel = channelList.get(index); isOk = channel.isActive() && channel.isWritable(); } return lastActiveChannel; } 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);  }