Discovery v5

discovery/build.gradle

@@ -0,0 +1,12 @@
+dependencies {
+  implementation project(':types')
+  implementation project(':util')
+  implementation project(':db:core')
+  implementation project(':chain')
+
+  implementation 'com.google.guava:guava'
+  implementation 'io.projectreactor:reactor-core'
+  implementation 'io.netty:netty-all'
+  implementation 'org.apache.logging.log4j:log4j-core'
+  implementation 'org.web3j:core'
+}

discovery/src/main/java/org/ethereum/beacon/discovery/DiscoveryManager.java

@@ -0,0 +1,35 @@
+package org.ethereum.beacon.discovery;
+
+import org.ethereum.beacon.discovery.enr.NodeRecord;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Discovery Manager, top interface for peer discovery mechanism as described at <a
+ * href="https://github.com/ethereum/devp2p/blob/master/discv5/discv5.md">https://github.com/ethereum/devp2p/blob/master/discv5/discv5.md</a>
+ */
+public interface DiscoveryManager {
+
+  void start();
+
+  void stop();
+
+  /**
+   * Initiates FINDNODE with node `nodeRecord`
+   *
+   * @param nodeRecord Ethereum Node record
+   * @param distance Distance to search for
+   * @return Future which is fired when reply is received or fails in timeout/not successful
+   *     handshake/bad message exchange.
+   */
+  CompletableFuture<Void> findNodes(NodeRecord nodeRecord, int distance);
+
+  /**
+   * Initiates PING with node `nodeRecord`
+   *
+   * @param nodeRecord Ethereum Node record
+   * @return Future which is fired when reply is received or fails in timeout/not successful
+   *     handshake/bad message exchange.
+   */
+  CompletableFuture<Void> ping(NodeRecord nodeRecord);
+}

discovery/src/main/java/org/ethereum/beacon/discovery/DiscoveryManagerImpl.java

