@Test public void testPurgeExpired() throws Exception { String consumerName = "c1"; String ip = "1.1.1.1"; long heartbeatTime = 1L; m_list.heartbeat(consumerName, heartbeatTime, ip); m_list.purgeExpired(10, 12L); Map<String, ClientContext> activeConsumers = m_list.getActiveConsumers(); assertEquals(0, activeConsumers.size()); assertTrue(m_list.getAndResetChanged()); m_list.heartbeat(consumerName, heartbeatTime, ip); m_list.purgeExpired(10, 1L); activeConsumers = m_list.getActiveConsumers(); assertEquals(1, activeConsumers.size()); TestHelper.assertClientContextEquals(consumerName, ip, heartbeatTime, activeConsumers.get(consumerName)); assertTrue(m_list.getAndResetChanged()); }
public void purgeExpired(long timeoutMillis, long now) { Iterator<Entry<String, ClientContext>> iter = m_consumers.entrySet().iterator(); while (iter.hasNext()) { Entry<String, ClientContext> entry = iter.next(); if (entry.getValue().getLastHeartbeatTime() + timeoutMillis < now) { iter.remove(); m_changed = true; } } }
ActiveConsumerList { public void purgeExpired(long timeoutMillis, long now) { Iterator<Entry<String, ClientContext>> iter = m_consumers.entrySet().iterator(); while (iter.hasNext()) { Entry<String, ClientContext> entry = iter.next(); if (entry.getValue().getLastHeartbeatTime() + timeoutMillis < now) { iter.remove(); m_changed = true; } } } }
ActiveConsumerList { public void purgeExpired(long timeoutMillis, long now) { Iterator<Entry<String, ClientContext>> iter = m_consumers.entrySet().iterator(); while (iter.hasNext()) { Entry<String, ClientContext> entry = iter.next(); if (entry.getValue().getLastHeartbeatTime() + timeoutMillis < now) { iter.remove(); m_changed = true; } } }  }
ActiveConsumerList { public void purgeExpired(long timeoutMillis, long now) { Iterator<Entry<String, ClientContext>> iter = m_consumers.entrySet().iterator(); while (iter.hasNext()) { Entry<String, ClientContext> entry = iter.next(); if (entry.getValue().getLastHeartbeatTime() + timeoutMillis < now) { iter.remove(); m_changed = true; } } }  void heartbeat(String consumerName, long heartbeatTime, String ip); void purgeExpired(long timeoutMillis, long now); boolean getAndResetChanged(); Map<String, ClientContext> getActiveConsumers(); }
ActiveConsumerList { public void purgeExpired(long timeoutMillis, long now) { Iterator<Entry<String, ClientContext>> iter = m_consumers.entrySet().iterator(); while (iter.hasNext()) { Entry<String, ClientContext> entry = iter.next(); if (entry.getValue().getLastHeartbeatTime() + timeoutMillis < now) { iter.remove(); m_changed = true; } } }  void heartbeat(String consumerName, long heartbeatTime, String ip); void purgeExpired(long timeoutMillis, long now); boolean getAndResetChanged(); Map<String, ClientContext> getActiveConsumers();  }