Add RpcAddress.fromURI and rename urls to uris

Author: zsxwing

core/src/main/scala/org/apache/spark/deploy/worker/WorkerWatcher.scala

@@ -31,7 +31,7 @@ private[spark] class WorkerWatcher(override val rpcEnv: RpcEnv, workerUrl: Strin
   override def onStart() {
     logInfo(s"Connecting to worker $workerUrl")
     if (!isTesting) {
-      rpcEnv.asyncSetupEndpointRefByUrl(workerUrl)
+      rpcEnv.asyncSetupEndpointRefByURI(workerUrl)
     }
   }
 
@@ -45,10 +45,8 @@ private[spark] class WorkerWatcher(override val rpcEnv: RpcEnv, workerUrl: Strin
   private var isTesting = false
 
   // Lets us filter events only from the worker's actor system
-  private val expectedHostPort = new java.net.URI(workerUrl)
-  private def isWorker(address: RpcAddress) = {
-    expectedHostPort.getHost == address.host && expectedHostPort.getPort == address.port
-  }
+  private val expectedAddress = RpcAddress.fromURIString(workerUrl)
+  private def isWorker(address: RpcAddress) = expectedAddress == address
 
   private def exitNonZero() = if (isTesting) isShutDown = true else System.exit(-1)
 

core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala

@@ -70,15 +70,15 @@ private[spark] abstract class RpcEnv(conf: SparkConf) {
   def setupThreadSafeEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef
 
   /**
-   * Retrieve the [[RpcEndpointRef]] represented by `url` asynchronously.
+   * Retrieve the [[RpcEndpointRef]] represented by `uri` asynchronously.
    */
-  def asyncSetupEndpointRefByUrl(url: String): Future[RpcEndpointRef]
+  def asyncSetupEndpointRefByURI(uri: String): Future[RpcEndpointRef]
 
   /**
-   * Retrieve the [[RpcEndpointRef]] represented by `url`. This is a blocking action.
+   * Retrieve the [[RpcEndpointRef]] represented by `uri`. This is a blocking action.
    */
-  def setupEndpointRefByUrl(url: String): RpcEndpointRef = {
-    Await.result(asyncSetupEndpointRefByUrl(url), defaultLookupTimeout)
+  def setupEndpointRefByURI(uri: String): RpcEndpointRef = {
+    Await.result(asyncSetupEndpointRefByURI(uri), defaultLookupTimeout)
   }
 
   /**
@@ -87,7 +87,7 @@ private[spark] abstract class RpcEnv(conf: SparkConf) {
    */
   def asyncSetupEndpointRef(
       systemName: String, address: RpcAddress, endpointName: String): Future[RpcEndpointRef] = {
-    asyncSetupEndpointRefByUrl(uriOf(systemName, address, endpointName))
+    asyncSetupEndpointRefByURI(uriOf(systemName, address, endpointName))
   }
 
   /**
@@ -96,7 +96,7 @@ private[spark] abstract class RpcEnv(conf: SparkConf) {
    */
   def setupEndpointRef(
       systemName: String, address: RpcAddress, endpointName: String): RpcEndpointRef = {
-    setupEndpointRefByUrl(uriOf(systemName, address, endpointName))
+    setupEndpointRefByURI(uriOf(systemName, address, endpointName))
   }
 
   /**
@@ -381,12 +381,18 @@ private[spark] case class RpcAddress(host: String, port: Int) {
 
 private[spark] object RpcAddress {
 
+  /**
+   * Return the [[RpcAddress]] represented by `uri`.
+   */
+  def fromURI(uri: URI): RpcAddress = {
+    RpcAddress(uri.getHost, uri.getPort)
+  }
+
   /**
    * Return the [[RpcAddress]] represented by `uri`.
    */
   def fromURIString(uri: String): RpcAddress = {
-    val u = new java.net.URI(uri)
-    RpcAddress(u.getHost, u.getPort)
+    fromURI(new java.net.URI(uri))
   }
 
   def fromSparkURL(sparkUrl: String): RpcAddress = {

core/src/main/scala/org/apache/spark/rpc/akka/AkkaRpcEnv.scala

@@ -220,9 +220,9 @@ private[spark] class AkkaRpcEnv private[akka] (
       address.port.getOrElse(defaultAddress.port))
   }
 
-  override def asyncSetupEndpointRefByUrl(url: String): Future[RpcEndpointRef] = {
+  override def asyncSetupEndpointRefByURI(uri: String): Future[RpcEndpointRef] = {
     import scala.concurrent.ExecutionContext.Implicits.global
-    actorSystem.actorSelection(url).resolveOne(defaultLookupTimeout).
+    actorSystem.actorSelection(uri).resolveOne(defaultLookupTimeout).
       map(new AkkaRpcEndpointRef(defaultAddress, _, conf))
   }