@Test public void publishMessageToTopicWithNoListeners() throws Exception { latch0.set(new CountDownLatch(1)); helixNotifier.publish(refTopics.get(0), refMessages.get(0)); try { awaitLatchOrTimeout(latch0.get(), 1); fail("should have thrown"); } catch (TimeoutException e) { } }
@Override public boolean publish(String topic, String message) { if (topic == null) { throw new IllegalArgumentException("topic cannot be null"); } if (message == null) { throw new IllegalArgumentException("message cannot be null"); } String topicPath = getTopicPath(topic); ZNRecord record = new ZNRecord(topicPath); record.setSimpleField(MESSAGE_KEY, message); boolean res = helixStore.set(topicPath, record, AccessOption.PERSISTENT); if (res) { logger.trace("message={} has been published for topic={}", message, topic); } else { logger.error("failed to publish message={} for topic={}", message, topic); } return res; }
HelixNotifier implements Notifier<String> { @Override public boolean publish(String topic, String message) { if (topic == null) { throw new IllegalArgumentException("topic cannot be null"); } if (message == null) { throw new IllegalArgumentException("message cannot be null"); } String topicPath = getTopicPath(topic); ZNRecord record = new ZNRecord(topicPath); record.setSimpleField(MESSAGE_KEY, message); boolean res = helixStore.set(topicPath, record, AccessOption.PERSISTENT); if (res) { logger.trace("message={} has been published for topic={}", message, topic); } else { logger.error("failed to publish message={} for topic={}", message, topic); } return res; } }
HelixNotifier implements Notifier<String> { @Override public boolean publish(String topic, String message) { if (topic == null) { throw new IllegalArgumentException("topic cannot be null"); } if (message == null) { throw new IllegalArgumentException("message cannot be null"); } String topicPath = getTopicPath(topic); ZNRecord record = new ZNRecord(topicPath); record.setSimpleField(MESSAGE_KEY, message); boolean res = helixStore.set(topicPath, record, AccessOption.PERSISTENT); if (res) { logger.trace("message={} has been published for topic={}", message, topic); } else { logger.error("failed to publish message={} for topic={}", message, topic); } return res; } HelixNotifier(HelixPropertyStore<ZNRecord> helixStore);  HelixNotifier(String zkServers, HelixPropertyStoreConfig storeConfig); }
HelixNotifier implements Notifier<String> { @Override public boolean publish(String topic, String message) { if (topic == null) { throw new IllegalArgumentException("topic cannot be null"); } if (message == null) { throw new IllegalArgumentException("message cannot be null"); } String topicPath = getTopicPath(topic); ZNRecord record = new ZNRecord(topicPath); record.setSimpleField(MESSAGE_KEY, message); boolean res = helixStore.set(topicPath, record, AccessOption.PERSISTENT); if (res) { logger.trace("message={} has been published for topic={}", message, topic); } else { logger.error("failed to publish message={} for topic={}", message, topic); } return res; } HelixNotifier(HelixPropertyStore<ZNRecord> helixStore);  HelixNotifier(String zkServers, HelixPropertyStoreConfig storeConfig); @Override boolean publish(String topic, String message); @Override void subscribe(String topic, TopicListener<String> topicListener); @Override void unsubscribe(String topic, TopicListener<String> topicListener); }
HelixNotifier implements Notifier<String> { @Override public boolean publish(String topic, String message) { if (topic == null) { throw new IllegalArgumentException("topic cannot be null"); } if (message == null) { throw new IllegalArgumentException("message cannot be null"); } String topicPath = getTopicPath(topic); ZNRecord record = new ZNRecord(topicPath); record.setSimpleField(MESSAGE_KEY, message); boolean res = helixStore.set(topicPath, record, AccessOption.PERSISTENT); if (res) { logger.trace("message={} has been published for topic={}", message, topic); } else { logger.error("failed to publish message={} for topic={}", message, topic); } return res; } HelixNotifier(HelixPropertyStore<ZNRecord> helixStore);  HelixNotifier(String zkServers, HelixPropertyStoreConfig storeConfig); @Override boolean publish(String topic, String message); @Override void subscribe(String topic, TopicListener<String> topicListener); @Override void unsubscribe(String topic, TopicListener<String> topicListener);  }