@Test public void testPublish() { topic.addMessageListener(this); topic.publish(MessageType.MESSAGE); assertEquals(1, msgCount.get()); topic.publish(MessageType.MESSAGE); assertEquals(2, msgCount.get()); topic.removeMessageListener(this); }
@Override public void publish(final T message) { Set<GroupTopicListener<T>> messageReceivers; synchronized (listeners) { messageReceivers = new HashSet<GroupTopicListener<T>>(listeners.keySet()); } final CountDownLatch latch = new CountDownLatch(messageReceivers.size()); for (final GroupTopicListener<T> l : messageReceivers) threadPool.execute(new Runnable() { @Override public void run() { l.onMessage(message); latch.countDown(); } }); try { latch.await(); } catch (InterruptedException e) { log.warn(e); } }
LocalTopic implements GroupTopic<T> { @Override public void publish(final T message) { Set<GroupTopicListener<T>> messageReceivers; synchronized (listeners) { messageReceivers = new HashSet<GroupTopicListener<T>>(listeners.keySet()); } final CountDownLatch latch = new CountDownLatch(messageReceivers.size()); for (final GroupTopicListener<T> l : messageReceivers) threadPool.execute(new Runnable() { @Override public void run() { l.onMessage(message); latch.countDown(); } }); try { latch.await(); } catch (InterruptedException e) { log.warn(e); } } }
LocalTopic implements GroupTopic<T> { @Override public void publish(final T message) { Set<GroupTopicListener<T>> messageReceivers; synchronized (listeners) { messageReceivers = new HashSet<GroupTopicListener<T>>(listeners.keySet()); } final CountDownLatch latch = new CountDownLatch(messageReceivers.size()); for (final GroupTopicListener<T> l : messageReceivers) threadPool.execute(new Runnable() { @Override public void run() { l.onMessage(message); latch.countDown(); } }); try { latch.await(); } catch (InterruptedException e) { log.warn(e); } }  }
LocalTopic implements GroupTopic<T> { @Override public void publish(final T message) { Set<GroupTopicListener<T>> messageReceivers; synchronized (listeners) { messageReceivers = new HashSet<GroupTopicListener<T>>(listeners.keySet()); } final CountDownLatch latch = new CountDownLatch(messageReceivers.size()); for (final GroupTopicListener<T> l : messageReceivers) threadPool.execute(new Runnable() { @Override public void run() { l.onMessage(message); latch.countDown(); } }); try { latch.await(); } catch (InterruptedException e) { log.warn(e); } }  @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 publish(final T message) { Set<GroupTopicListener<T>> messageReceivers; synchronized (listeners) { messageReceivers = new HashSet<GroupTopicListener<T>>(listeners.keySet()); } final CountDownLatch latch = new CountDownLatch(messageReceivers.size()); for (final GroupTopicListener<T> l : messageReceivers) threadPool.execute(new Runnable() { @Override public void run() { l.onMessage(message); latch.countDown(); } }); try { latch.await(); } catch (InterruptedException e) { log.warn(e); } }  @Override void addMessageListener(GroupTopicListener<T> listener); @Override void removeMessageListener(GroupTopicListener<T> listener); @Override void publish(final T message);  }