public void onMessage(final Packet packet, final DiscoveryPeer sender) {
    discoveryProtocolLogger.logReceivedPacket(sender, packet);

    // Message from self. This should not happen.
    if (sender.getId().equals(localPeer.getId())) {
      return;
    }

    final DiscoveryPeer peer = resolvePeer(sender);
    if (peer == null) {
      return;
    }
    final Bytes peerId = peer.getId();
    switch (packet.getType()) {
      case PING:
        if (peerPermissions.allowInboundBonding(peer)) {
          final PingPacketData ping = packet.getPacketData(PingPacketData.class).get();
          if (!PeerDiscoveryStatus.BONDED.equals(peer.getStatus())
              && (bondingPeers.getIfPresent(sender.getId()) == null)) {
            bond(peer);
          }
          respondToPing(ping, packet.getHash(), peer);
        }
        break;
      case PONG:
        matchInteraction(packet)
            .ifPresent(
                interaction -> {
                  if (filterOnEnrForkId) {
                    requestENR(peer);
                  }
                  bondingPeers.invalidate(peerId);
                  checkBeforeAddingToPeerTable(peer);
                  recursivePeerRefreshState.onBondingComplete(peer);
                  Optional.ofNullable(cachedEnrRequests.getIfPresent(peerId))
                      .ifPresent(cachedEnrRequest -> processEnrRequest(peer, cachedEnrRequest));
                });
        break;
      case NEIGHBORS:
        matchInteraction(packet)
            .ifPresent(
                interaction ->
                    recursivePeerRefreshState.onNeighboursReceived(
                        peer, getPeersFromNeighborsPacket(packet)));
        break;
      case FIND_NEIGHBORS:
        if (PeerDiscoveryStatus.BONDED.equals(peer.getStatus())
            && peerPermissions.allowInboundNeighborsRequest(peer)) {
          final FindNeighborsPacketData fn =
              packet.getPacketData(FindNeighborsPacketData.class).get();
          respondToFindNeighbors(fn, peer);
        }

        break;
      case ENR_REQUEST:
        if (PeerDiscoveryStatus.BONDED.equals(peer.getStatus())) {
          processEnrRequest(peer, packet);
        } else if (PeerDiscoveryStatus.BONDING.equals(peer.getStatus())) {
          LOG.trace("ENR_REQUEST cached for bonding peer Id: {}", peerId);
          // Due to UDP, it may happen that we receive the ENR_REQUEST just before the PONG.
          // Because peers want to send the ENR_REQUEST directly after the pong.
          // If this happens we don't want to ignore the request but process when bonded.
          // this cache allows to keep the request and to respond after having processed the PONG
          cachedEnrRequests.put(peerId, packet);
        }
        break;
      case ENR_RESPONSE:
        matchInteraction(packet)
            .ifPresent(
                interaction -> {
                  final Optional<EnrResponsePacketData> packetData =
                      packet.getPacketData(EnrResponsePacketData.class);
                  final NodeRecord enr = packetData.get().getEnr();
                  peer.setNodeRecord(enr);
                });
        break;
    }
  }
Handles an incoming message and processes it based on the state machine for the {@link
DiscoveryPeer}.

<p>The callback will be called with the canonical representation of the sender Peer as stored
in our table, or with an empty Optional if the message was out of band and we didn't process
it.

@param packet The incoming message.
@param sender The sender.
ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/discovery/internal/PeerDiscoveryController.java