@@ -0,0 +1,156 @@
+package org.ethereum.beacon.discovery;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.enr.EnrField;
+import org.ethereum.beacon.discovery.enr.NodeRecord;
+import org.ethereum.beacon.discovery.enr.NodeRecordFactory;
+import org.ethereum.beacon.discovery.network.DiscoveryClient;
+import org.ethereum.beacon.discovery.network.NettyDiscoveryClientImpl;
+import org.ethereum.beacon.discovery.network.NettyDiscoveryServer;
+import org.ethereum.beacon.discovery.network.NettyDiscoveryServerImpl;
+import org.ethereum.beacon.discovery.network.NetworkParcel;
+import org.ethereum.beacon.discovery.pipeline.Envelope;
+import org.ethereum.beacon.discovery.pipeline.Field;
+import org.ethereum.beacon.discovery.pipeline.Pipeline;
+import org.ethereum.beacon.discovery.pipeline.PipelineImpl;
+import org.ethereum.beacon.discovery.pipeline.handler.AuthHeaderMessagePacketHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.BadPacketHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.IncomingDataPacker;
+import org.ethereum.beacon.discovery.pipeline.handler.MessageHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.MessagePacketHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.NewTaskHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.NextTaskHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.NodeIdToSession;
+import org.ethereum.beacon.discovery.pipeline.handler.NodeSessionRequestHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.NotExpectedIncomingPacketHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.OutgoingParcelHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.UnknownPacketTagToSender;
+import org.ethereum.beacon.discovery.pipeline.handler.UnknownPacketTypeByStatus;
+import org.ethereum.beacon.discovery.pipeline.handler.WhoAreYouAttempt;
+import org.ethereum.beacon.discovery.pipeline.handler.WhoAreYouPacketHandler;
+import org.ethereum.beacon.discovery.pipeline.handler.WhoAreYouSessionResolver;
+import org.ethereum.beacon.discovery.storage.AuthTagRepository;
+import org.ethereum.beacon.discovery.storage.NodeBucketStorage;
+import org.ethereum.beacon.discovery.storage.NodeTable;
+import org.ethereum.beacon.discovery.task.TaskOptions;
+import org.ethereum.beacon.discovery.task.TaskType;
+import org.ethereum.beacon.schedulers.Scheduler;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+import reactor.core.publisher.ReplayProcessor;
+import tech.pegasys.artemis.util.bytes.Bytes4;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class DiscoveryManagerImpl implements DiscoveryManager {
+  private static final Logger logger = LogManager.getLogger(DiscoveryManagerImpl.class);
+  private final ReplayProcessor<NetworkParcel> outgoingMessages = ReplayProcessor.cacheLast();
+  private final FluxSink<NetworkParcel> outgoingSink = outgoingMessages.sink();
+  private final NettyDiscoveryServer discoveryServer;
+  private final Scheduler scheduler;
+  private final Pipeline incomingPipeline = new PipelineImpl();
+  private final Pipeline outgoingPipeline = new PipelineImpl();
+  private final NodeRecordFactory nodeRecordFactory;
+  private DiscoveryClient discoveryClient;
+  private CountDownLatch clientStarted = new CountDownLatch(1);
+
+  public DiscoveryManagerImpl(
+      NodeTable nodeTable,
+      NodeBucketStorage nodeBucketStorage,
+      NodeRecord homeNode,
+      BytesValue homeNodePrivateKey,
+      NodeRecordFactory nodeRecordFactory,
+      Scheduler serverScheduler,
+      Scheduler taskScheduler) {
+    AuthTagRepository authTagRepo = new AuthTagRepository();
+    this.scheduler = serverScheduler;
+    this.nodeRecordFactory = nodeRecordFactory;
+    this.discoveryServer =
+        new NettyDiscoveryServerImpl(
+            ((Bytes4) homeNode.get(EnrField.IP_V4)), (int) homeNode.get(EnrField.UDP_V4));
+    discoveryServer.useDatagramChannel(
+        channel -> {
+          discoveryClient = new NettyDiscoveryClientImpl(outgoingMessages, channel);
+          clientStarted.countDown();
+        });
+    NodeIdToSession nodeIdToSession =
+        new NodeIdToSession(
+            homeNode,
+            homeNodePrivateKey,
+            nodeBucketStorage,
+            authTagRepo,
+            nodeTable,
+            outgoingPipeline);
+    incomingPipeline
+        .addHandler(new IncomingDataPacker())
+        .addHandler(new WhoAreYouAttempt(homeNode.getNodeId()))
+        .addHandler(new WhoAreYouSessionResolver(authTagRepo))
+        .addHandler(new UnknownPacketTagToSender(homeNode))
+        .addHandler(nodeIdToSession)
+        .addHandler(new UnknownPacketTypeByStatus())
+        .addHandler(new NotExpectedIncomingPacketHandler())
+        .addHandler(new WhoAreYouPacketHandler(outgoingPipeline, taskScheduler))
+        .addHandler(
+            new AuthHeaderMessagePacketHandler(outgoingPipeline, taskScheduler, nodeRecordFactory))
+        .addHandler(new MessagePacketHandler())
+        .addHandler(new MessageHandler(nodeRecordFactory))
+        .addHandler(new BadPacketHandler());
+    outgoingPipeline
+        .addHandler(new OutgoingParcelHandler(outgoingSink))
+        .addHandler(new NodeSessionRequestHandler())
+        .addHandler(nodeIdToSession)
+        .addHandler(new NewTaskHandler())
+        .addHandler(new NextTaskHandler(outgoingPipeline, taskScheduler));
+  }
+
+  @Override
+  public void start() {
+    incomingPipeline.build();
+    outgoingPipeline.build();
+    Flux.from(discoveryServer.getIncomingPackets()).subscribe(incomingPipeline::push);
+    discoveryServer.start(scheduler);
+    try {
+      clientStarted.await(2, TimeUnit.SECONDS);
+    } catch (InterruptedException e) {
+      throw new RuntimeException("Failed to start client", e);
+    }
+  }
+
+  @Override
+  public void stop() {
+    discoveryServer.stop();
+  }
+
+  private CompletableFuture<Void> executeTaskImpl(
+      NodeRecord nodeRecord, TaskType taskType, TaskOptions taskOptions) {
+    Envelope envelope = new Envelope();
+    envelope.put(Field.NODE, nodeRecord);
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    envelope.put(Field.TASK, taskType);
+    envelope.put(Field.FUTURE, future);
+    envelope.put(Field.TASK_OPTIONS, taskOptions);
+    outgoingPipeline.push(envelope);
+    return future;
+  }
+
+  @Override
+  public CompletableFuture<Void> findNodes(NodeRecord nodeRecord, int distance) {
+    return executeTaskImpl(nodeRecord, TaskType.FINDNODE, new TaskOptions(true, distance));
+  }
+
+  @Override
+  public CompletableFuture<Void> ping(NodeRecord nodeRecord) {
+    return executeTaskImpl(nodeRecord, TaskType.PING, new TaskOptions(true));
+  }
+
+  @VisibleForTesting
+  Publisher<NetworkParcel> getOutgoingMessages() {
+    return outgoingMessages;
+  }
+}

