implemented minimal infrastructure to timeout

Author: rssh

src/main/scala/gopher/channels/Selector.scala

@@ -6,6 +6,7 @@ import akka.pattern._
 import scala.concurrent._
 import scala.concurrent.duration._
 import scala.language.postfixOps
+import scala.util._
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.ConcurrentLinkedQueue
@@ -33,9 +34,10 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
 
   def addTimeoutSkip(f: Skip[A] => Option[Future[Continuated[A]]], timeout: FiniteDuration):Unit =
   {
-   if (timeoutRecord.isEmpty) {
-     val locked = makeLocked(Skip(f,this))
-     timeoutRecord = Some(new TimeoutRecord(nOperations.get,timeout.toMillis,locked))
+   if (!timeoutRecord.isDefined) {
+     timeoutRecord.lastNOperations = nOperations.get
+     timeoutRecord.timeout = timeout
+     timeoutRecord.waiter = makeLocked(Skip(f,this))
    } else {
      throw new IllegalStateException("select must have only one timeout entry")
    }
@@ -123,6 +125,8 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
   private[this] def unlockAfter(f:Future[Continuated[A]], flowTermination: FlowTermination[A], dstr: String): Future[Continuated[A]] =
     f.transform(
          next => { if (mustUnlock(dstr,flowTermination)) {
+                         if (timeoutRecord.isDefined)
+                             scheduleTimeout()
                          makeLocked(next)
                    } else Never 
                  },
@@ -143,27 +147,39 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
   private[this] def scheduleTimeout():Unit =
   {
 
-    val scheduler = api.actorSystem.scheduler
-
     def tickOperation():Unit =
     {
-     if (!isCompleted) {
-      for(tr <- timeoutRecord) { 
+     if (!isCompleted && timeoutRecord.isDefined) {
         val currentNOperations = nOperations.get()
-        if (currentNOperations == tr.lastNOperations) {
+        if (currentNOperations == timeoutRecord.lastNOperations) {
            // fire
-           ???
-        } else {
-           val now = System.currentTimeMillis()
-           val nextTime = lastOperationEnd.get() + tr.timeoutMillis
-           tr.lastNOperations = nOperations.get()
-           scheduler.scheduleOnce((nextTime - now).millis)(tickOperation)
+           timeoutRecord.waiter match {
+              // TODO: add timeout field to skip
+             case sk@Skip(f,ft) => f(sk) foreach { futureNext =>
+              futureNext.onComplete {
+                case Success(next) => if (!isCompleted) {
+                                       next match {
+                                        case sk@Skip(f,ft) if (ft eq this) => timeoutRecord.waiter = sk
+                                        case other => 
+                                                   timeoutRecord.waiter = Never
+                                                   api.continuatedProcessorRef ! other
+                                       }
+                                      }
+                case Failure(ex) => if (!isCompleted) ft.doThrow(ex)
+              }
+             }
+             case other => api.continuatedProcessorRef ! other
+           }
         }
       }
-     }
     }
 
-    tickOperation()
+    if (timeoutRecord.isDefined) {
+      // TODO: make CAS 
+      timeoutRecord.lastNOperations = nOperations.get()
+      val scheduler = api.actorSystem.scheduler
+      scheduler.scheduleOnce(timeoutRecord.timeout)(tickOperation)
+    }
 
   }
 
@@ -223,34 +239,33 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
      }
    }
 
+  private[this] val log = api.actorSystem.log
 
   // false when unlocked, true otherwise.
   private[this] val lockFlag: AtomicBoolean = new AtomicBoolean(false)
 
-  // when last operation was started
-  private[this] val lastOperationEnd: AtomicLong = new AtomicLong(0L)
-  private[this] var haveTimeout: Boolean = false
-
   // number of operations, increased during each lock/unlock.
-  //  used for idle detection.
+  //  used for idle and timeout detection 
   private[channels] val nOperations = new AtomicLong();
 
   private[this] val waiters: ConcurrentLinkedQueue[Continuated[A]] = new ConcurrentLinkedQueue()
   private[this] val idleWaiters: ConcurrentLinkedQueue[Continuated[A]] = new ConcurrentLinkedQueue()
 
   private[this] class TimeoutRecord(
                         var lastNOperations: Long,
-                        var timeoutMillis: Long,
+                        var timeout: FiniteDuration,
                         var waiter: Continuated[A]
-                      ) 
-  private[this] var timeoutRecord: Option[TimeoutRecord] = None
+                      )  {
+     def isDefined:Boolean = (waiter != Never)
+  }
+  
+  private[this] val timeoutRecord: TimeoutRecord = new TimeoutRecord(0L,0 milliseconds, Never)
 
   private[this] val processor = api.continuatedProcessorRef
 
   private[this] implicit val executionContext: ExecutionContext = api.executionContext
 
 
-
 }