final val ProviderClass: String =

      setup.get[BootstrapSetup]

        .flatMap(_.actorRefProvider).map(_.identifier)

        .getOrElse(getString("akka.actor.provider")) match {

          case "local"   ⇒ classOf[LocalActorRefProvider].getName

          // these two cannot be referenced by class as they may not be on the classpath

          case "remote"  ⇒ "akka.remote.RemoteActorRefProvider"

          case "cluster" ⇒ "akka.cluster.ClusterActorRefProvider"

          case fqcn      ⇒ fqcn

        }

def init(system: ActorSystemImpl): Unit = {

    local.init(system)

    actorRefResolveThreadLocalCache = ActorRefResolveThreadLocalCache(system)

    remotingTerminator = system.systemActorOf(

      remoteSettings.configureDispatcher(Props(classOf[RemotingTerminator], local.systemGuardian)),

      "remoting-terminator")

    val internals = Internals(

      remoteDaemon = {

        val d = new RemoteSystemDaemon(

          system,

          local.rootPath / "remote",

          rootGuardian,

          remotingTerminator,

          _log,

          untrustedMode = remoteSettings.UntrustedMode)

        local.registerExtraNames(Map(("remote", d)))

        d

      },

      transport =

        if (remoteSettings.Artery.Enabled) remoteSettings.Artery.Transport match {

          case ArterySettings.AeronUpd ⇒ new ArteryAeronUdpTransport(system, this)

          case ArterySettings.Tcp      ⇒ new ArteryTcpTransport(system, this, tlsEnabled = false)

          case ArterySettings.TlsTcp   ⇒ new ArteryTcpTransport(system, this, tlsEnabled = true)

        }

        else new Remoting(system, this))

    _internals = internals

    remotingTerminator ! internals

    _log = Logging.withMarker(eventStream, getClass.getName)

    // this enables reception of remote requests

    transport.start()

    _remoteWatcher = createRemoteWatcher(system)

    remoteDeploymentWatcher = createRemoteDeploymentWatcher(system)

  }

private final case class Internals(transport: RemoteTransport, remoteDaemon: InternalActorRef)

    extends NoSerializationVerificationNeeded

// Start assumes that it cannot be followed by another start() without having a shutdown() first

  override def start(): Unit = {

    endpointManager match {

      case None ⇒

        log.info("Starting remoting")

        val manager: ActorRef = system.systemActorOf(

          configureDispatcher(Props(classOf[EndpointManager], provider.remoteSettings.config, log)).withDeploy(Deploy.local),

          Remoting.EndpointManagerName)

        endpointManager = Some(manager)

        try {

          val addressesPromise: Promise[Seq[(AkkaProtocolTransport, Address)]] = Promise()

          manager ! Listen(addressesPromise)

          val transports: Seq[(AkkaProtocolTransport, Address)] = Await.result(

            addressesPromise.future,

            StartupTimeout.duration)

          if (transports.isEmpty) throw new RemoteTransportException("No transport drivers were loaded.", null)

          transportMapping = transports.groupBy {

            case (transport, _) ⇒ transport.schemeIdentifier

          } map { case (k, v) ⇒ k → v.toSet }

          defaultAddress = transports.head._2

          addresses = transports.map { _._2 }.toSet

          log.info("Remoting started; listening on addresses :" + addresses.mkString("[", ", ", "]"))

          manager ! StartupFinished

          eventPublisher.notifyListeners(RemotingListenEvent(addresses))

        } catch {

          case e: TimeoutException ⇒

            notifyError("Startup timed out. This is usually related to actor system host setting or host name resolution misconfiguration.", e)

            throw e

          case NonFatal(e) ⇒

            notifyError("Startup failed", e)

            throw e

        }

      case Some(_) ⇒

        log.warning("Remoting was already started. Ignoring start attempt.")

    }

  }

private def listens: Future[Seq[(AkkaProtocolTransport, Address, Promise[AssociationEventListener])]] = {

    /*

     * Constructs chains of adapters on top of each driver as given in configuration. The resulting structure looks

     * like the following:

     *   AkkaProtocolTransport <- Adapter <- ... <- Adapter <- Driver

     *

     * The transports variable contains only the heads of each chains (the AkkaProtocolTransport instances).

     */

    val transports: Seq[AkkaProtocolTransport] = for ((fqn, adapters, config) ← settings.Transports) yield {

      val args = Seq(classOf[ExtendedActorSystem] → context.system, classOf[Config] → config)

      // Loads the driver -- the bottom element of the chain.

      // The chain at this point:

      //   Driver

      val driver = extendedSystem.dynamicAccess

        .createInstanceFor[Transport](fqn, args).recover({

          case exception ⇒ throw new IllegalArgumentException(

            s"Cannot instantiate transport [$fqn]. " +

              "Make sure it extends [akka.remote.transport.Transport] and has constructor with " +

              "[akka.actor.ExtendedActorSystem] and [com.typesafe.config.Config] parameters", exception)

        }).get

      // Iteratively decorates the bottom level driver with a list of adapters.

      // The chain at this point:

      //   Adapter <- ... <- Adapter <- Driver

      val wrappedTransport =

        adapters.map { TransportAdaptersExtension.get(context.system).getAdapterProvider }.foldLeft(driver) {

          (t: Transport, provider: TransportAdapterProvider) ⇒

            // The TransportAdapterProvider will wrap the given Transport and returns with a wrapped one

            provider.create(t, context.system.asInstanceOf[ExtendedActorSystem])

        }

      // Apply AkkaProtocolTransport wrapper to the end of the chain

      // The chain at this point:

      //   AkkaProtocolTransport <- Adapter <- ... <- Adapter <- Driver

      new AkkaProtocolTransport(wrappedTransport, context.system, new AkkaProtocolSettings(conf), AkkaPduProtobufCodec)

    }

    // Collect all transports, listen addresses and listener promises in one future

    Future.sequence(transports.map { transport ⇒

      transport.listen map { case (address, listenerPromise) ⇒ (transport, address, listenerPromise) }

    })

  }