class PeerDiscoveryController {

  public void start() {
    if (!started.compareAndSet(false, true)) {
      throw new IllegalStateException("The peer table had already been started");
    }

    final List<DiscoveryPeer> initialDiscoveryPeers =
        bootstrapNodes.stream()
            .filter(peerPermissions::isAllowedInPeerTable)
            .collect(Collectors.toList());
    initialDiscoveryPeers.forEach(peerTable::tryAdd);

    recursivePeerRefreshState =
        new RecursivePeerRefreshState(
            this::bond,
            this::findNodes,
            timerUtil,
            localPeer,
            peerTable,
            peerPermissions,
            PEER_REFRESH_ROUND_TIMEOUT_IN_SECONDS,
            100);

    peerPermissions.subscribeUpdate(this::handlePermissionsUpdate);

    recursivePeerRefreshState.start(initialDiscoveryPeers, localPeer.getId());

    final long refreshTimerId =
        timerUtil.setPeriodic(
            Math.min(REFRESH_CHECK_INTERVAL_MILLIS, tableRefreshIntervalMs),
            this::refreshTableIfRequired);
    tableRefreshTimerId = OptionalLong.of(refreshTimerId);

    cleanTableTimerId =
        OptionalLong.of(
            timerUtil.setPeriodic(cleanPeerTableIntervalMs, this::cleanPeerTableIfRequired));
  }

  private  PeerDiscoveryController(
      final NodeKey nodeKey,
      final DiscoveryPeer localPeer,
      final PeerTable peerTable,
      final Collection<DiscoveryPeer> bootstrapNodes,
      final OutboundMessageHandler outboundMessageHandler,
      final TimerUtil timerUtil,
      final AsyncExecutor workerExecutor,
      final long tableRefreshIntervalMs,
      final long cleanPeerTableIntervalMs,
      final PeerRequirement peerRequirement,
      final PeerPermissions peerPermissions,
      final Subscribers<PeerBondedObserver> peerBondedObservers,
      final MetricsSystem metricsSystem);

}