Add AbstractSkippingEnvelopeHandler so all handlers will skip BAD_PACKET (#218)

Author: zilm13

src/main/java/org/ethereum/beacon/discovery/pipeline/AbstractSkippingEnvelopeHandler.java

@@ -0,0 +1,24 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.ethereum.beacon.discovery.pipeline;
+
+/**
+ * Base class for {@link EnvelopeHandler}s that should be skipped once an envelope has been marked
+ * with {@link Field#BAD_PACKET}. Subclasses implement {@link #handlePacket(Envelope)} and get the
+ * skip behaviour for free; the terminal bad-packet handler implements {@link EnvelopeHandler}
+ * directly.
+ */
+public abstract class AbstractSkippingEnvelopeHandler implements EnvelopeHandler {
+
+  @Override
+  public final void handle(final Envelope envelope) {
+    if (envelope.contains(Field.BAD_PACKET)) {
+      return;
+    }
+    handlePacket(envelope);
+  }
+
+  protected abstract void handlePacket(Envelope envelope);
+}

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/HandshakeMessagePacketHandler.java

@@ -13,8 +13,8 @@
 import org.ethereum.beacon.discovery.AddressAccessPolicy;
 import org.ethereum.beacon.discovery.message.V5Message;
 import org.ethereum.beacon.discovery.packet.HandshakeMessagePacket;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import org.ethereum.beacon.discovery.pipeline.Pipeline;
@@ -27,7 +27,7 @@
 import org.ethereum.beacon.discovery.util.Functions;
 
 /** Handles {@link HandshakeMessagePacket} in {@link Field#PACKET_HANDSHAKE} field */
-public class HandshakeMessagePacketHandler implements EnvelopeHandler {
+public class HandshakeMessagePacketHandler extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(HandshakeMessagePacketHandler.class);
   private final Pipeline outgoingPipeline;
   private final Scheduler scheduler;
@@ -49,7 +49,7 @@ public HandshakeMessagePacketHandler(
   }
 
   @Override
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireField(Field.PACKET_HANDSHAKE, envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/IncomingDataPacker.java

@@ -9,15 +9,15 @@
 import org.apache.tuweni.bytes.Bytes;
 import org.ethereum.beacon.discovery.packet.Packet;
 import org.ethereum.beacon.discovery.packet.RawPacket;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import org.ethereum.beacon.discovery.type.Bytes16;
 import org.ethereum.beacon.discovery.util.DecodeException;
 
 /** Handles raw BytesValue incoming data in {@link Field#INCOMING} */
-public class IncomingDataPacker implements EnvelopeHandler {
+public class IncomingDataPacker extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(IncomingDataPacker.class);
   public static final int MAX_PACKET_SIZE = 1280;
   public static final int MIN_PACKET_SIZE = 63;
@@ -28,7 +28,7 @@ public IncomingDataPacker(Bytes homeNodeId) {
   }
 
   @Override
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireField(Field.INCOMING, envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/MessageHandler.java

@@ -10,16 +10,16 @@
 import org.ethereum.beacon.discovery.message.V5Message;
 import org.ethereum.beacon.discovery.message.handler.EnrUpdateTracker.EnrUpdater;
 import org.ethereum.beacon.discovery.message.handler.ExternalAddressSelector;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import org.ethereum.beacon.discovery.processor.DiscoveryV5MessageProcessor;
 import org.ethereum.beacon.discovery.processor.MessageProcessor;
 import org.ethereum.beacon.discovery.schema.NodeSession;
 import org.ethereum.beacon.discovery.storage.LocalNodeRecordStore;
 
-public class MessageHandler implements EnvelopeHandler {
+public class MessageHandler extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(MessageHandler.class);
   private final MessageProcessor messageProcessor;
 
@@ -36,7 +36,7 @@ public MessageHandler(
   }
 
   @Override
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireField(Field.MESSAGE, envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/MessagePacketHandler.java

@@ -9,8 +9,8 @@
 import org.ethereum.beacon.discovery.message.V5Message;
 import org.ethereum.beacon.discovery.packet.MessagePacket;
 import org.ethereum.beacon.discovery.packet.OrdinaryMessagePacket;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import org.ethereum.beacon.discovery.schema.NodeRecordFactory;
@@ -19,7 +19,7 @@
 import org.ethereum.beacon.discovery.util.DecryptException;
 
 /** Handles {@link MessagePacket} in {@link Field#PACKET_MESSAGE} field */
-public class MessagePacketHandler implements EnvelopeHandler {
+public class MessagePacketHandler extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(MessagePacketHandler.class);
   private final NodeRecordFactory nodeRecordFactory;
 
@@ -28,7 +28,7 @@ public MessagePacketHandler(NodeRecordFactory nodeRecordFactory) {
   }
 
   @Override
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireField(Field.PACKET_MESSAGE, envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/NewTaskHandler.java

@@ -6,20 +6,20 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import org.ethereum.beacon.discovery.pipeline.info.Request;
 import org.ethereum.beacon.discovery.schema.NodeSession;
 
 /** Enqueues task in session for any task found in {@link Field#REQUEST} */
-public class NewTaskHandler implements EnvelopeHandler {
+public class NewTaskHandler extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(NewTaskHandler.class);
 
   @Override
   @SuppressWarnings("rawtypes")
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireField(Field.REQUEST, envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/NextTaskHandler.java

@@ -11,8 +11,8 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.tuweni.bytes.Bytes;
 import org.ethereum.beacon.discovery.message.V5Message;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import org.ethereum.beacon.discovery.pipeline.Pipeline;
@@ -23,7 +23,7 @@
 import org.ethereum.beacon.discovery.task.TaskStatus;
 
 /** Gets next request task in session and processes it */
-public class NextTaskHandler implements EnvelopeHandler {
+public class NextTaskHandler extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(NextTaskHandler.class);
   private static final int DEFAULT_DELAY_MS = 1000;
   private static final int RANDOM_MESSAGE_SIZE = 128;
@@ -46,7 +46,7 @@ public static void tryToSendAwaitTaskIfAny(
   }
 
   @Override
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireSessionWithNodeRecord(envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/NodeSessionManager.java

@@ -19,8 +19,8 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.tuweni.bytes.Bytes;
 import org.ethereum.beacon.discovery.crypto.Signer;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import org.ethereum.beacon.discovery.pipeline.Pipeline;
@@ -37,7 +37,7 @@
  * Performs {@link Field#SESSION_LOOKUP} request. Looks up for Node session based on NodeId, which
  * should be in request field and stores it in {@link Field#SESSION} field.
  */
-public class NodeSessionManager implements EnvelopeHandler {
+public class NodeSessionManager extends AbstractSkippingEnvelopeHandler {
   private static final int SESSION_CLEANUP_DELAY_SECONDS = 180;
   private static final int REQUEST_CLEANUP_DELAY_SECONDS = 60;
   private static final Logger LOG = LogManager.getLogger(NodeSessionManager.class);
@@ -67,7 +67,7 @@ public NodeSessionManager(
   }
 
   @Override
-  public void handle(final Envelope envelope) {
+  protected void handlePacket(final Envelope envelope) {
     if (!HandlerUtil.requireField(Field.SESSION_LOOKUP, envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/NodeSessionRequestHandler.java

@@ -6,20 +6,20 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 
 /**
  * Searches for node in {@link Field#NODE} and requests session resolving using {@link
  * Field#SESSION_LOOKUP}
  */
-public class NodeSessionRequestHandler implements EnvelopeHandler {
+public class NodeSessionRequestHandler extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(NodeSessionRequestHandler.class);
 
   @Override
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireField(Field.NODE, envelope)) {
       return;
     }

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/OutgoingParcelHandler.java

@@ -8,8 +8,8 @@
 import org.apache.logging.log4j.Logger;
 import org.ethereum.beacon.discovery.AddressAccessPolicy;
 import org.ethereum.beacon.discovery.network.NetworkParcel;
+import org.ethereum.beacon.discovery.pipeline.AbstractSkippingEnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Envelope;
-import org.ethereum.beacon.discovery.pipeline.EnvelopeHandler;
 import org.ethereum.beacon.discovery.pipeline.Field;
 import org.ethereum.beacon.discovery.pipeline.HandlerUtil;
 import reactor.core.publisher.FluxSink;
@@ -19,7 +19,7 @@
  * we have outgoing parcel at the very first stage. Handler pushes it to `outgoingSink` stream which
  * is linked with discovery client.
  */
-public class OutgoingParcelHandler implements EnvelopeHandler {
+public class OutgoingParcelHandler extends AbstractSkippingEnvelopeHandler {
   private static final Logger LOG = LogManager.getLogger(OutgoingParcelHandler.class);
 
   private final FluxSink<NetworkParcel> outgoingSink;
@@ -32,7 +32,7 @@ public OutgoingParcelHandler(
   }
 
   @Override
-  public void handle(Envelope envelope) {
+  protected void handlePacket(Envelope envelope) {
     if (!HandlerUtil.requireField(Field.INCOMING, envelope)) {
       return;
     }