### Question:
GossipHandler implements Function<MessageApi, CompletableFuture<ValidationResult>> { public void gossip(Bytes bytes) { if (!processedMessages.add(bytes)) { 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)); } GossipHandler(
      final Topic topic, final PubsubPublisherApi publisher, final TopicHandler handler); @Override SafeFuture<ValidationResult> apply(final MessageApi message); void gossip(Bytes bytes);  }### Answer:
@Test public void gossip_newMessage() { final Bytes message = Bytes.fromHexString("0x01"); gossipHandler.gossip(message); verify(publisher).publish(toByteBuf(message), topic); }
@Test public void gossip_duplicateMessage() { final Bytes message = Bytes.fromHexString("0x01"); gossipHandler.gossip(message); gossipHandler.gossip(message); verify(publisher).publish(toByteBuf(message), topic); }
@Test public void gossip_distinctMessages() { final Bytes message1 = Bytes.fromHexString("0x01"); final Bytes message2 = Bytes.fromHexString("0x01"); gossipHandler.gossip(message1); gossipHandler.gossip(message2); verify(publisher).publish(toByteBuf(message1), topic); verify(publisher).publish(toByteBuf(message2), topic); }