idle nwwo works in terms of select timeout

Author: rssh

src/main/scala/gopher/GopherAPI.scala

@@ -51,6 +51,9 @@ class GopherAPI(as: ActorSystem, es: ExecutionContext)
   def makeChannel[A](capacity: Int = 0): Channel[A] =
       Channel[A](capacity)(this)
 
+  /**
+   * create effected input with given thread-policy
+   */
   def makeEffectedInput[A](in: Input[A], threadingPolicy: ThreadingPolicy = ThreadingPolicy.Single): EffectedInput[A] =
      EffectedInput(in,threadingPolicy)
 
@@ -104,9 +107,18 @@ class GopherAPI(as: ActorSystem, es: ExecutionContext)
    **/
   def config: Config = as.settings.config.atKey("gopher")
 
+  lazy val idleTimeout: FiniteDuration = {
+    val m = try {
+              config.getInt("idle-detection-tick")
+            } catch {
+              case ex: ConfigException.Missing => 100
+            }
+    m.milliseconds
+  }
+
   def currentFlow = CurrentFlowTermination
 
-  private[gopher] val idleDetector = new IdleDetector(this)
+  //private[gopher] val idleDetector = new IdleDetector(this)
 
   private[gopher] val continuatedProcessorRef: ActorRef = {
     val props = Props(classOf[ChannelProcessor], this)

src/main/scala/gopher/channels/FoldSelectorBuilder.scala

@@ -43,14 +43,9 @@ trait FoldSelectorBuilder[T] extends SelectorBuilder[T]
         withTimeout(t){ sk => Some(f(ec,sk.flowTermination,t) map Function.const(sk)  ) }
 
 
-
-
   def idle(body:T): FoldSelectorBuilder[T] =
          macro SelectorBuilder.idleImpl[T,FoldSelectorBuilder[T]]
 
-   @inline
-   def idleWithFlowTerminationAsync[A](f: (ExecutionContext, FlowTermination[T]) => Future[T] ): this.type =
-       withIdle{ st => Some(f(ec,st.flowTermination) map Function.const(st)) }
 
 }
 

src/main/scala/gopher/channels/ForeverSelectorBuilder.scala

@@ -49,9 +49,6 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
    def idle(body:Unit): ForeverSelectorBuilder =
          macro SelectorBuilder.idleImpl[Unit,ForeverSelectorBuilder]
 
-   @inline
-   def idleWithFlowTerminationAsync(f: (ExecutionContext, FlowTermination[Unit]) => Future[Unit] ): ForeverSelectorBuilder =
-      withIdle{ st => Some(f(ec,st.flowTermination) map Function.const(st)) }
 
    /**
     * provide syntax for running select loop inside go (or async) block

src/main/scala/gopher/channels/IdleDetector.scala

@@ -47,13 +47,15 @@ class InputSelectorBuilder[T](override val api: GopherAPI) extends SelectorBuild
                                   x=>proxy.awrite(x)
                           } map Function.const(cw)) })
 
+/*
    def idle(body: T): InputSelectorBuilder[T] = 
         macro SelectorBuilder.idleImpl[T,InputSelectorBuilder[T]]
 
    @inline
    def idleWithFlowTerminationAsync(f: (ExecutionContext, FlowTermination[T]) => Future[T] ): this.type =
        withIdle{ sk => Some(f(ec,sk.flowTermination) flatMap(x => 
                             proxy.awrite(x)) map(Function.const(sk)) ) }
+*/
 
    def timeout(t:FiniteDuration)(f: FiniteDuration => T): InputSelectorBuilder[T] =
         macro SelectorBuilder.timeoutImpl[T,InputSelectorBuilder[T]]

src/main/scala/gopher/channels/InputSelectorBuilder.scala

@@ -46,13 +46,9 @@ trait OnceSelectorBuilder[T] extends SelectorBuilder[T@uncheckedVariance]
         withTimeout(t){ sk => Some(f(ec,sk.flowTermination,t).map(x => Done(x,sk.flowTermination)) ) }
 
 
