=act Extract AtomicCancellable in Scheduler.

Author: He-Pin

akka-actor/src/main/scala/akka/actor/LightArrayRevolverScheduler.scala

@@ -16,6 +16,7 @@ import scala.util.control.NonFatal
 
 import com.typesafe.config.Config
 
+import akka.actor.Scheduler.AtomicCancellable
 import akka.dispatch.AbstractNodeQueue
 import akka.event.LoggingAdapter
 import akka.util.Helpers
@@ -102,49 +103,24 @@ class LightArrayRevolverScheduler(config: Config, log: LoggingAdapter, threadFac
   override def schedule(initialDelay: FiniteDuration, delay: FiniteDuration, runnable: Runnable)(
       implicit executor: ExecutionContext): Cancellable = {
     checkMaxDelay(roundUp(delay).toNanos)
-    try new AtomicReference[Cancellable](InitialRepeatMarker) with Cancellable { self =>
-      compareAndSet(
-        InitialRepeatMarker,
+    new AtomicCancellable(
+      InitialRepeatMarker,
+      stateRef =>
         schedule(
           executor,
           new AtomicLong(clock() + initialDelay.toNanos) with Runnable {
             override def run(): Unit = {
               try {
                 runnable.run()
                 val driftNanos = clock() - getAndAdd(delay.toNanos)
-                if (self.get != null)
-                  swap(schedule(executor, this, Duration.fromNanos(Math.max(delay.toNanos - driftNanos, 1))))
+                if (stateRef.get != null)
+                  stateRef.swap(schedule(executor, this, Duration.fromNanos(Math.max(delay.toNanos - driftNanos, 1))))
               } catch {
                 case _: SchedulerException => // ignore failure to enqueue or terminated target actor
               }
             }
           },
           roundUp(initialDelay)))
-
-      @tailrec private def swap(c: Cancellable): Unit = {
-        get match {
-          case null => if (c != null) c.cancel()
-          case old  => if (!compareAndSet(old, c)) swap(c)
-        }
-      }
-
-      final def cancel(): Boolean = {
-        @tailrec def tailrecCancel(): Boolean = {
-          get match {
-            case null => false
-            case c =>
-              if (c.cancel()) compareAndSet(c, null)
-              else compareAndSet(c, null) || tailrecCancel()
-          }
-        }
-
-        tailrecCancel()
-      }
-
-      override def isCancelled: Boolean = get == null
-    } catch {
-      case cause @ SchedulerException(msg) => throw new IllegalStateException(msg, cause)
-    }
   }
 
   override def scheduleOnce(delay: FiniteDuration, runnable: Runnable)(

akka-actor/src/main/scala/akka/actor/Scheduler.scala

@@ -6,13 +6,13 @@ package akka.actor
 
 import java.util.concurrent.atomic.AtomicReference
 
+import scala.annotation.nowarn
 import scala.annotation.tailrec
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 import scala.util.control.NoStackTrace
 
-import scala.annotation.nowarn
-
+import akka.actor.Scheduler.AtomicCancellable
 import akka.annotation.InternalApi
 import akka.util.JavaDurationConverters
 
@@ -72,49 +72,23 @@ trait Scheduler {
    */
   def scheduleWithFixedDelay(initialDelay: FiniteDuration, delay: FiniteDuration)(runnable: Runnable)(
       implicit executor: ExecutionContext): Cancellable = {
-    try new AtomicReference[Cancellable](Cancellable.initialNotCancelled) with Cancellable { self =>
-      compareAndSet(
-        Cancellable.initialNotCancelled,
-        scheduleOnce(
-          initialDelay,
-          new Runnable {
-            override def run(): Unit = {
-              try {
-                runnable.run()
-                if (self.get != null)
-                  swap(scheduleOnce(delay, this))
-              } catch {
-                // ignore failure to enqueue or terminated target actor
-                case _: SchedulerException                                                                         =>
-                case e: IllegalStateException if e.getCause != null && e.getCause.isInstanceOf[SchedulerException] =>
-              }
+    new AtomicCancellable(Cancellable.initialNotCancelled, stateRef => {
+      scheduleOnce(
+        initialDelay,
+        new Runnable {
+          override def run(): Unit = {
+            try {
+              runnable.run()
+              if (stateRef.get != null)
+                stateRef.swap(scheduleOnce(delay, this))
+            } catch {
+              // ignore failure to enqueue or terminated target actor
+              case _: SchedulerException                                                                         =>
+              case e: IllegalStateException if e.getCause != null && e.getCause.isInstanceOf[SchedulerException] =>
             }
-          }))
-
-      @tailrec private def swap(c: Cancellable): Unit = {
-        get match {
-          case null => if (c != null) c.cancel()
-          case old  => if (!compareAndSet(old, c)) swap(c)
-        }
-      }
-
-      final def cancel(): Boolean = {
-        @tailrec def tailrecCancel(): Boolean = {
-          get match {
-            case null => false
-            case c =>
-              if (c.cancel()) compareAndSet(c, null)
-              else compareAndSet(c, null) || tailrecCancel()
           }
-        }
-
-        tailrecCancel()
-      }
-
-      override def isCancelled: Boolean = get == null
-    } catch {
-      case SchedulerException(msg) => throw new IllegalStateException(msg)
-    }
+        })
+    })
   }
 
   /**
@@ -562,4 +536,40 @@ object Scheduler {
    * a custom implementation of `Scheduler` must also implement this.
    */
   trait TaskRunOnClose extends Runnable
+
+  private[akka] class AtomicCancellable(initialValue: Cancellable, next: AtomicCancellable => Cancellable)
+      extends AtomicReference[Cancellable](initialValue)
+      with Cancellable { self =>
+
+    try {
+      compareAndSet(initialValue, next(self))
+    } catch {
+      case cause @ SchedulerException(msg) => throw new IllegalStateException(msg, cause)
+    }
+
+    @tailrec final private[akka] def swap(c: Cancellable): Unit = {
+      get match {
+        case null => if (c != null) c.cancel()
+        case old =>
+          if (!compareAndSet(old, c))
+            swap(c)
+      }
+    }
+
+    final def cancel(): Boolean = {
+      @tailrec def tailrecCancel(): Boolean = {
+        get match {
+          case null => false
+          case c =>
+            if (c.cancel()) compareAndSet(c, null)
+            else compareAndSet(c, null) || tailrecCancel()
+        }
+      }
+
+      tailrecCancel()
+    }
+
+    final override def isCancelled: Boolean = get == null
+
+  }
 }