@Test public void testAddMessageListener() { assertEquals(0, msgCount.get()); topic.publish(MessageType.MESSAGE); assertEquals(0, msgCount.get()); topic.addMessageListener(this); topic.publish(MessageType.MESSAGE); assertEquals(1, msgCount.get()); topic.removeMessageListener(this); }
@Override public void addMessageListener(GroupTopicListener<T> listener) { synchronized (listeners) { if (false == listeners.containsKey(listener)) listeners.put(listener, new AtomicInteger()); listeners.get(listener).incrementAndGet(); } }
LocalTopic implements GroupTopic<T> { @Override public void addMessageListener(GroupTopicListener<T> listener) { synchronized (listeners) { if (false == listeners.containsKey(listener)) listeners.put(listener, new AtomicInteger()); listeners.get(listener).incrementAndGet(); } } }
LocalTopic implements GroupTopic<T> { @Override public void addMessageListener(GroupTopicListener<T> listener) { synchronized (listeners) { if (false == listeners.containsKey(listener)) listeners.put(listener, new AtomicInteger()); listeners.get(listener).incrementAndGet(); } }  }
LocalTopic implements GroupTopic<T> { @Override public void addMessageListener(GroupTopicListener<T> listener) { synchronized (listeners) { if (false == listeners.containsKey(listener)) listeners.put(listener, new AtomicInteger()); listeners.get(listener).incrementAndGet(); } }  @Override void addMessageListener(GroupTopicListener<T> listener); @Override void removeMessageListener(GroupTopicListener<T> listener); @Override void publish(final T message); }
LocalTopic implements GroupTopic<T> { @Override public void addMessageListener(GroupTopicListener<T> listener) { synchronized (listeners) { if (false == listeners.containsKey(listener)) listeners.put(listener, new AtomicInteger()); listeners.get(listener).incrementAndGet(); } }  @Override void addMessageListener(GroupTopicListener<T> listener); @Override void removeMessageListener(GroupTopicListener<T> listener); @Override void publish(final T message);  }