-
    def idle(body: T): OnceSelectorBuilder[T] = 
         macro SelectorBuilder.idleImpl[T,OnceSelectorBuilder[T]]
 
-   @inline
-   def idleWithFlowTerminationAsync(f: (ExecutionContext, FlowTermination[T]) => Future[T] ): this.type =
-       withIdle{ sk => Some(f(ec,sk.flowTermination) map(x => Done(x,sk.flowTermination)) ) }
 
    def foreach(f:Any=>T):T = 
         macro SelectorBuilder.foreachImpl[T]

src/main/scala/gopher/channels/OnceSelectorBuilder.scala

@@ -27,11 +27,6 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
    waiters add makeLocked(ContWrite(f,ch,this))
   }
 
-  def addIdleSkip(f: Skip[A] => Option[Future[Continuated[A]]]): Unit =
-  {
-   idleWaiters add makeLocked(Skip(f,this))
-  }
-
   def addTimeout(timeout:FiniteDuration, f: Skip[A] => Option[Future[Continuated[A]]]):Unit =
   {
    if (!timeoutRecord.isDefined) {
@@ -49,19 +44,9 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
     if (timeoutRecord.isDefined) {
        scheduleTimeout()
     }
-    api.idleDetector put this
     future
   }
 
-  private[channels]  def startIdles: Unit =
-  {
-    if (idleWaiters.isEmpty) {
-       api.idleDetector.remove(this)
-    } else {
-       sendWaits(idleWaiters) 
-    }
-  }
-
   private[this] def makeLocked(block: Continuated[A]): Continuated[A] =
   {
       block match {
@@ -252,7 +237,6 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
   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,
@@ -273,4 +257,3 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
 
 
 
-

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

@@ -30,10 +30,13 @@ trait SelectorBuilder[A]
 
    def onIdle(arg: SkipSelectorArgument[A]): this.type =
    {
-     selector.addIdleSkip(arg.normalizedFun)
+     withTimeout(api.idleTimeout)(arg.normalizedFun)
      this
    }
 
+   def onTimeout(t:FiniteDuration)(arg: SkipSelectorArgument[A]): this.type =
+     withTimeout(t)(arg.normalizedFun)
+
    @inline
    def withReader[B](ch:Input[B], f: ContRead[B,A] => Option[ContRead.In[B]=>Future[Continuated[A]]]): this.type =
    {
@@ -51,8 +54,7 @@ trait SelectorBuilder[A]
    @inline
    def withIdle(f: Skip[A] => Option[Future[Continuated[A]]]):this.type =
    {
-     selector.addIdleSkip(f)
-     this
+     withTimeout(api.idleTimeout)(f)
    }
 
    @inline
@@ -183,13 +185,7 @@ object SelectorBuilder
    def idleImpl[T:c.WeakTypeTag,S](c:Context)(body:c.Expr[T]):c.Expr[S] =
    {
      import c.universe._
-     SelectorBuilder.buildAsyncCall[T,S](c)(Nil,body.tree,
-                   { (nvaldefs, nbody) =>
-                      q"""${c.prefix}.idleWithFlowTerminationAsync(
-                                    ${Function(nvaldefs,nbody)}
-                          )
-                       """
-                   })
+     c.Expr[S](q"${c.prefix}.timeout(${c.prefix}.api.idleTimeout)(_ => ${body})")
    }
 
    def timeoutImpl[T:c.WeakTypeTag,S](c:Context)(t:c.Expr[FiniteDuration])(f:c.Expr[FiniteDuration=>T]):c.Expr[S] = 
@@ -455,7 +451,7 @@ object SelectorBuilder
     if (!caseDef.guard.isEmpty) {
       c.abort(caseDef.guard.pos,"guard is not supported in select case")
     }
-    q"${builderName}.idle(${caseDef.body})"
+    q"${builderName}.timeout(${builderName}.api.idleTimeout)( _ => ${caseDef.body})"
    }
 
    def mapImpl[T:c.WeakTypeTag](c:Context)(f:c.Expr[Any=>T]):c.Expr[Input[T]] =