raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

second round of FiniteDuration business, including cluster fixes

Browse source 
- make Scheduler only accept FiniteDuration, which has quite some
  knock-on effects
This commit is contained in:
Roland 2012-09-18 09:58:30 +02:00
parent 8a63101839
commit 35b7a9e338
30 changed files with 125 additions and 104 deletions
Download patch file Download diff file Expand all files Collapse all files

4
akka-actor-tests/src/test/scala/akka/actor/FSMActorSpec.scala
View file

@ -5,7 +5,6 @@
package akka.actor
import language.postfixOps
import org.scalatest.{ BeforeAndAfterAll, BeforeAndAfterEach }
import akka.testkit._
import TestEvent.Mute
@ -15,6 +14,7 @@ import com.typesafe.config.ConfigFactory
import scala.concurrent.Await
import akka.util.Timeout
import scala.concurrent.util.Duration
import scala.concurrent.util.FiniteDuration
object FSMActorSpec {
val timeout = Timeout(2 seconds)
@ -33,7 +33,7 @@ object FSMActorSpec {
case object Locked extends LockState
case object Open extends LockState
class Lock(code: String, timeout: Duration, latches: Latches) extends Actor with FSM[LockState, CodeState] {
class Lock(code: String, timeout: FiniteDuration, latches: Latches) extends Actor with FSM[LockState, CodeState] {
import latches._

7
akka-actor-tests/src/test/scala/akka/actor/IOActor.scala
View file

@ -5,7 +5,6 @@
package akka.actor
import language.postfixOps
import akka.util.ByteString
import scala.concurrent.{ ExecutionContext, Await, Future, Promise }
import scala.concurrent.util.{ Duration, Deadline }
@ -17,6 +16,7 @@ import akka.pattern.ask
import java.net.{ Socket, InetSocketAddress, InetAddress, SocketAddress }
import scala.util.Failure
import annotation.tailrec
import scala.concurrent.util.FiniteDuration
object IOActorSpec {
@ -244,7 +244,10 @@ class IOActorSpec extends AkkaSpec with DefaultTimeout {
* @param filter determines which exceptions should be retried
* @return a future containing the result or the last exception before a limit was hit.
*/
def retry[T](count: Option[Int] = None, timeout: Option[Duration] = None, delay: Option[Duration] = Some(100 millis), filter: Option[Throwable ⇒ Boolean] = None)(future: ⇒ Future[T])(implicit executor: ExecutionContext): Future[T] = {
def retry[T](count: Option[Int] = None,
timeout: Option[FiniteDuration] = None,
delay: Option[FiniteDuration] = Some(100 millis),
filter: Option[Throwable ⇒ Boolean] = None)(future: ⇒ Future[T])(implicit executor: ExecutionContext): Future[T] = {
val promise = Promise[T]()

3
akka-actor/src/main/java/akka/util/internal/HashedWheelTimer.java
View file

@ -25,6 +25,7 @@ import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import scala.concurrent.util.Duration;
import scala.concurrent.util.FiniteDuration;
import akka.event.LoggingAdapter;
import akka.util.Unsafe;
@ -241,7 +242,7 @@ public class HashedWheelTimer implements Timer {
return new HashedWheelTimeout(this, task, time);
}
public Timeout newTimeout(TimerTask task, Duration delay) {
public Timeout newTimeout(TimerTask task, FiniteDuration delay) {
final long currentTime = System.nanoTime();
if (task == null) {

6
akka-actor/src/main/java/akka/util/internal/Timer.java
View file

@ -15,9 +15,9 @@
*/
package akka.util.internal;
import scala.concurrent.util.Duration;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import scala.concurrent.util.FiniteDuration;
/**
* Schedules {@link TimerTask}s for one-time future execution in a background
@ -42,7 +42,7 @@ public interface Timer {
* @throws IllegalStateException if this timer has been
* {@linkplain #stop() stopped} already
*/
Timeout newTimeout(TimerTask task, Duration delay);
Timeout newTimeout(TimerTask task, FiniteDuration delay);
/**
* Releases all resources acquired by this {@link Timer} and cancels all

19
akka-actor/src/main/scala/akka/actor/FSM.scala
View file

@ -4,11 +4,11 @@
package akka.actor
import language.implicitConversions
import akka.util._
import scala.concurrent.util.Duration
import scala.collection.mutable
import akka.routing.{ Deafen, Listen, Listeners }
import scala.concurrent.util.FiniteDuration
object FSM {
@ -92,7 +92,7 @@ object FSM {
private val scheduler = context.system.scheduler
private implicit val executionContext = context.dispatcher
def schedule(actor: ActorRef, timeout: Duration): Unit =
def schedule(actor: ActorRef, timeout: FiniteDuration): Unit =
ref = Some(
if (repeat) scheduler.schedule(timeout, timeout, actor, this)
else scheduler.scheduleOnce(timeout, actor, this))
@ -121,15 +121,18 @@ object FSM {
* name, the state data, possibly custom timeout, stop reason and replies
* accumulated while processing the last message.
*/
case class State[S, D](stateName: S, stateData: D, timeout: Option[Duration] = None, stopReason: Option[Reason] = None, replies: List[Any] = Nil) {
case class State[S, D](stateName: S, stateData: D, timeout: Option[FiniteDuration] = None, stopReason: Option[Reason] = None, replies: List[Any] = Nil) {
/**
* Modify state transition descriptor to include a state timeout for the
* next state. This timeout overrides any default timeout set for the next
* state.
*
* Use Duration.Inf to deactivate an existing timeout.
*/
def forMax(timeout: Duration): State[S, D] = {
copy(timeout = Some(timeout))
def forMax(timeout: Duration): State[S, D] = timeout match {
case f: FiniteDuration ⇒ copy(timeout = Some(f))
case _ ⇒ copy(timeout = None)
}
/**
@ -245,7 +248,7 @@ trait FSM[S, D] extends Listeners with ActorLogging {
type State = FSM.State[S, D]
type StateFunction = scala.PartialFunction[Event, State]
type Timeout = Option[Duration]
type Timeout = Option[FiniteDuration]
type TransitionHandler = PartialFunction[(S, S), Unit]
/*
@ -279,7 +282,7 @@ trait FSM[S, D] extends Listeners with ActorLogging {
* @param stateTimeout default state timeout for this state
* @param stateFunction partial function describing response to input
*/
final def when(stateName: S, stateTimeout: Duration = null)(stateFunction: StateFunction): Unit =
final def when(stateName: S, stateTimeout: FiniteDuration = null)(stateFunction: StateFunction): Unit =
register(stateName, stateFunction, Option(stateTimeout))
/**
@ -339,7 +342,7 @@ trait FSM[S, D] extends Listeners with ActorLogging {
* @param repeat send once if false, scheduleAtFixedRate if true
* @return current state descriptor
*/
final def setTimer(name: String, msg: Any, timeout: Duration, repeat: Boolean): State = {
final def setTimer(name: String, msg: Any, timeout: FiniteDuration, repeat: Boolean): State = {
if (debugEvent)
log.debug("setting " + (if (repeat) "repeating " else "") + "timer '" + name + "'/" + timeout + ": " + msg)
if (timers contains name) {

35
akka-actor/src/main/scala/akka/actor/Scheduler.scala
View file

@ -13,6 +13,7 @@ import java.util.concurrent.atomic.AtomicReference
import scala.annotation.tailrec
import akka.util.internal._
import concurrent.ExecutionContext
import scala.concurrent.util.FiniteDuration
//#scheduler
/**
@ -34,8 +35,8 @@ trait Scheduler {
* Java & Scala API
*/
def schedule(
initialDelay: Duration,
frequency: Duration,
initialDelay: FiniteDuration,
frequency: FiniteDuration,
receiver: ActorRef,
message: Any)(implicit executor: ExecutionContext): Cancellable
@ -48,7 +49,7 @@ trait Scheduler {
* Scala API
*/
def schedule(
initialDelay: Duration, frequency: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable
initialDelay: FiniteDuration, frequency: FiniteDuration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable
/**
* Schedules a function to be run repeatedly with an initial delay and
@ -59,7 +60,7 @@ trait Scheduler {
* Java API
*/
def schedule(
initialDelay: Duration, frequency: Duration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable
initialDelay: FiniteDuration, frequency: FiniteDuration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable
/**
* Schedules a Runnable to be run once with a delay, i.e. a time period that
@ -67,7 +68,7 @@ trait Scheduler {
*
* Java & Scala API
*/
def scheduleOnce(delay: Duration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable
def scheduleOnce(delay: FiniteDuration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable
/**
* Schedules a message to be sent once with a delay, i.e. a time period that has
@ -75,7 +76,7 @@ trait Scheduler {
*
* Java & Scala API
*/
def scheduleOnce(delay: Duration, receiver: ActorRef, message: Any)(implicit executor: ExecutionContext): Cancellable
def scheduleOnce(delay: FiniteDuration, receiver: ActorRef, message: Any)(implicit executor: ExecutionContext): Cancellable
/**
* Schedules a function to be run once with a delay, i.e. a time period that has
@ -83,7 +84,7 @@ trait Scheduler {
*
* Scala API
*/
def scheduleOnce(delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable
def scheduleOnce(delay: FiniteDuration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable
}
//#scheduler
@ -120,8 +121,8 @@ trait Cancellable {
* returned from stop().
*/
class DefaultScheduler(hashedWheelTimer: HashedWheelTimer, log: LoggingAdapter) extends Scheduler with Closeable {
override def schedule(initialDelay: Duration,
delay: Duration,
override def schedule(initialDelay: FiniteDuration,
delay: FiniteDuration,
receiver: ActorRef,
message: Any)(implicit executor: ExecutionContext): Cancellable = {
val continuousCancellable = new ContinuousCancellable
@ -142,12 +143,12 @@ class DefaultScheduler(hashedWheelTimer: HashedWheelTimer, log: LoggingAdapter)
initialDelay))
}
override def schedule(initialDelay: Duration,
delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
override def schedule(initialDelay: FiniteDuration,
delay: FiniteDuration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
schedule(initialDelay, delay, new Runnable { override def run = f })
override def schedule(initialDelay: Duration,
delay: Duration,
override def schedule(initialDelay: FiniteDuration,
delay: FiniteDuration,
runnable: Runnable)(implicit executor: ExecutionContext): Cancellable = {
val continuousCancellable = new ContinuousCancellable
continuousCancellable.init(
@ -163,20 +164,20 @@ class DefaultScheduler(hashedWheelTimer: HashedWheelTimer, log: LoggingAdapter)
initialDelay))
}
override def scheduleOnce(delay: Duration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable =
override def scheduleOnce(delay: FiniteDuration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable =
new DefaultCancellable(
hashedWheelTimer.newTimeout(
new TimerTask() { def run(timeout: HWTimeout): Unit = executor.execute(runnable) },
delay))
override def scheduleOnce(delay: Duration, receiver: ActorRef, message: Any)(implicit executor: ExecutionContext): Cancellable =
override def scheduleOnce(delay: FiniteDuration, receiver: ActorRef, message: Any)(implicit executor: ExecutionContext): Cancellable =
scheduleOnce(delay, new Runnable { override def run = receiver ! message })
override def scheduleOnce(delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
override def scheduleOnce(delay: FiniteDuration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
scheduleOnce(delay, new Runnable { override def run = f })
private trait ContinuousScheduling { this: TimerTask ⇒
def scheduleNext(timeout: HWTimeout, delay: Duration, delegator: ContinuousCancellable) {
def scheduleNext(timeout: HWTimeout, delay: FiniteDuration, delegator: ContinuousCancellable) {
try delegator.swap(timeout.getTimer.newTimeout(this, delay)) catch { case _: IllegalStateException ⇒ } // stop recurring if timer is stopped
}
}

33
akka-actor/src/main/scala/akka/actor/cell/ReceiveTimeout.scala
View file

@ -9,9 +9,10 @@ import akka.actor.ActorCell
import akka.actor.ActorCell.emptyCancellable
import akka.actor.Cancellable
import scala.concurrent.util.Duration
import scala.concurrent.util.FiniteDuration
private[akka] object ReceiveTimeout {
final val emptyReceiveTimeoutData: (Duration, Cancellable) = (Duration.Undefined, ActorCell.emptyCancellable)
final val emptyReceiveTimeoutData: (Option[FiniteDuration], Cancellable) = (None, ActorCell.emptyCancellable)
}
private[akka] trait ReceiveTimeout { this: ActorCell ⇒
@ -19,28 +20,32 @@ private[akka] trait ReceiveTimeout { this: ActorCell ⇒
import ReceiveTimeout._
import ActorCell._
private var receiveTimeoutData: (Duration, Cancellable) = emptyReceiveTimeoutData
private var receiveTimeoutData: (Option[FiniteDuration], Cancellable) = emptyReceiveTimeoutData
final def receiveTimeout: Option[Duration] = receiveTimeoutData._1 match {
case Duration.Undefined ⇒ None
case duration ⇒ Some(duration)
final def receiveTimeout: Option[FiniteDuration] = receiveTimeoutData._1
final def setReceiveTimeout(timeout: Option[FiniteDuration]): Unit =
receiveTimeoutData = receiveTimeoutData.copy(_1 = timeout)
final def setReceiveTimeout(timeout: Duration): Unit = {
import Duration._
setReceiveTimeout(timeout match {
case x if x eq Undefined ⇒ None
case Inf | MinusInf ⇒ throw new IllegalArgumentException("receiveTimeout cannot be infinite")
case f: FiniteDuration ⇒
if (f < Zero) throw new IllegalArgumentException("receiveTimeout cannot be negative")
else Some(f)
})
}
final def setReceiveTimeout(timeout: Option[Duration]): Unit = setReceiveTimeout(timeout.getOrElse(Duration.Undefined))
final def setReceiveTimeout(timeout: Duration): Unit =
receiveTimeoutData = (
if (Duration.Undefined == timeout || timeout.toMillis < 1) Duration.Undefined else timeout,
receiveTimeoutData._2)
final def resetReceiveTimeout(): Unit = setReceiveTimeout(None)
final def checkReceiveTimeout() {
val recvtimeout = receiveTimeoutData
if (Duration.Undefined != recvtimeout._1 && !mailbox.hasMessages) {
if (recvtimeout._1.isDefined && !mailbox.hasMessages) {
recvtimeout._2.cancel() //Cancel any ongoing future
//Only reschedule if desired and there are currently no more messages to be processed
receiveTimeoutData = (recvtimeout._1, system.scheduler.scheduleOnce(recvtimeout._1, self, akka.actor.ReceiveTimeout)(this.dispatcher))
receiveTimeoutData = (recvtimeout._1, system.scheduler.scheduleOnce(recvtimeout._1.get, self, akka.actor.ReceiveTimeout)(this.dispatcher))
} else cancelReceiveTimeout()
}

4
akka-actor/src/main/scala/akka/actor/dsl/Inbox.scala
View file

@ -129,10 +129,10 @@ trait Inbox { this: ActorDSL.type ⇒
val next = clientsByTimeout.head.deadline
import context.dispatcher
if (currentDeadline.isEmpty) {
currentDeadline = Some((next, context.system.scheduler.scheduleOnce(next.timeLeft, self, Kick)))
currentDeadline = Some((next, context.system.scheduler.scheduleOnce(next.timeLeft.asInstanceOf[FiniteDuration], self, Kick)))
} else if (currentDeadline.get._1 != next) {
currentDeadline.get._2.cancel()
currentDeadline = Some((next, context.system.scheduler.scheduleOnce(next.timeLeft, self, Kick)))
currentDeadline = Some((next, context.system.scheduler.scheduleOnce(next.timeLeft.asInstanceOf[FiniteDuration], self, Kick)))
}
}
}

3
akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala
View file

@ -16,6 +16,7 @@ import scala.concurrent.forkjoin.{ ForkJoinTask, ForkJoinPool }
import scala.concurrent.util.Duration
import scala.concurrent.{ ExecutionContext, Await, Awaitable }
import scala.util.control.NonFatal
import scala.concurrent.util.FiniteDuration
final case class Envelope private (val message: Any, val sender: ActorRef)
@ -316,7 +317,7 @@ abstract class MessageDispatcher(val prerequisites: DispatcherPrerequisites) ext
*
* INTERNAL API
*/
protected[akka] def shutdownTimeout: Duration
protected[akka] def shutdownTimeout: FiniteDuration
/**
* After the call to this method, the dispatcher mustn't begin any new message processing for the specified reference

3
akka-actor/src/main/scala/akka/dispatch/BalancingDispatcher.scala
View file

@ -11,6 +11,7 @@ import akka.util.Helpers
import java.util.{ Comparator, Iterator }
import java.util.concurrent.{ Executor, LinkedBlockingQueue, ConcurrentLinkedQueue, ConcurrentSkipListSet }
import akka.actor.ActorSystemImpl
import scala.concurrent.util.FiniteDuration
/**
* An executor based event driven dispatcher which will try to redistribute work from busy actors to idle actors. It is assumed
@ -33,7 +34,7 @@ class BalancingDispatcher(
throughputDeadlineTime: Duration,
mailboxType: MailboxType,
_executorServiceFactoryProvider: ExecutorServiceFactoryProvider,
_shutdownTimeout: Duration,
_shutdownTimeout: FiniteDuration,
attemptTeamWork: Boolean)
extends Dispatcher(_prerequisites, _id, throughput, throughputDeadlineTime, mailboxType, _executorServiceFactoryProvider, _shutdownTimeout) {

3
akka-actor/src/main/scala/akka/dispatch/Dispatcher.scala
View file

@ -12,6 +12,7 @@ import java.util.concurrent.{ ExecutorService, RejectedExecutionException }
import scala.concurrent.forkjoin.ForkJoinPool
import scala.concurrent.util.Duration
import scala.concurrent.Awaitable
import scala.concurrent.util.FiniteDuration
/**
* The event-based ``Dispatcher`` binds a set of Actors to a thread pool backed up by a
@ -32,7 +33,7 @@ class Dispatcher(
val throughputDeadlineTime: Duration,
val mailboxType: MailboxType,
executorServiceFactoryProvider: ExecutorServiceFactoryProvider,
val shutdownTimeout: Duration)
val shutdownTimeout: FiniteDuration)
extends MessageDispatcher(_prerequisites) {
private class LazyExecutorServiceDelegate(factory: ExecutorServiceFactory) extends ExecutorServiceDelegate {

3
akka-actor/src/main/scala/akka/dispatch/PinnedDispatcher.scala
View file

@ -6,6 +6,7 @@ package akka.dispatch
import akka.actor.ActorCell
import scala.concurrent.util.Duration
import scala.concurrent.util.FiniteDuration
/**
* Dedicates a unique thread for each actor passed in as reference. Served through its messageQueue.
@ -18,7 +19,7 @@ class PinnedDispatcher(
_actor: ActorCell,
_id: String,
_mailboxType: MailboxType,
_shutdownTimeout: Duration,
_shutdownTimeout: FiniteDuration,
_threadPoolConfig: ThreadPoolConfig = ThreadPoolConfig())
extends Dispatcher(_prerequisites,
_id,

3
akka-actor/src/main/scala/akka/pattern/FutureTimeoutSupport.scala
View file

@ -8,13 +8,14 @@ import scala.concurrent.util.Duration
import scala.concurrent.{ ExecutionContext, Promise, Future }
import akka.actor._
import scala.util.control.NonFatal
import scala.concurrent.util.FiniteDuration
trait FutureTimeoutSupport {
/**
* Returns a [[scala.concurrent.Future]] that will be completed with the success or failure of the provided value
* after the specified duration.
*/
def after[T](duration: Duration, using: Scheduler)(value: ⇒ Future[T])(implicit ec: ExecutionContext): Future[T] =
def after[T](duration: FiniteDuration, using: Scheduler)(value: ⇒ Future[T])(implicit ec: ExecutionContext): Future[T] =
if (duration.isFinite() && duration.length < 1) {
try value catch { case NonFatal(t) ⇒ Future.failed(t) }
} else {

4
akka-actor/src/main/scala/akka/pattern/Patterns.scala
View file

@ -111,13 +111,13 @@ object Patterns {
* Returns a [[scala.concurrent.Future]] that will be completed with the success or failure of the provided Callable
* after the specified duration.
*/
def after[T](duration: Duration, scheduler: Scheduler, context: ExecutionContext, value: Callable[Future[T]]): Future[T] =
def after[T](duration: FiniteDuration, scheduler: Scheduler, context: ExecutionContext, value: Callable[Future[T]]): Future[T] =
scalaAfter(duration, scheduler)(value.call())(context)
/**
* Returns a [[scala.concurrent.Future]] that will be completed with the success or failure of the provided value
* after the specified duration.
*/
def after[T](duration: Duration, scheduler: Scheduler, context: ExecutionContext, value: Future[T]): Future[T] =
def after[T](duration: FiniteDuration, scheduler: Scheduler, context: ExecutionContext, value: Future[T]): Future[T] =
scalaAfter(duration, scheduler)(value)(context)
}

6
akka-actor/src/main/scala/akka/routing/Routing.scala
View file

@ -283,7 +283,7 @@ class RouteeProvider(val context: ActorContext, val routeeProps: Props, val resi
* The reason for the delay is to give concurrent messages a chance to be
* placed in mailbox before sending PoisonPill.
*/
def removeRoutees(nrOfInstances: Int, stopDelay: Duration): Unit = {
def removeRoutees(nrOfInstances: Int, stopDelay: FiniteDuration): Unit = {
if (nrOfInstances <= 0) {
throw new IllegalArgumentException("Expected positive nrOfInstances, got [%s]".format(nrOfInstances))
} else if (nrOfInstances > 0) {
@ -298,7 +298,7 @@ class RouteeProvider(val context: ActorContext, val routeeProps: Props, val resi
* Give concurrent messages a chance to be placed in mailbox before
* sending PoisonPill.
*/
protected def delayedStop(scheduler: Scheduler, abandon: Iterable[ActorRef], stopDelay: Duration): Unit = {
protected def delayedStop(scheduler: Scheduler, abandon: Iterable[ActorRef], stopDelay: FiniteDuration): Unit = {
if (abandon.nonEmpty) {
if (stopDelay <= Duration.Zero) {
abandon foreach (_ ! PoisonPill)
@ -1332,7 +1332,7 @@ case class DefaultResizer(
* messages a chance to be placed in mailbox before sending PoisonPill.
* Use 0 seconds to skip delay.
*/
stopDelay: Duration = 1.second,
stopDelay: FiniteDuration = 1.second,
/**
* Number of messages between resize operation.
* Use 1 to resize before each message.

16
akka-cluster/src/main/scala/akka/cluster/Cluster.scala
View file

@ -5,7 +5,6 @@
package akka.cluster
import language.implicitConversions
import akka.actor._
import akka.actor.Status._
import akka.ConfigurationException
@ -20,13 +19,12 @@ import scala.concurrent.util.{ Duration, Deadline }
import scala.concurrent.forkjoin.ThreadLocalRandom
import scala.annotation.tailrec
import scala.collection.immutable.SortedSet
import java.io.Closeable
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicReference
import akka.util.internal.HashedWheelTimer
import concurrent.{ ExecutionContext, Await }
import scala.concurrent.util.FiniteDuration
/**
* Cluster Extension Id and factory for creating Cluster extension.
@ -111,26 +109,26 @@ class Cluster(val system: ExtendedActorSystem) extends Extension {
new Scheduler with Closeable {
override def close(): Unit = () // we are using system.scheduler, which we are not responsible for closing
override def schedule(initialDelay: Duration, frequency: Duration,
override def schedule(initialDelay: FiniteDuration, frequency: FiniteDuration,
receiver: ActorRef, message: Any)(implicit executor: ExecutionContext): Cancellable =
systemScheduler.schedule(initialDelay, frequency, receiver, message)
override def schedule(initialDelay: Duration, frequency: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
override def schedule(initialDelay: FiniteDuration, frequency: FiniteDuration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
systemScheduler.schedule(initialDelay, frequency)(f)
override def schedule(initialDelay: Duration, frequency: Duration,
override def schedule(initialDelay: FiniteDuration, frequency: FiniteDuration,
runnable: Runnable)(implicit executor: ExecutionContext): Cancellable =
systemScheduler.schedule(initialDelay, frequency, runnable)
override def scheduleOnce(delay: Duration,
override def scheduleOnce(delay: FiniteDuration,
runnable: Runnable)(implicit executor: ExecutionContext): Cancellable =
systemScheduler.scheduleOnce(delay, runnable)
override def scheduleOnce(delay: Duration, receiver: ActorRef,
override def scheduleOnce(delay: FiniteDuration, receiver: ActorRef,
message: Any)(implicit executor: ExecutionContext): Cancellable =
systemScheduler.scheduleOnce(delay, receiver, message)
override def scheduleOnce(delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
override def scheduleOnce(delay: FiniteDuration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
systemScheduler.scheduleOnce(delay)(f)
}
}

11
akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala
View file

@ -16,6 +16,7 @@ import akka.cluster.MemberStatus._
import akka.cluster.ClusterEvent._
import language.existentials
import language.postfixOps
import scala.concurrent.util.FiniteDuration
/**
* Base trait for all cluster messages. All ClusterMessage's are serializable.
@ -191,32 +192,32 @@ private[cluster] final class ClusterCoreDaemon extends Actor with ActorLogging {
// start periodic gossip to random nodes in cluster
val gossipTask =
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(GossipInterval), GossipInterval) {
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(GossipInterval).asInstanceOf[FiniteDuration], GossipInterval) {
self ! GossipTick
}
// start periodic heartbeat to all nodes in cluster
val heartbeatTask =
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(HeartbeatInterval), HeartbeatInterval) {
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(HeartbeatInterval).asInstanceOf[FiniteDuration], HeartbeatInterval) {
self ! HeartbeatTick
}
// start periodic cluster failure detector reaping (moving nodes condemned by the failure detector to unreachable list)
val failureDetectorReaperTask =
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(UnreachableNodesReaperInterval), UnreachableNodesReaperInterval) {
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(UnreachableNodesReaperInterval).asInstanceOf[FiniteDuration], UnreachableNodesReaperInterval) {
self ! ReapUnreachableTick
}
// start periodic leader action management (only applies for the current leader)
private val leaderActionsTask =
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(LeaderActionsInterval), LeaderActionsInterval) {
FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(LeaderActionsInterval).asInstanceOf[FiniteDuration], LeaderActionsInterval) {
self ! LeaderActionsTick
}
// start periodic publish of current state
private val publishStateTask: Option[Cancellable] =
if (PublishStatsInterval == Duration.Zero) None
else Some(FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(PublishStatsInterval), PublishStatsInterval) {
else Some(FixedRateTask(scheduler, PeriodicTasksInitialDelay.max(PublishStatsInterval).asInstanceOf[FiniteDuration], PublishStatsInterval) {
self ! PublishStatsTick
})

9
akka-cluster/src/main/scala/akka/cluster/FixedRateTask.scala
View file

@ -9,14 +9,15 @@ import java.util.concurrent.atomic.{ AtomicBoolean, AtomicLong }
import akka.actor.{ Scheduler, Cancellable }
import scala.concurrent.util.Duration
import concurrent.ExecutionContext
import scala.concurrent.util.FiniteDuration
/**
* INTERNAL API
*/
private[akka] object FixedRateTask {
def apply(scheduler: Scheduler,
initalDelay: Duration,
delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): FixedRateTask =
initalDelay: FiniteDuration,
delay: FiniteDuration)(f: ⇒ Unit)(implicit executor: ExecutionContext): FixedRateTask =
new FixedRateTask(scheduler, initalDelay, delay, new Runnable { def run(): Unit = f })
}
@ -28,8 +29,8 @@ private[akka] object FixedRateTask {
* initialDelay.
*/
private[akka] class FixedRateTask(scheduler: Scheduler,
initalDelay: Duration,
delay: Duration,
initalDelay: FiniteDuration,
delay: FiniteDuration,
task: Runnable)(implicit executor: ExecutionContext)
extends Runnable with Cancellable {

8
akka-cluster/src/multi-jvm/scala/akka/cluster/LargeClusterSpec.scala
View file

@ -19,6 +19,7 @@ import akka.remote.testconductor.RoleName
import akka.actor.Props
import akka.actor.Actor
import akka.cluster.MemberStatus._
import scala.concurrent.util.FiniteDuration
object LargeClusterMultiJvmSpec extends MultiNodeConfig {
// each jvm simulates a datacenter with many nodes
@ -122,8 +123,9 @@ abstract class LargeClusterSpec
systems foreach { Cluster(_) }
}
def expectedMaxDuration(totalNodes: Int): Duration =
5.seconds + (2.seconds * totalNodes)
def expectedMaxDuration(totalNodes: Int): FiniteDuration =
// this cast will always succeed, but the compiler does not know about it ...
(5.seconds + (2.seconds * totalNodes)).asInstanceOf[FiniteDuration]
def joinAll(from: RoleName, to: RoleName, totalNodes: Int, runOnRoles: RoleName*): Unit = {
val joiningClusters = systems.map(Cluster(_)).toSet
@ -271,7 +273,7 @@ abstract class LargeClusterSpec
val unreachableNodes = nodesPerDatacenter
val liveNodes = nodesPerDatacenter * 4
within(30.seconds + (3.seconds * liveNodes)) {
within((30.seconds + (3.seconds * liveNodes)).asInstanceOf[FiniteDuration]) {
val startGossipCounts = Map.empty[Cluster, Long] ++
systems.map(sys ⇒ (Cluster(sys) -> Cluster(sys).readView.latestStats.receivedGossipCount))
def gossipCount(c: Cluster): Long = {

4
akka-cluster/src/multi-jvm/scala/akka/cluster/MultiNodeClusterSpec.scala
View file

@ -4,7 +4,6 @@
package akka.cluster
import language.implicitConversions
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory
import akka.actor.{ Address, ExtendedActorSystem }
@ -18,6 +17,7 @@ import org.scalatest.exceptions.TestFailedException
import java.util.concurrent.ConcurrentHashMap
import akka.actor.ActorPath
import akka.actor.RootActorPath
import scala.concurrent.util.FiniteDuration
object MultiNodeClusterSpec {
@ -174,7 +174,7 @@ trait MultiNodeClusterSpec extends Suite with STMultiNodeSpec { self: MultiNodeS
def awaitUpConvergence(
numberOfMembers: Int,
canNotBePartOfMemberRing: Seq[Address] = Seq.empty[Address],
timeout: Duration = 20.seconds): Unit = {
timeout: FiniteDuration = 20.seconds): Unit = {
within(timeout) {
awaitCond(clusterView.members.size == numberOfMembers)
awaitCond(clusterView.members.forall(_.status == MemberStatus.Up))

4
akka-docs/java/code/docs/actor/japi/FaultHandlingDocSample.java
View file

@ -139,7 +139,7 @@ public class FaultHandlingDocSample {
if (msg.equals(Start) && progressListener == null) {
progressListener = getSender();
getContext().system().scheduler().schedule(
Duration.Zero(), Duration.parse("1 second"), getSelf(), Do, getContext().dispatcher()
Duration.Zero(), Duration.create(1, "second"), getSelf(), Do, getContext().dispatcher()
);
} else if (msg.equals(Do)) {
counterService.tell(new Increment(1), getSelf());
@ -299,7 +299,7 @@ public class FaultHandlingDocSample {
counter.tell(new UseStorage(null), getSelf());
// Try to re-establish storage after while
getContext().system().scheduler().scheduleOnce(
Duration.parse("10 seconds"), getSelf(), Reconnect, getContext().dispatcher()
Duration.create(10, "seconds"), getSelf(), Reconnect, getContext().dispatcher()
);
} else if (msg.equals(Reconnect)) {
// Re-establish storage after the scheduled delay

2
akka-docs/java/code/docs/future/FutureDocTestBase.java
View file

@ -518,7 +518,7 @@ public class FutureDocTestBase {
//#after
final ExecutionContext ec = system.dispatcher();
Future<String> failExc = Futures.failed(new IllegalStateException("OHNOES1"));
Future<String> delayed = Patterns.after(Duration.parse("500 millis"),
Future<String> delayed = Patterns.after(Duration.create(500, "millis"),
system.scheduler(), ec, failExc);
Future<String> future = future(new Callable<String>() {
public String call() throws InterruptedException {

2
akka-docs/java/code/docs/zeromq/ZeromqDocTestBase.java
View file

@ -187,7 +187,7 @@ public class ZeromqDocTestBase {
@Override
public void preStart() {
getContext().system().scheduler()
.schedule(Duration.parse("1 second"), Duration.parse("1 second"), getSelf(), TICK, getContext().dispatcher());
.schedule(Duration.create(1, "second"), Duration.create(1, "second"), getSelf(), TICK, getContext().dispatcher());
}
@Override

5
akka-remote-tests/src/main/scala/akka/remote/testconductor/Conductor.scala
View file

@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
import akka.util.{ Timeout }
import scala.concurrent.util.{ Deadline, Duration }
import scala.reflect.classTag
import scala.concurrent.util.FiniteDuration
sealed trait Direction {
def includes(other: Direction): Boolean
@ -559,7 +560,7 @@ private[akka] class BarrierCoordinator extends Actor with LoggingFSM[BarrierCoor
}
onTransition {
case Idle -> Waiting ⇒ setTimer("Timeout", StateTimeout, nextStateData.deadline.timeLeft, false)
case Idle -> Waiting ⇒ setTimer("Timeout", StateTimeout, nextStateData.deadline.timeLeft.asInstanceOf[FiniteDuration], false)
case Waiting -> Idle ⇒ cancelTimer("Timeout")
}
@ -570,7 +571,7 @@ private[akka] class BarrierCoordinator extends Actor with LoggingFSM[BarrierCoor
val enterDeadline = getDeadline(timeout)
// we only allow the deadlines to get shorter
if (enterDeadline.timeLeft < deadline.timeLeft) {
setTimer("Timeout", StateTimeout, enterDeadline.timeLeft, false)
setTimer("Timeout", StateTimeout, enterDeadline.timeLeft.asInstanceOf[FiniteDuration], false)
handleBarrier(d.copy(arrived = together, deadline = enterDeadline))
} else
handleBarrier(d.copy(arrived = together))

13
akka-remote-tests/src/main/scala/akka/remote/testconductor/NetworkFailureInjector.scala
View file

@ -4,20 +4,17 @@
package akka.remote.testconductor
import language.postfixOps
import java.net.InetSocketAddress
import scala.annotation.tailrec
import scala.collection.immutable.Queue
import org.jboss.netty.buffer.ChannelBuffer
import org.jboss.netty.channel.{ SimpleChannelHandler, MessageEvent, Channels, ChannelStateEvent, ChannelHandlerContext, ChannelFutureListener, ChannelFuture }
import akka.actor.{ Props, LoggingFSM, Address, ActorSystem, ActorRef, ActorLogging, Actor, FSM }
import akka.event.Logging
import akka.remote.netty.ChannelAddress
import scala.concurrent.util.Duration
import scala.concurrent.util.duration._
import scala.concurrent.util.FiniteDuration
/**
* INTERNAL API.
@ -331,20 +328,20 @@ private[akka] class ThrottleActor(channelContext: ChannelHandlerContext)
* lead to the correct rate on average, with increased latency of the order of
* HWT granularity.
*/
private def schedule(d: Data): (Data, Seq[Send], Option[Duration]) = {
private def schedule(d: Data): (Data, Seq[Send], Option[FiniteDuration]) = {
val now = System.nanoTime
@tailrec def rec(d: Data, toSend: Seq[Send]): (Data, Seq[Send], Option[Duration]) = {
@tailrec def rec(d: Data, toSend: Seq[Send]): (Data, Seq[Send], Option[FiniteDuration]) = {
if (d.queue.isEmpty) (d, toSend, None)
else {
val timeForPacket = d.lastSent + (1000 * size(d.queue.head.msg) / d.rateMBit).toLong
if (timeForPacket <= now) rec(Data(timeForPacket, d.rateMBit, d.queue.tail), toSend :+ d.queue.head)
else {
val splitThreshold = d.lastSent + packetSplitThreshold.toNanos
if (now < splitThreshold) (d, toSend, Some((timeForPacket - now).nanos min (splitThreshold - now).nanos))
if (now < splitThreshold) (d, toSend, Some(((timeForPacket - now).nanos min (splitThreshold - now).nanos).asInstanceOf[FiniteDuration]))
else {
val microsToSend = (now - d.lastSent) / 1000
val (s1, s2) = split(d.queue.head, (microsToSend * d.rateMBit / 8).toInt)
(d.copy(queue = s2 +: d.queue.tail), toSend :+ s1, Some((timeForPacket - now).nanos min packetSplitThreshold))
(d.copy(queue = s2 +: d.queue.tail), toSend :+ s1, Some(((timeForPacket - now).nanos min packetSplitThreshold).asInstanceOf[FiniteDuration]))
}
}
}

2
akka-remote-tests/src/main/scala/akka/remote/testconductor/Player.scala
View file

@ -278,7 +278,7 @@ private[akka] class PlayerHandler(
event.getCause match {
case c: ConnectException if reconnects > 0 ⇒
reconnects -= 1
scheduler.scheduleOnce(nextAttempt.timeLeft)(reconnect())
scheduler.scheduleOnce(nextAttempt.timeLeft.asInstanceOf[FiniteDuration])(reconnect())
case e ⇒ fsm ! ConnectionFailure(e.getMessage)
}
}

5
akka-samples/akka-sample-fsm/src/main/scala/Buncher.scala
View file

@ -7,6 +7,7 @@ import akka.actor.ActorRefFactory
import scala.reflect.ClassTag
import scala.concurrent.util.Duration
import akka.actor.{ FSM, Actor, ActorRef }
import scala.concurrent.util.FiniteDuration
/*
* generic typed object buncher.
@ -35,7 +36,7 @@ object GenericBuncher {
}
}
abstract class GenericBuncher[A: ClassTag, B](val singleTimeout: Duration, val multiTimeout: Duration)
abstract class GenericBuncher[A: ClassTag, B](val singleTimeout: FiniteDuration, val multiTimeout: FiniteDuration)
extends Actor with FSM[GenericBuncher.State, B] {
import GenericBuncher._
import FSM._
@ -85,7 +86,7 @@ object Buncher {
val Flush = GenericBuncher.Flush
}
class Buncher[A: ClassTag](singleTimeout: Duration, multiTimeout: Duration)
class Buncher[A: ClassTag](singleTimeout: FiniteDuration, multiTimeout: FiniteDuration)
extends GenericBuncher[A, List[A]](singleTimeout, multiTimeout) {
import Buncher._

4
akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnFsm.scala
View file

@ -4,11 +4,11 @@
package sample.fsm.dining.fsm
import language.postfixOps
import akka.actor._
import akka.actor.FSM._
import scala.concurrent.util.Duration
import scala.concurrent.util.duration._
import scala.concurrent.util.FiniteDuration
/*
* Some messages for the chopstick
@ -159,7 +159,7 @@ class FSMHakker(name: String, left: ActorRef, right: ActorRef) extends Actor wit
// Initialize the hakker
initialize
private def startThinking(duration: Duration): State = {
private def startThinking(duration: FiniteDuration): State = {
goto(Thinking) using TakenChopsticks(None, None) forMax duration
}
}

5
akka-testkit/src/main/scala/akka/testkit/TestFSMRef.scala
View file

@ -7,6 +7,7 @@ package akka.testkit
import akka.actor._
import scala.concurrent.util.Duration
import akka.dispatch.DispatcherPrerequisites
import scala.concurrent.util.FiniteDuration
/**
* This is a specialised form of the TestActorRef with support for querying and
@ -59,14 +60,14 @@ class TestFSMRef[S, D, T <: Actor](
* corresponding transition initiated from within the FSM, including timeout
* and stop handling.
*/
def setState(stateName: S = fsm.stateName, stateData: D = fsm.stateData, timeout: Duration = null, stopReason: Option[FSM.Reason] = None) {
def setState(stateName: S = fsm.stateName, stateData: D = fsm.stateData, timeout: FiniteDuration = null, stopReason: Option[FSM.Reason] = None) {
fsm.applyState(FSM.State(stateName, stateData, Option(timeout), stopReason))
}
/**
* Proxy for FSM.setTimer.
*/
def setTimer(name: String, msg: Any, timeout: Duration, repeat: Boolean) {
def setTimer(name: String, msg: Any, timeout: FiniteDuration, repeat: Boolean) {
fsm.setTimer(name, msg, timeout, repeat)
}

3
akka-zeromq/src/main/scala/akka/zeromq/SocketOption.scala
View file

@ -9,6 +9,7 @@ import akka.actor.ActorRef
import scala.concurrent.util.duration._
import scala.concurrent.util.Duration
import org.zeromq.ZMQ.{ Poller, Socket }
import scala.concurrent.util.FiniteDuration
/**
* Marker trait representing request messages for zeromq
@ -152,7 +153,7 @@ case class PollDispatcher(name: String) extends SocketMeta
* An option containing the duration a poll cycle should wait for a message before it loops
* @param duration
*/
case class PollTimeoutDuration(duration: Duration = 100 millis) extends SocketMeta
case class PollTimeoutDuration(duration: FiniteDuration = 100 millis) extends SocketMeta
/**
* Start listening with this server socket on the specified address