@Test void onConnect_shouldNotDisconnectIncomingPeerWhenStatusMessageSent() { final Peer peer = createPeer(1); final Eth2Peer eth2Peer = getEth2Peer(peer); when(eth2Peer.hasStatus()).thenReturn(true); when(peer.connectionInitiatedLocally()).thenReturn(false); peerManager.onConnect(peer); assertThat(asyncRunner.hasDelayedActions()).isTrue(); setInitialPeerStatus(peer); asyncRunner.executeQueuedActions(); verify(eth2Peer, never()).disconnectCleanly(DisconnectReason.REMOTE_FAULT); }
@Override public void onConnect(final Peer peer) { Eth2Peer eth2Peer = eth2PeerFactory.create(peer, rpcMethods); final boolean wasAdded = connectedPeerMap.putIfAbsent(peer.getId(), eth2Peer) == null; if (!wasAdded) { LOG.debug("Duplicate peer connection detected for peer {}. Ignoring peer.", peer.getId()); return; } peer.setDisconnectRequestHandler(reason -> eth2Peer.sendGoodbye(reason.getReasonCode())); if (peer.connectionInitiatedLocally()) { eth2Peer .sendStatus() .finish( () -> LOG.trace("Sent status to {}", peer.getId()), RootCauseExceptionHandler.builder() .addCatch( RpcException.class, err -> { LOG.trace("Status message rejected by {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately( Optional.of(DisconnectReason.REMOTE_FAULT), true); }) .defaultCatch( err -> { LOG.debug("Failed to send status to {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately(Optional.empty(), true); })); } else { ensureStatusReceived(eth2Peer); } eth2Peer.subscribeInitialStatus( (status) -> { connectSubscribers.forEach(c -> c.onConnected(eth2Peer)); setUpPeriodicTasksForPeer(eth2Peer); }); }
Eth2PeerManager implements PeerLookup, PeerHandler { @Override public void onConnect(final Peer peer) { Eth2Peer eth2Peer = eth2PeerFactory.create(peer, rpcMethods); final boolean wasAdded = connectedPeerMap.putIfAbsent(peer.getId(), eth2Peer) == null; if (!wasAdded) { LOG.debug("Duplicate peer connection detected for peer {}. Ignoring peer.", peer.getId()); return; } peer.setDisconnectRequestHandler(reason -> eth2Peer.sendGoodbye(reason.getReasonCode())); if (peer.connectionInitiatedLocally()) { eth2Peer .sendStatus() .finish( () -> LOG.trace("Sent status to {}", peer.getId()), RootCauseExceptionHandler.builder() .addCatch( RpcException.class, err -> { LOG.trace("Status message rejected by {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately( Optional.of(DisconnectReason.REMOTE_FAULT), true); }) .defaultCatch( err -> { LOG.debug("Failed to send status to {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately(Optional.empty(), true); })); } else { ensureStatusReceived(eth2Peer); } eth2Peer.subscribeInitialStatus( (status) -> { connectSubscribers.forEach(c -> c.onConnected(eth2Peer)); setUpPeriodicTasksForPeer(eth2Peer); }); } }
Eth2PeerManager implements PeerLookup, PeerHandler { @Override public void onConnect(final Peer peer) { Eth2Peer eth2Peer = eth2PeerFactory.create(peer, rpcMethods); final boolean wasAdded = connectedPeerMap.putIfAbsent(peer.getId(), eth2Peer) == null; if (!wasAdded) { LOG.debug("Duplicate peer connection detected for peer {}. Ignoring peer.", peer.getId()); return; } peer.setDisconnectRequestHandler(reason -> eth2Peer.sendGoodbye(reason.getReasonCode())); if (peer.connectionInitiatedLocally()) { eth2Peer .sendStatus() .finish( () -> LOG.trace("Sent status to {}", peer.getId()), RootCauseExceptionHandler.builder() .addCatch( RpcException.class, err -> { LOG.trace("Status message rejected by {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately( Optional.of(DisconnectReason.REMOTE_FAULT), true); }) .defaultCatch( err -> { LOG.debug("Failed to send status to {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately(Optional.empty(), true); })); } else { ensureStatusReceived(eth2Peer); } eth2Peer.subscribeInitialStatus( (status) -> { connectSubscribers.forEach(c -> c.onConnected(eth2Peer)); setUpPeriodicTasksForPeer(eth2Peer); }); } Eth2PeerManager(
      final AsyncRunner asyncRunner,
      final CombinedChainDataClient combinedChainDataClient,
      final RecentChainData storageClient,
      final MetricsSystem metricsSystem,
      final Eth2PeerFactory eth2PeerFactory,
      final StatusMessageFactory statusMessageFactory,
      final MetadataMessagesFactory metadataMessagesFactory,
      final RpcEncoding rpcEncoding,
      final Duration eth2RpcPingInterval,
      final int eth2RpcOutstandingPingThreshold,
      final Duration eth2StatusUpdateInterval); }
Eth2PeerManager implements PeerLookup, PeerHandler { @Override public void onConnect(final Peer peer) { Eth2Peer eth2Peer = eth2PeerFactory.create(peer, rpcMethods); final boolean wasAdded = connectedPeerMap.putIfAbsent(peer.getId(), eth2Peer) == null; if (!wasAdded) { LOG.debug("Duplicate peer connection detected for peer {}. Ignoring peer.", peer.getId()); return; } peer.setDisconnectRequestHandler(reason -> eth2Peer.sendGoodbye(reason.getReasonCode())); if (peer.connectionInitiatedLocally()) { eth2Peer .sendStatus() .finish( () -> LOG.trace("Sent status to {}", peer.getId()), RootCauseExceptionHandler.builder() .addCatch( RpcException.class, err -> { LOG.trace("Status message rejected by {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately( Optional.of(DisconnectReason.REMOTE_FAULT), true); }) .defaultCatch( err -> { LOG.debug("Failed to send status to {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately(Optional.empty(), true); })); } else { ensureStatusReceived(eth2Peer); } eth2Peer.subscribeInitialStatus( (status) -> { connectSubscribers.forEach(c -> c.onConnected(eth2Peer)); setUpPeriodicTasksForPeer(eth2Peer); }); } Eth2PeerManager(
      final AsyncRunner asyncRunner,
      final CombinedChainDataClient combinedChainDataClient,
      final RecentChainData storageClient,
      final MetricsSystem metricsSystem,
      final Eth2PeerFactory eth2PeerFactory,
      final StatusMessageFactory statusMessageFactory,
      final MetadataMessagesFactory metadataMessagesFactory,
      final RpcEncoding rpcEncoding,
      final Duration eth2RpcPingInterval,
      final int eth2RpcOutstandingPingThreshold,
      final Duration eth2StatusUpdateInterval); static Eth2PeerManager create(
      final AsyncRunner asyncRunner,
      final RecentChainData recentChainData,
      final StorageQueryChannel historicalChainData,
      final MetricsSystem metricsSystem,
      final AttestationSubnetService attestationSubnetService,
      final RpcEncoding rpcEncoding,
      final Optional<Checkpoint> requiredCheckpoint,
      final Duration eth2RpcPingInterval,
      final int eth2RpcOutstandingPingThreshold,
      final Duration eth2StatusUpdateInterval,
      final TimeProvider timeProvider,
      final int peerRateLimit,
      final int peerRequestLimit); MetadataMessage getMetadataMessage(); @Override void onConnect(final Peer peer); long subscribeConnect(final PeerConnectedSubscriber<Eth2Peer> subscriber); void unsubscribeConnect(final long subscriptionId); @Override void onDisconnect(@NotNull final Peer peer); BeaconChainMethods getBeaconChainMethods(); @Override Optional<Eth2Peer> getConnectedPeer(NodeId nodeId); Optional<Eth2Peer> getPeer(NodeId peerId); Stream<Eth2Peer> streamPeers(); SafeFuture<?> sendGoodbyeToPeers(); }
Eth2PeerManager implements PeerLookup, PeerHandler { @Override public void onConnect(final Peer peer) { Eth2Peer eth2Peer = eth2PeerFactory.create(peer, rpcMethods); final boolean wasAdded = connectedPeerMap.putIfAbsent(peer.getId(), eth2Peer) == null; if (!wasAdded) { LOG.debug("Duplicate peer connection detected for peer {}. Ignoring peer.", peer.getId()); return; } peer.setDisconnectRequestHandler(reason -> eth2Peer.sendGoodbye(reason.getReasonCode())); if (peer.connectionInitiatedLocally()) { eth2Peer .sendStatus() .finish( () -> LOG.trace("Sent status to {}", peer.getId()), RootCauseExceptionHandler.builder() .addCatch( RpcException.class, err -> { LOG.trace("Status message rejected by {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately( Optional.of(DisconnectReason.REMOTE_FAULT), true); }) .defaultCatch( err -> { LOG.debug("Failed to send status to {}: {}", peer.getId(), err); eth2Peer.disconnectImmediately(Optional.empty(), true); })); } else { ensureStatusReceived(eth2Peer); } eth2Peer.subscribeInitialStatus( (status) -> { connectSubscribers.forEach(c -> c.onConnected(eth2Peer)); setUpPeriodicTasksForPeer(eth2Peer); }); } Eth2PeerManager(
      final AsyncRunner asyncRunner,
      final CombinedChainDataClient combinedChainDataClient,
      final RecentChainData storageClient,
      final MetricsSystem metricsSystem,
      final Eth2PeerFactory eth2PeerFactory,
      final StatusMessageFactory statusMessageFactory,
      final MetadataMessagesFactory metadataMessagesFactory,
      final RpcEncoding rpcEncoding,
      final Duration eth2RpcPingInterval,
      final int eth2RpcOutstandingPingThreshold,
      final Duration eth2StatusUpdateInterval); static Eth2PeerManager create(
      final AsyncRunner asyncRunner,
      final RecentChainData recentChainData,
      final StorageQueryChannel historicalChainData,
      final MetricsSystem metricsSystem,
      final AttestationSubnetService attestationSubnetService,
      final RpcEncoding rpcEncoding,
      final Optional<Checkpoint> requiredCheckpoint,
      final Duration eth2RpcPingInterval,
      final int eth2RpcOutstandingPingThreshold,
      final Duration eth2StatusUpdateInterval,
      final TimeProvider timeProvider,
      final int peerRateLimit,
      final int peerRequestLimit); MetadataMessage getMetadataMessage(); @Override void onConnect(final Peer peer); long subscribeConnect(final PeerConnectedSubscriber<Eth2Peer> subscriber); void unsubscribeConnect(final long subscriptionId); @Override void onDisconnect(@NotNull final Peer peer); BeaconChainMethods getBeaconChainMethods(); @Override Optional<Eth2Peer> getConnectedPeer(NodeId nodeId); Optional<Eth2Peer> getPeer(NodeId peerId); Stream<Eth2Peer> streamPeers(); SafeFuture<?> sendGoodbyeToPeers();  }