fix using optionals + test (#3711)

Signed-off-by: Frank Li <b439988l@gmail.com>

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/network/exceptions/PeerChannelClosedException.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright contributors to Hyperledger Besu
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.hyperledger.besu.ethereum.p2p.network.exceptions;
+
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.PeerInfo;
+
+public class PeerChannelClosedException extends RuntimeException {
+  public PeerChannelClosedException(final PeerInfo peerInfo) {
+    super(String.format("Failed to create connection for peer %s", peerInfo));
+  }
+}

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/rlpx/connections/netty/DeFramer.java

@@ -16,6 +16,7 @@ package org.hyperledger.besu.ethereum.p2p.rlpx.connections.netty;
 
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.BreachOfProtocolException;
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.IncompatiblePeerException;
+import org.hyperledger.besu.ethereum.p2p.network.exceptions.PeerChannelClosedException;
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.PeerDisconnectedException;
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.UnexpectedPeerConnectionException;
 import org.hyperledger.besu.ethereum.p2p.peers.DefaultPeer;
@@ -125,11 +126,17 @@ final class DeFramer extends ByteToMessageDecoder {
                 subProtocols,
                 localNode.getPeerInfo().getCapabilities(),
                 peerInfo.getCapabilities());
-        final Peer peer = expectedPeer.orElse(createPeer(peerInfo, ctx));
+        final Optional<Peer> peer = expectedPeer.or(() -> createPeer(peerInfo, ctx));
+        if (peer.isEmpty()) {
+          LOG.debug("Failed to create connection for peer {}", peerInfo);
+          connectFuture.completeExceptionally(new PeerChannelClosedException(peerInfo));
+          ctx.close();
+          return;
+        }
         final PeerConnection connection =
             new NettyPeerConnection(
                 ctx,
-                peer,
+                peer.get(),
                 peerInfo,
                 capabilityMultiplexer,
                 connectionEventDispatcher,
@@ -193,17 +200,21 @@ final class DeFramer extends ByteToMessageDecoder {
     }
   }
 
-  private Peer createPeer(final PeerInfo peerInfo, final ChannelHandlerContext ctx) {
+  private Optional<Peer> createPeer(final PeerInfo peerInfo, final ChannelHandlerContext ctx) {
     final InetSocketAddress remoteAddress = ((InetSocketAddress) ctx.channel().remoteAddress());
+    if (remoteAddress == null) {
+      return Optional.empty();
+    }
     int port = peerInfo.getPort();
-    return DefaultPeer.fromEnodeURL(
+    return Optional.of(
-        EnodeURLImpl.builder()
+        DefaultPeer.fromEnodeURL(
-            .nodeId(peerInfo.getNodeId())
+            EnodeURLImpl.builder()
-            .ipAddress(remoteAddress.getAddress())
+                .nodeId(peerInfo.getNodeId())
-            .listeningPort(port)
+                .ipAddress(remoteAddress.getAddress())
-            // Discovery information is unknown, so disable it
+                .listeningPort(port)
-            .disableDiscovery()
+                // Discovery information is unknown, so disable it
-            .build());
+                .disableDiscovery()
+                .build()));
   }
 
   @Override

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/rlpx/connections/netty/DeFramerTest.java

@@ -26,6 +26,7 @@ import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.BreachOfProtocolException;
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.IncompatiblePeerException;
+import org.hyperledger.besu.ethereum.p2p.network.exceptions.PeerChannelClosedException;
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.PeerDisconnectedException;
 import org.hyperledger.besu.ethereum.p2p.network.exceptions.UnexpectedPeerConnectionException;
 import org.hyperledger.besu.ethereum.p2p.peers.DefaultPeer;
@@ -341,6 +342,29 @@ public class DeFramerTest {
     assertThat(out).isEmpty();
   }
 
+  @Test
+  public void decode_shouldHandleRemoteSocketAddressIsNull() {
+    final Peer peer = createRemotePeer();
+    final PeerInfo remotePeerInfo =
+        new PeerInfo(p2pVersion, clientId, capabilities, 0, peer.getId());
+    HelloMessage helloMessage = HelloMessage.create(remotePeerInfo);
+    ByteBuf data = Unpooled.wrappedBuffer(helloMessage.getData().toArray());
+    when(framer.deframe(any()))
+        .thenReturn(new RawMessage(helloMessage.getCode(), helloMessage.getData()))
+        .thenReturn(null);
+    when(ctx.channel().remoteAddress()).thenReturn(null);
+    ChannelFuture future = NettyMocks.channelFuture(true);
+    when(ctx.writeAndFlush(any())).thenReturn(future);
+    List<Object> out = new ArrayList<>();
+    deFramer.decode(ctx, data, out);
+
+    assertThat(connectFuture).isDone();
+    assertThat(connectFuture).isCompletedExceptionally();
+    assertThatThrownBy(connectFuture::get).hasCauseInstanceOf(PeerChannelClosedException.class);
+    verify(ctx).close();
+    assertThat(out).isEmpty();
+  }
+
   @Test
   public void decode_shouldHandleInvalidMessage() {
     MessageData messageData = PingMessage.get();