class GossipHandler implements Function<MessageApi, CompletableFuture<ValidationResult>> {

  public void gossip(Bytes bytes) {
    if (!processedMessages.add(bytes)) {
      // We've already gossiped this data
      return;
    }

    LOG.trace("Gossiping {}: {} bytes", topic, bytes.size());
    SafeFuture.of(publisher.publish(Unpooled.wrappedBuffer(bytes.toArrayUnsafe()), topic))
        .finish(
            () -> LOG.trace("Successfully gossiped message on {}", topic),
            err -> LOG.debug("Failed to gossip message on " + topic, err));
  }

  public  GossipHandler(
      final Topic topic, final PubsubPublisherApi publisher, final TopicHandler handler);

}