@Test public void handleMessage_valid() { topicHandler.setValidationResult(InternalValidationResult.ACCEPT); final SafeFuture<ValidationResult> result = topicHandler.handleMessage(blockBytes); asyncRunner.executeQueuedActions(); assertThatSafeFuture(result).isCompletedWithValue(ValidationResult.Valid); }
@Override public SafeFuture<ValidationResult> handleMessage(final Bytes bytes) { return SafeFuture.of(() -> deserialize(bytes)) .thenApply(this::wrapMessage) .thenCompose( wrapped -> asyncRunner.runAsync( () -> validateData(wrapped) .thenApply( internalValidation -> { processMessage(wrapped, internalValidation); return internalValidation.getGossipSubValidationResult(); }))) .exceptionally(this::handleMessageProcessingError); }
Eth2TopicHandler implements TopicHandler { @Override public SafeFuture<ValidationResult> handleMessage(final Bytes bytes) { return SafeFuture.of(() -> deserialize(bytes)) .thenApply(this::wrapMessage) .thenCompose( wrapped -> asyncRunner.runAsync( () -> validateData(wrapped) .thenApply( internalValidation -> { processMessage(wrapped, internalValidation); return internalValidation.getGossipSubValidationResult(); }))) .exceptionally(this::handleMessageProcessingError); } }
Eth2TopicHandler implements TopicHandler { @Override public SafeFuture<ValidationResult> handleMessage(final Bytes bytes) { return SafeFuture.of(() -> deserialize(bytes)) .thenApply(this::wrapMessage) .thenCompose( wrapped -> asyncRunner.runAsync( () -> validateData(wrapped) .thenApply( internalValidation -> { processMessage(wrapped, internalValidation); return internalValidation.getGossipSubValidationResult(); }))) .exceptionally(this::handleMessageProcessingError); } protected  Eth2TopicHandler(final AsyncRunner asyncRunner); }
Eth2TopicHandler implements TopicHandler { @Override public SafeFuture<ValidationResult> handleMessage(final Bytes bytes) { return SafeFuture.of(() -> deserialize(bytes)) .thenApply(this::wrapMessage) .thenCompose( wrapped -> asyncRunner.runAsync( () -> validateData(wrapped) .thenApply( internalValidation -> { processMessage(wrapped, internalValidation); return internalValidation.getGossipSubValidationResult(); }))) .exceptionally(this::handleMessageProcessingError); } protected  Eth2TopicHandler(final AsyncRunner asyncRunner); @Override SafeFuture<ValidationResult> handleMessage(final Bytes bytes); T deserialize(Bytes bytes); String getTopic(); abstract Bytes4 getForkDigest(); abstract GossipEncoding getGossipEncoding(); abstract String getTopicName(); abstract Class<T> getValueType(); }
Eth2TopicHandler implements TopicHandler { @Override public SafeFuture<ValidationResult> handleMessage(final Bytes bytes) { return SafeFuture.of(() -> deserialize(bytes)) .thenApply(this::wrapMessage) .thenCompose( wrapped -> asyncRunner.runAsync( () -> validateData(wrapped) .thenApply( internalValidation -> { processMessage(wrapped, internalValidation); return internalValidation.getGossipSubValidationResult(); }))) .exceptionally(this::handleMessageProcessingError); } protected  Eth2TopicHandler(final AsyncRunner asyncRunner); @Override SafeFuture<ValidationResult> handleMessage(final Bytes bytes); T deserialize(Bytes bytes); String getTopic(); abstract Bytes4 getForkDigest(); abstract GossipEncoding getGossipEncoding(); abstract String getTopicName(); abstract Class<T> getValueType();  }