discovery/src/main/java/org/ethereum/beacon/discovery/DiscoveryMessageProcessor.java

@@ -0,0 +1,10 @@
+package org.ethereum.beacon.discovery;
+
+import org.ethereum.beacon.discovery.message.DiscoveryMessage;
+
+/** Handles discovery messages of several types */
+public interface DiscoveryMessageProcessor<M extends DiscoveryMessage> {
+  Protocol getSupportedIdentity();
+
+  void handleMessage(M message, NodeSession session);
+}

discovery/src/main/java/org/ethereum/beacon/discovery/DiscoveryV5MessageProcessor.java

@@ -0,0 +1,49 @@
+package org.ethereum.beacon.discovery;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.enr.NodeRecordFactory;
+import org.ethereum.beacon.discovery.message.DiscoveryV5Message;
+import org.ethereum.beacon.discovery.message.MessageCode;
+import org.ethereum.beacon.discovery.message.handler.FindNodeHandler;
+import org.ethereum.beacon.discovery.message.handler.MessageHandler;
+import org.ethereum.beacon.discovery.message.handler.NodesHandler;
+import org.ethereum.beacon.discovery.message.handler.PingHandler;
+import org.ethereum.beacon.discovery.message.handler.PongHandler;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * {@link DiscoveryV5Message} v5 messages processor. Uses several handlers, one fo each type of v5
+ * message to handle appropriate message.
+ */
+public class DiscoveryV5MessageProcessor implements DiscoveryMessageProcessor<DiscoveryV5Message> {
+  private static final Logger logger = LogManager.getLogger(DiscoveryV5MessageProcessor.class);
+  private final Map<MessageCode, MessageHandler> messageHandlers = new HashMap<>();
+  private final NodeRecordFactory nodeRecordFactory;
+
+  public DiscoveryV5MessageProcessor(NodeRecordFactory nodeRecordFactory) {
+    messageHandlers.put(MessageCode.PING, new PingHandler());
+    messageHandlers.put(MessageCode.PONG, new PongHandler());
+    messageHandlers.put(MessageCode.FINDNODE, new FindNodeHandler());
+    messageHandlers.put(MessageCode.NODES, new NodesHandler());
+    this.nodeRecordFactory = nodeRecordFactory;
+  }
+
+  @Override
+  public Protocol getSupportedIdentity() {
+    return Protocol.V5;
+  }
+
+  @Override
+  public void handleMessage(DiscoveryV5Message message, NodeSession session) {
+    MessageCode code = message.getCode();
+    MessageHandler messageHandler = messageHandlers.get(code);
+    logger.trace(() -> String.format("Handling message %s in session %s", message, session));
+    if (messageHandler == null) {
+      throw new RuntimeException("Not implemented yet");
+    }
+    messageHandler.handle(message.create(nodeRecordFactory), session);
+  }
+}