@Test public void testPublish() throws Exception { messageReceivedLatch = new CountDownLatch(1); BiConsumer<String, byte[]> subscriber = this::subscriber; pubSubService.registerSubscriber("topicTest", subscriber); pubSubService.publish("topicTest", "foo".getBytes(UTF_8)); if (!messageReceivedLatch.await(5, TimeUnit.SECONDS)) { fail("message not received in 5s"); } assertEquals(Collections.singletonList("topicTest=foo"), messages); pubSubService.unregisterSubscriber("topicTest", subscriber); messages.clear(); pubSubService.publish("topicTest", "bar".getBytes(UTF_8)); Thread.sleep(500); assertEquals(Collections.emptyList(), messages); }
@Override public void publish(String topic, byte[] message) { localPublish(topic, message); }
MemPubSubProvider extends AbstractPubSubProvider { @Override public void publish(String topic, byte[] message) { localPublish(topic, message); } }
MemPubSubProvider extends AbstractPubSubProvider { @Override public void publish(String topic, byte[] message) { localPublish(topic, message); }  }
MemPubSubProvider extends AbstractPubSubProvider { @Override public void publish(String topic, byte[] message) { localPublish(topic, message); }  @Override void initialize(Map<String, String> options, Map<String, List<BiConsumer<String, byte[]>>> subscribers); @Override void close(); @Override void publish(String topic, byte[] message); }
MemPubSubProvider extends AbstractPubSubProvider { @Override public void publish(String topic, byte[] message) { localPublish(topic, message); }  @Override void initialize(Map<String, String> options, Map<String, List<BiConsumer<String, byte[]>>> subscribers); @Override void close(); @Override void publish(String topic, byte[] message);  }