Status.open, Status.awaitOpen: utilities to wait for Statuses

These are useful utilities for acting on Status changes. For example
to replace ClientRegistry.expAllRegisteredClientsResolved(), you can
wait for a Service to be ready to serve (this includes a complete
resolve):

	Status.awaitOpen(service.status)

RB_ID=537741

finagle-core/src/main/scala/com/twitter/finagle/Status.scala

@@ -1,6 +1,6 @@
 package com.twitter.finagle
 
-import com.twitter.util.Future
+import com.twitter.util.{Await, Future}
 import scala.math.Ordering
 
 /**
@@ -23,6 +23,8 @@ sealed trait Status
  * (An [[scala.math.Ordering]] is defined in these terms.)
  */
 object Status {
+  class ClosedException extends Exception("Status was Closed; expected Open")
+
   implicit val StatusOrdering: Ordering[Status] = Ordering.by({
     case Open => 3
     case Busy(_) => 2
@@ -61,6 +63,30 @@ object Status {
   def bestOf[T](ts: Iterable[T], status: T => Status): Status =
     ts.foldLeft(Closed: Status)((a, e) => best(a, status(e)))
 
+  /**
+   * Open returns a [[com.twitter.util.Future]] that is satisfied
+   * when the status returned by `get` is [[Open]]. It returns
+   * an exceptional [[com.twitter.util.Future]] should it be
+   * [[Closed]].
+   */
+  def whenOpen(get: => Status): Future[Unit] = 
+    get match {
+      case Open => Future.Done
+      case Busy(p) => p before whenOpen(get)
+      case Closed => Future.exception(new ClosedException)
+    }
+
+  /**
+   * A blocking version of [[whenOpen]]; this method returns 
+   * when the status has become [[Open]]. This call
+   * blocks and should only be used outside of Finagle
+   * threads to halt progress until the status is [[Open]].
+   *
+   * @throws [[ClosedException]] if the status becomes [[Closed]].
+   */
+  def awaitOpen(get: => Status): Unit =
+    Await.result(whenOpen(get))
+
   /**
    * An open [[Service]] or [[ServiceFactory]] is ready to be used.
    * It can service requests or sessions immediately.

finagle-core/src/test/scala/com/twitter/finagle/StatusTest.scala

@@ -1,6 +1,6 @@
 package com.twitter.finagle
 
-import com.twitter.util.{Future, Promise}
+import com.twitter.util.{Await, Future, Promise}
 import org.junit.runner.RunWith
 import org.scalacheck.{Arbitrary, Gen}
 import org.scalatest.FunSuite
@@ -48,6 +48,23 @@ class StatusTest extends FunSuite with AssertionsForJUnit with GeneratorDrivenPr
     assert(p3.isDefined)
   }
 
+  test("Status.whenOpen") {
+    val p1, p2 = new Promise[Unit]
+    var status = Seq(Status.Busy(p1), Status.Busy(p2), Status.Open)
+    val open = Status.whenOpen {
+      val Seq(hd, rest@_*) = status
+      status = rest
+      hd
+    }
+
+    assert(!open.isDefined)
+    p1.setDone()
+    assert(!open.isDefined)
+    p2.setDone()
+    assert(open.isDefined)
+    Await.result(open)  // no exceptions
+  }
+
   test("Ordering spot check") {
     val ord = Array(Status.Closed, foreverBusy, Status.Open)
     val idx2 = for { left <- Gen.choose(0, ord.length-1);