=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,48 +103,23 @@ 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) { self =>
+      final override protected def next(): Cancellable =
         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)
+                if (self.get() != null)
                   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)
+          roundUp(initialDelay))
     }
   }
 

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

@@ -12,6 +12,7 @@ import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 import scala.util.control.NoStackTrace
 
+import akka.actor.Scheduler.AtomicCancellable
 import akka.annotation.InternalApi
 import akka.util.JavaDurationConverters
 
@@ -71,48 +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,
+    new AtomicCancellable(Cancellable.initialNotCancelled) {
+      final override protected def next(): Cancellable =
         scheduleOnce(
           initialDelay,
           new Runnable {
             override def run(): Unit = {
               try {
                 runnable.run()
-                if (self.get != null)
+                if (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] =>
               }
             }
-          }))
-
-      @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)
+          })
     }
   }
 
@@ -561,4 +537,42 @@ object Scheduler {
    * a custom implementation of `Scheduler` must also implement this.
    */
   trait TaskRunOnClose extends Runnable
+
+  private[akka] abstract class AtomicCancellable(initialValue: Cancellable)
+      extends AtomicReference[Cancellable](initialValue)
+      with Cancellable { self =>
+
+    try {
+      compareAndSet(initialValue, next())
+    } catch {
+      case cause @ SchedulerException(msg) => throw new IllegalStateException(msg, cause)
+    }
+
+    protected def next(): Cancellable
+
+    @tailrec final protected 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
+
+  }
 }