@Test public void testSendWhenHealthy() throws Exception { when(healthchecker.isHealthy()).thenReturn(true); final String topic = "Event"; sender.send(topic, new byte['x']); verify(pubsub).publishAsync(eq(prefix + topic), any(Message.class)); }
@Override public void send(final String topic, final byte[] message) { final String combinedTopic = topicPrefix + topic; if (!healthchecker.isHealthy()) { log.warn("will not publish message to pubsub topic={} as the pubsub client " + "appears to be unhealthy", combinedTopic); return; } try { Futures.addCallback( JdkFutureAdapters.listenInPoolThread( pubsub.publishAsync(combinedTopic, Message.of(ByteArray.copyFrom(message)))), new FutureCallback<String>() { @Override public void onSuccess(@Nullable final String ackId) { log.debug("Sent an event to Google PubSub, topic: {}, ack: {}", combinedTopic, ackId); } @Override public void onFailure(final Throwable th) { log.warn("Unable to send an event to Google PubSub, topic: {}", combinedTopic, th); } }, MoreExecutors.directExecutor()); } catch (Exception e) { log.warn("Failed to publish Google PubSub message, topic: {}", combinedTopic, e); } }
GooglePubSubSender implements EventSender { @Override public void send(final String topic, final byte[] message) { final String combinedTopic = topicPrefix + topic; if (!healthchecker.isHealthy()) { log.warn("will not publish message to pubsub topic={} as the pubsub client " + "appears to be unhealthy", combinedTopic); return; } try { Futures.addCallback( JdkFutureAdapters.listenInPoolThread( pubsub.publishAsync(combinedTopic, Message.of(ByteArray.copyFrom(message)))), new FutureCallback<String>() { @Override public void onSuccess(@Nullable final String ackId) { log.debug("Sent an event to Google PubSub, topic: {}, ack: {}", combinedTopic, ackId); } @Override public void onFailure(final Throwable th) { log.warn("Unable to send an event to Google PubSub, topic: {}", combinedTopic, th); } }, MoreExecutors.directExecutor()); } catch (Exception e) { log.warn("Failed to publish Google PubSub message, topic: {}", combinedTopic, e); } } }
GooglePubSubSender implements EventSender { @Override public void send(final String topic, final byte[] message) { final String combinedTopic = topicPrefix + topic; if (!healthchecker.isHealthy()) { log.warn("will not publish message to pubsub topic={} as the pubsub client " + "appears to be unhealthy", combinedTopic); return; } try { Futures.addCallback( JdkFutureAdapters.listenInPoolThread( pubsub.publishAsync(combinedTopic, Message.of(ByteArray.copyFrom(message)))), new FutureCallback<String>() { @Override public void onSuccess(@Nullable final String ackId) { log.debug("Sent an event to Google PubSub, topic: {}, ack: {}", combinedTopic, ackId); } @Override public void onFailure(final Throwable th) { log.warn("Unable to send an event to Google PubSub, topic: {}", combinedTopic, th); } }, MoreExecutors.directExecutor()); } catch (Exception e) { log.warn("Failed to publish Google PubSub message, topic: {}", combinedTopic, e); } } private  GooglePubSubSender(final PubSub pubSub,
                             final String topicPrefix,
                             final HealthChecker healthchecker); }
GooglePubSubSender implements EventSender { @Override public void send(final String topic, final byte[] message) { final String combinedTopic = topicPrefix + topic; if (!healthchecker.isHealthy()) { log.warn("will not publish message to pubsub topic={} as the pubsub client " + "appears to be unhealthy", combinedTopic); return; } try { Futures.addCallback( JdkFutureAdapters.listenInPoolThread( pubsub.publishAsync(combinedTopic, Message.of(ByteArray.copyFrom(message)))), new FutureCallback<String>() { @Override public void onSuccess(@Nullable final String ackId) { log.debug("Sent an event to Google PubSub, topic: {}, ack: {}", combinedTopic, ackId); } @Override public void onFailure(final Throwable th) { log.warn("Unable to send an event to Google PubSub, topic: {}", combinedTopic, th); } }, MoreExecutors.directExecutor()); } catch (Exception e) { log.warn("Failed to publish Google PubSub message, topic: {}", combinedTopic, e); } } private  GooglePubSubSender(final PubSub pubSub,
                             final String topicPrefix,
                             final HealthChecker healthchecker); static GooglePubSubSender create(final PubSub pubSub,
                                          final String topicPrefix,
                                          final HealthChecker healthchecker); @Override void start(); @Override void stop(); @Override void send(final String topic, final byte[] message); }
GooglePubSubSender implements EventSender { @Override public void send(final String topic, final byte[] message) { final String combinedTopic = topicPrefix + topic; if (!healthchecker.isHealthy()) { log.warn("will not publish message to pubsub topic={} as the pubsub client " + "appears to be unhealthy", combinedTopic); return; } try { Futures.addCallback( JdkFutureAdapters.listenInPoolThread( pubsub.publishAsync(combinedTopic, Message.of(ByteArray.copyFrom(message)))), new FutureCallback<String>() { @Override public void onSuccess(@Nullable final String ackId) { log.debug("Sent an event to Google PubSub, topic: {}, ack: {}", combinedTopic, ackId); } @Override public void onFailure(final Throwable th) { log.warn("Unable to send an event to Google PubSub, topic: {}", combinedTopic, th); } }, MoreExecutors.directExecutor()); } catch (Exception e) { log.warn("Failed to publish Google PubSub message, topic: {}", combinedTopic, e); } } private  GooglePubSubSender(final PubSub pubSub,
                             final String topicPrefix,
                             final HealthChecker healthchecker); static GooglePubSubSender create(final PubSub pubSub,
                                          final String topicPrefix,
                                          final HealthChecker healthchecker); @Override void start(); @Override void stop(); @Override void send(final String topic, final byte[] message);  }