@Test public void testHeartbeat() throws Exception { Pair<String, String> t1g1 = new Pair<>("t1", "g1"); Pair<String, String> t1g2 = new Pair<>("t1", "g2"); Pair<String, String> t2g1 = new Pair<>("t2", "g1"); m_holder.heartbeat(t1g1, "c1", "1.1.1.1"); m_holder.heartbeat(t1g1, "c2", "1.1.1.2"); m_holder.heartbeat(t1g2, "c3", "1.1.1.3"); m_holder.heartbeat(t2g1, "c1", "1.1.1.1"); ActiveConsumerList activeConsumerList = m_holder.getActiveConsumerList(t1g1); Map<String, ClientContext> activeConsumers = activeConsumerList.getActiveConsumers(); assertEquals(2, activeConsumers.size()); TestHelper.assertClientContextEquals("c1", "1.1.1.1", activeConsumers.get("c1")); TestHelper.assertClientContextEquals("c2", "1.1.1.2", activeConsumers.get("c2")); activeConsumerList = m_holder.getActiveConsumerList(t1g2); activeConsumers = activeConsumerList.getActiveConsumers(); assertEquals(1, activeConsumers.size()); TestHelper.assertClientContextEquals("c3", "1.1.1.3", activeConsumers.get("c3")); activeConsumerList = m_holder.getActiveConsumerList(t2g1); activeConsumers = activeConsumerList.getActiveConsumers(); assertEquals(1, activeConsumers.size()); TestHelper.assertClientContextEquals("c1", "1.1.1.1", activeConsumers.get("c1")); }
public void heartbeat(Pair<String, String> topicGroup, String consumerName, String ip) { m_lock.lock(); try { if (!m_activeConsumerLists.containsKey(topicGroup)) { m_activeConsumerLists.put(topicGroup, new ActiveConsumerList()); } ActiveConsumerList activeConsumerList = m_activeConsumerLists.get(topicGroup); activeConsumerList.heartbeat(consumerName, m_systemClockService.now(), ip); } finally { m_lock.unlock(); } }
ActiveConsumerListHolder { public void heartbeat(Pair<String, String> topicGroup, String consumerName, String ip) { m_lock.lock(); try { if (!m_activeConsumerLists.containsKey(topicGroup)) { m_activeConsumerLists.put(topicGroup, new ActiveConsumerList()); } ActiveConsumerList activeConsumerList = m_activeConsumerLists.get(topicGroup); activeConsumerList.heartbeat(consumerName, m_systemClockService.now(), ip); } finally { m_lock.unlock(); } } }
ActiveConsumerListHolder { public void heartbeat(Pair<String, String> topicGroup, String consumerName, String ip) { m_lock.lock(); try { if (!m_activeConsumerLists.containsKey(topicGroup)) { m_activeConsumerLists.put(topicGroup, new ActiveConsumerList()); } ActiveConsumerList activeConsumerList = m_activeConsumerLists.get(topicGroup); activeConsumerList.heartbeat(consumerName, m_systemClockService.now(), ip); } finally { m_lock.unlock(); } }  }
ActiveConsumerListHolder { public void heartbeat(Pair<String, String> topicGroup, String consumerName, String ip) { m_lock.lock(); try { if (!m_activeConsumerLists.containsKey(topicGroup)) { m_activeConsumerLists.put(topicGroup, new ActiveConsumerList()); } ActiveConsumerList activeConsumerList = m_activeConsumerLists.get(topicGroup); activeConsumerList.heartbeat(consumerName, m_systemClockService.now(), ip); } finally { m_lock.unlock(); } }  void setSystemClockService(SystemClockService systemClockService); void heartbeat(Pair<String, String> topicGroup, String consumerName, String ip); Map<Pair<String, String>, Map<String, ClientContext>> scanChanges(long timeout, TimeUnit timeUnit); ActiveConsumerList getActiveConsumerList(Pair<String, String> topicGroup); }
ActiveConsumerListHolder { public void heartbeat(Pair<String, String> topicGroup, String consumerName, String ip) { m_lock.lock(); try { if (!m_activeConsumerLists.containsKey(topicGroup)) { m_activeConsumerLists.put(topicGroup, new ActiveConsumerList()); } ActiveConsumerList activeConsumerList = m_activeConsumerLists.get(topicGroup); activeConsumerList.heartbeat(consumerName, m_systemClockService.now(), ip); } finally { m_lock.unlock(); } }  void setSystemClockService(SystemClockService systemClockService); void heartbeat(Pair<String, String> topicGroup, String consumerName, String ip); Map<Pair<String, String>, Map<String, ClientContext>> scanChanges(long timeout, TimeUnit timeUnit); ActiveConsumerList getActiveConsumerList(Pair<String, String> topicGroup);  }