gt:                 Channel peer = peers.get(i);
context: /*
   * This file is part of RskJ
   * Copyright (C) 2017 RSK Labs Ltd.
   * (derived from ethereumJ library, Copyright (c) 2016 <ether.camp>)
   *
   * This program is free software: you can redistribute it and/or modify
   * it under the terms of the GNU Lesser General Public License as published by
   * the Free Software Foundation, either version 3 of the License, or
   * (at your option) any later version.
   *
   * This program is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
   * GNU Lesser General Public License for more details.
   *
   * You should have received a copy of the GNU Lesser General Public License
   * along with this program. If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.ethereum.net.server;
  
  import co.rsk.config.RskSystemProperties;
  import co.rsk.net.Peer;
  import co.rsk.net.NodeID;
  import co.rsk.net.Status;
  import co.rsk.net.messages.*;
  import co.rsk.scoring.InetAddressBlock;
  import com.google.common.annotations.VisibleForTesting;
  import org.ethereum.config.NodeFilter;
  import org.ethereum.core.Block;
  import org.ethereum.core.BlockIdentifier;
  import org.ethereum.core.Transaction;
  import org.ethereum.net.message.ReasonCode;
  import org.ethereum.sync.SyncPool;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nonnull;
  import java.net.InetAddress;
  import java.time.Duration;
  import java.time.Instant;
  import java.util.*;
  import java.util.concurrent.*;
  import java.util.stream.Collectors;
  
  /**
   * @author Roman Mandeleil
   * @since 11.11.2014
   */
  public class ChannelManagerImpl implements ChannelManager {
  
      private static final Logger logger = LoggerFactory.getLogger("net");
  
      // If the inbound peer connection was dropped by us with a reason message
      // then we ban that peer IP on any connections for some time to protect from
      // too active peers
      private static final Duration INBOUND_CONNECTION_BAN_TIMEOUT = Duration.ofSeconds(10);
      private final Object activePeersLock = new Object();
      private final Map<NodeID, Channel> activePeers;
  
      // Using a concurrent list
      // (the add and remove methods copy an internal array,
      // but the iterator directly use the internal array)
      private final List<Channel> newPeers;
  
      private final ScheduledExecutorService mainWorker;
      private final Map<InetAddress, Instant> disconnectionsTimeouts;
      private final Object disconnectionTimeoutsLock = new Object();
  
      private final SyncPool syncPool;
      private final NodeFilter trustedPeers;
      private final int maxActivePeers;
      private final int maxConnectionsAllowed;
      private final int networkCIDR;
  
      public ChannelManagerImpl(RskSystemProperties config, SyncPool syncPool) {
          this.mainWorker = Executors.newSingleThreadScheduledExecutor(target -> new Thread(target, "newPeersProcessor"));
          this.syncPool = syncPool;
          this.maxActivePeers = config.maxActivePeers();
          this.trustedPeers = config.trustedPeers();
          this.disconnectionsTimeouts = new HashMap<>();
          this.activePeers = new ConcurrentHashMap<>();
          this.newPeers = new CopyOnWriteArrayList<>();
          this.maxConnectionsAllowed = config.maxConnectionsAllowed();
          this.networkCIDR = config.networkCIDR();
      }
  
      @Override
      public void start() {
          mainWorker.scheduleWithFixedDelay(this::handleNewPeersAndDisconnections, 0, 1, TimeUnit.SECONDS);
      }
  
      @Override
      public void stop() {
          mainWorker.shutdown();
      }
  
      private void handleNewPeersAndDisconnections() {
          this.tryProcessNewPeers();
          this.cleanDisconnections();
      }
  
      @VisibleForTesting
      public void tryProcessNewPeers() {
          if (newPeers.isEmpty()) {
              return;
          }
  
          try {
              processNewPeers();
          } catch (Exception e) {
              logger.error("Error", e);
          }
      }
  
      private void cleanDisconnections() {
          synchronized (disconnectionTimeoutsLock) {
              Instant now = Instant.now();
              disconnectionsTimeouts.values().removeIf(v -> !isRecent(v, now));
          }
      }
  
      private void processNewPeers() {
          synchronized (newPeers) {
              List<Channel> processedChannels = new ArrayList<>();
              newPeers.stream().filter(Channel::isProtocolsInitialized).forEach(c -> processNewPeer(c, processedChannels));
              newPeers.removeAll(processedChannels);
          }
      }
  
      private void processNewPeer(Channel channel, List<Channel> processedChannels) {
          ReasonCode reason = getNewPeerDisconnectionReason(channel);
          if (reason != null) {
              disconnect(channel, reason);
          } else {
              addToActives(channel);
          }
          processedChannels.add(channel);
      }
  
      private ReasonCode getNewPeerDisconnectionReason(Channel channel) {
          if (activePeers.containsKey(channel.getNodeId())) {
              return ReasonCode.DUPLICATE_PEER;
          }
  
          if (!channel.isActive() && activePeers.size() >= maxActivePeers && !trustedPeers.accept(channel.getNode())) {
              return ReasonCode.TOO_MANY_PEERS;
          }
  
          return null;
      }
  
      private void disconnect(Channel peer, ReasonCode reason) {
          logger.debug("Disconnecting peer with reason {} : {}", reason, peer);
          peer.disconnect(reason);
          synchronized (disconnectionTimeoutsLock) {
              disconnectionsTimeouts.put(peer.getInetSocketAddress().getAddress(),
                      Instant.now().plus(INBOUND_CONNECTION_BAN_TIMEOUT));
          }
      }
  
      public boolean isRecentlyDisconnected(InetAddress peerAddress) {
          synchronized (disconnectionTimeoutsLock) {
              return isRecent(disconnectionsTimeouts.getOrDefault(peerAddress, Instant.EPOCH), Instant.now());
          }
      }
  
      private boolean isRecent(Instant disconnectionTimeout, Instant currentTime) {
          return currentTime.isBefore(disconnectionTimeout);
      }
  
      private void addToActives(Channel peer) {
          if (peer.isUsingNewProtocol() || peer.hasEthStatusSucceeded()) {
              syncPool.add(peer);
              synchronized (activePeersLock) {
                  activePeers.put(peer.getNodeId(), peer);
              }
          }
      }
  
      /**
       * broadcastBlock Propagates a block message across active peers
       *
       * @param block new Block to be sent
       * @return a set containing the ids of the peers that received the block.
       */
      @Nonnull
      public Set<NodeID> broadcastBlock(@Nonnull final Block block) {
  
          final Set<NodeID> nodesIdsBroadcastedTo = new HashSet<>();
          final BlockIdentifier bi = new BlockIdentifier(block.getHash().getBytes(), block.getNumber());
          final Message newBlock = new BlockMessage(block);
          final Message newBlockHashes = new NewBlockHashesMessage(Arrays.asList(bi));
          synchronized (activePeersLock) {
              // Get a randomized list with all the peers that don't have the block yet.
              activePeers.values().forEach(c -> logger.trace("RSK activePeers: {}", c));
              List<Channel> peers = new ArrayList<>(activePeers.values());
              Collections.shuffle(peers);
  
              int sqrt = (int) Math.floor(Math.sqrt(peers.size()));
              for (int i = 0; i < sqrt; i++) {
  