* ActorRef actor = Actors.actorOf(MyUntypedActor.class, "my-actor-address");
- * actor.start();
* actor.tell(message, context);
* actor.stop();
*
* You can create and start the actor in one statement like this:
*
- * val actor = Actors.actorOf(MyActor.class, "my-actor-address").start();
+ * val actor = Actors.actorOf(MyActor.class, "my-actor-address");
*
*/
public static ActorRef actorOf(final Class type, final String address) {
@@ -101,13 +98,12 @@ public class Actors {
* Example in Java:
*
* ActorRef actor = Actors.actorOf(MyUntypedActor.class, "my-actor-address");
- * actor.start();
* actor.tell(message, context);
* actor.stop();
*
* You can create and start the actor in one statement like this:
*
- * val actor = Actors.actorOf(MyActor.class).start();
+ * val actor = Actors.actorOf(MyActor.class);
*
*/
public static ActorRef actorOf(final Class type) {
diff --git a/akka-actor/src/main/scala/akka/actor/Actor.scala b/akka-actor/src/main/scala/akka/actor/Actor.scala
index 28ee27bae3..cadc06d290 100644
--- a/akka-actor/src/main/scala/akka/actor/Actor.scala
+++ b/akka-actor/src/main/scala/akka/actor/Actor.scala
@@ -240,13 +240,12 @@ object Actor {
*
* import Actor._
* val actor = actorOf[MyActor]
- * actor.start()
* actor ! message
* actor.stop()
*
* You can create and start the actor in one statement like this:
*
- * val actor = actorOf[MyActor].start()
+ * val actor = actorOf[MyActor]
*
*/
def actorOf[T <: Actor: Manifest](address: String): ActorRef =
@@ -258,13 +257,12 @@ object Actor {
*
* import Actor._
* val actor = actorOf[MyActor]
- * actor.start
* actor ! message
* actor.stop
*
* You can create and start the actor in one statement like this:
*
- * val actor = actorOf[MyActor].start
+ * val actor = actorOf[MyActor]
*
*/
def actorOf[T <: Actor: Manifest]: ActorRef =
@@ -276,13 +274,12 @@ object Actor {
*
* import Actor._
* val actor = actorOf(classOf[MyActor])
- * actor.start()
* actor ! message
* actor.stop()
*
* You can create and start the actor in one statement like this:
*
- * val actor = actorOf(classOf[MyActor]).start()
+ * val actor = actorOf(classOf[MyActor])
*
*/
def actorOf[T <: Actor](clazz: Class[T]): ActorRef =
@@ -293,18 +290,15 @@ object Actor {
*
* import Actor._
* val actor = actorOf(classOf[MyActor])
- * actor.start
* actor ! message
* actor.stop
*
* You can create and start the actor in one statement like this:
*
- * val actor = actorOf(classOf[MyActor]).start
+ * val actor = actorOf(classOf[MyActor])
*
*/
- def actorOf[T <: Actor](clazz: Class[T], address: String): ActorRef = {
- createActor(address, () ⇒ newLocalActorRef(clazz, address))
- }
+ def actorOf[T <: Actor](clazz: Class[T], address: String): ActorRef = actorOf(Props(clazz), address)
/**
* Creates an ActorRef out of the Actor. Allows you to pass in a factory function
@@ -315,16 +309,15 @@ object Actor {
*
* import Actor._
* val actor = actorOf(new MyActor)
- * actor.start()
* actor ! message
* actor.stop()
*
* You can create and start the actor in one statement like this:
*
- * val actor = actorOf(new MyActor).start()
+ * val actor = actorOf(new MyActor)
*
*/
- def actorOf[T <: Actor](factory: ⇒ T): ActorRef = actorOf(factory, new UUID().toString)
+ def actorOf[T <: Actor](factory: ⇒ T): ActorRef = actorOf(factory, newUuid().toString)
/**
* Creates an ActorRef out of the Actor. Allows you to pass in a factory function
@@ -335,18 +328,15 @@ object Actor {
*
* import Actor._
* val actor = actorOf(new MyActor)
- * actor.start
* actor ! message
* actor.stop
*
* You can create and start the actor in one statement like this:
*
- * val actor = actorOf(new MyActor).start
+ * val actor = actorOf(new MyActor)
*
*/
- def actorOf[T <: Actor](creator: ⇒ T, address: String): ActorRef = {
- createActor(address, () ⇒ new LocalActorRef(Props(creator = () ⇒ creator), address))
- }
+ def actorOf[T <: Actor](creator: ⇒ T, address: String): ActorRef = actorOf(Props(creator), address)
/**
* Creates an ActorRef out of the Actor. Allows you to pass in a factory (Creator
* self.dispatcher = ...
- * self.trapExit = ...
* self.faultHandler = ...
* self.lifeCycle = ...
* self.sender
@@ -612,7 +595,6 @@ trait Actor {
* self.reply(..)
* self.link(..)
* self.unlink(..)
- * self.start(..)
* self.stop(..)
*
*/
@@ -645,7 +627,7 @@ trait Actor {
/**
* User overridable callback.
*
- * Is called when an Actor is started by invoking 'actor.start()'.
+ * Is called when an Actor is started by invoking 'actor'.
*/
def preStart() {}
diff --git a/akka-actor/src/main/scala/akka/actor/ActorRef.scala b/akka-actor/src/main/scala/akka/actor/ActorRef.scala
index 1a1f439ad5..5ee4d6e4c7 100644
--- a/akka-actor/src/main/scala/akka/actor/ActorRef.scala
+++ b/akka-actor/src/main/scala/akka/actor/ActorRef.scala
@@ -53,7 +53,6 @@ object Props {
final val defaultLifeCycle: LifeCycle = Permanent
final val defaultFaultHandler: FaultHandlingStrategy = NoFaultHandlingStrategy
final val defaultSupervisor: Option[ActorRef] = None
- final val defaultLocalOnly: Boolean = false
/**
* The default Props instance, uses the settings from the Props object starting with default*
@@ -85,8 +84,13 @@ object Props {
*/
def apply(creator: ⇒ Actor): Props = default.withCreator(creator)
- def apply(behavior: (ScalaActorRef with SelfActorRef) ⇒ Actor.Receive): Props =
- apply(new Actor { def receive = behavior(self) })
+ /**
+ * Returns a Props that has default values except for "creator" which will be a function that creates an instance
+ * using the supplied thunk
+ */
+ def apply(creator: Creator[_ <: Actor]): Props = default.withCreator(creator.create)
+
+ def apply(behavior: (ScalaActorRef with SelfActorRef) ⇒ Actor.Receive): Props = apply(new Actor { def receive = behavior(self) })
}
/**
@@ -94,12 +98,11 @@ object Props {
*/
case class Props(creator: () ⇒ Actor = Props.defaultCreator,
deployId: String = Props.defaultDeployId,
- dispatcher: MessageDispatcher = Props.defaultDispatcher,
+ @transient dispatcher: MessageDispatcher = Props.defaultDispatcher,
timeout: Timeout = Props.defaultTimeout,
lifeCycle: LifeCycle = Props.defaultLifeCycle,
faultHandler: FaultHandlingStrategy = Props.defaultFaultHandler,
- supervisor: Option[ActorRef] = Props.defaultSupervisor,
- localOnly: Boolean = Props.defaultLocalOnly) {
+ supervisor: Option[ActorRef] = Props.defaultSupervisor) {
/**
* No-args constructor that sets all the default values
* Java API
@@ -111,8 +114,7 @@ case class Props(creator: () ⇒ Actor = Props.defaultCreator,
timeout = Props.defaultTimeout,
lifeCycle = Props.defaultLifeCycle,
faultHandler = Props.defaultFaultHandler,
- supervisor = Props.defaultSupervisor,
- localOnly = Props.defaultLocalOnly)
+ supervisor = Props.defaultSupervisor)
/**
* Returns a new Props with the specified creator set
@@ -173,12 +175,6 @@ case class Props(creator: () ⇒ Actor = Props.defaultCreator,
* Scala API
*/
def withSupervisor(s: scala.Option[ActorRef]) = copy(supervisor = s)
-
- /**
- * Returns a new Props with the specified localOnly set
- * Java and Scala API
- */
- def withLocalOnly(l: Boolean) = copy(localOnly = l)
}
/**
@@ -191,14 +187,13 @@ case class Props(creator: () ⇒ Actor = Props.defaultCreator,
* import Actor._
*
* val actor = actorOf[MyActor]
- * actor.start()
* actor ! message
* actor.stop()
*
*
* You can also create and start actors like this:
* - * val actor = actorOf[MyActor].start() + * val actor = actorOf[MyActor] ** * Here is an example on how to create an actor with a non-default constructor. @@ -206,7 +201,6 @@ case class Props(creator: () ⇒ Actor = Props.defaultCreator, * import Actor._ * * val actor = actorOf(new MyActor(...)) - * actor.start() * actor ! message * actor.stop() * @@ -218,6 +212,7 @@ abstract class ActorRef extends ActorRefShared with UntypedChannel with ReplyCha // Only mutable for RemoteServer in order to maintain identity across nodes @volatile protected[akka] var _uuid = newUuid + @volatile protected[this] var _status: ActorRefInternals.StatusType = ActorRefInternals.UNSTARTED @@ -241,19 +236,6 @@ abstract class ActorRef extends ActorRefShared with UntypedChannel with ReplyCha def uuid = _uuid - /** - * Is the actor running? - */ - def isRunning: Boolean = _status match { - case ActorRefInternals.BEING_RESTARTED | ActorRefInternals.RUNNING ⇒ true - case _ ⇒ false - } - - /** - * Is the actor shut down? - */ - def isShutdown: Boolean = _status == ActorRefInternals.SHUTDOWN - /** * Only for internal use. UUID is effectively final. */ @@ -312,11 +294,6 @@ abstract class ActorRef extends ActorRefShared with UntypedChannel with ReplyCha else forward(message)(sender) } - /** - * Starts up the actor and its message queue. - */ - def start(): this.type - /** * Shuts down the actor its dispatcher and message queue. * Alias for 'stop'. @@ -328,7 +305,20 @@ abstract class ActorRef extends ActorRefShared with UntypedChannel with ReplyCha /** * Shuts down the actor its dispatcher and message queue. */ - def stop() + def stop(): Unit + + /** + * Is the actor running? + */ + def isRunning: Boolean = _status match { //TODO Remove this method + case ActorRefInternals.BEING_RESTARTED | ActorRefInternals.RUNNING ⇒ true + case _ ⇒ false + } + + /** + * Is the actor shut down? + */ + def isShutdown: Boolean = _status == ActorRefInternals.SHUTDOWN /** * Links an other actor to this actor. Links are unidirectional and means that a the linking actor will @@ -511,10 +501,10 @@ abstract class SelfActorRef extends ActorRef with ForwardableChannel { self: Loc * * @author Jonas Bonér */ -class LocalActorRef private[akka] (private[this] val props: Props, val address: String) +class LocalActorRef private[akka] (private[this] val props: Props, val address: String, val systemService: Boolean) extends SelfActorRef with ScalaActorRef { - protected[akka] val guard = new ReentrantGuard + protected[akka] val guard = new ReentrantGuard //TODO FIXME remove the last synchronization point @volatile protected[akka] var _futureTimeout: Option[ScheduledFuture[AnyRef]] = None @@ -550,12 +540,12 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: "Could not create Serializer object for [" + this.getClass.getName + "]") } - private lazy val hasReplicationStorage: Boolean = { + private lazy val hasReplicationStorage: Boolean = if (!systemService) { import DeploymentConfig._ isReplicated(replicationSchemeFor(Deployer.deploymentFor(address)).getOrElse(Transient)) - } + } else false - private lazy val replicationStorage: Option[TransactionLog] = { + private lazy val replicationStorage: Option[TransactionLog] = if (!systemService) { import DeploymentConfig._ val replicationScheme = replicationSchemeFor(Deployer.deploymentFor(address)).getOrElse(Transient) if (isReplicated(replicationScheme)) { @@ -569,7 +559,7 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: throw new ConfigurationException("Unknown replication storage type [" + replicationScheme + "]") } } else None - } + } else None // If it was started inside "newActor", initialize it if (isRunning) initializeActorInstance @@ -582,13 +572,12 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: __receiveTimeout: Option[Long], __hotswap: Stack[PartialFunction[Any, Unit]]) = { - this(__props, __address) + this(__props, __address, systemService = false) //Doesn't make any sense to move a system service _uuid = __uuid hotswap = __hotswap receiveTimeout = __receiveTimeout setActorSelfFields(actorInstance.get(), this) //TODO Why is this needed? - start() } // ========= PUBLIC FUNCTIONS ========= @@ -601,9 +590,8 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: /** * Starts up the actor and its message queue. */ - def start(): this.type = guard.withGuard[this.type] { - if (isShutdown) throw new ActorStartException( - "Can't restart an actor that has been shut down with 'stop' or 'exit'") + protected[akka] def startInternal(): this.type = guard.withGuard[this.type] { + if (isShutdown) throw new ActorStartException("Can't restart an actor that has been shut down with 'stop' or 'exit'") if (!isRunning) { dispatcher.attach(this) _status = ActorRefInternals.RUNNING @@ -642,8 +630,13 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: a.postStop() } finally { currentMessage = null - - setActorSelfFields(actorInstance.get, null) + try { //When a supervisor is stopped, it's linked actors should also be stopped + val i = _linkedActors.values.iterator + while (i.hasNext) { + i.next.stop() + i.remove + } + } finally { setActorSelfFields(actorInstance.get, null) } } } @@ -741,19 +734,20 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: } protected[akka] def postMessageToMailbox(message: Any, channel: UntypedChannel): Unit = - dispatcher dispatchMessage new MessageInvocation(this, message, channel) + if (isRunning) dispatcher dispatchMessage new MessageInvocation(this, message, channel) + else throw new ActorInitializationException("Actor has not been started, you need to invoke 'actor' before using it") protected[akka] def postMessageToMailboxAndCreateFutureResultWithTimeout( message: Any, timeout: Timeout, - channel: UntypedChannel): Future[Any] = { + channel: UntypedChannel): Future[Any] = if (isRunning) { val future = channel match { case f: ActorPromise ⇒ f case _ ⇒ new ActorPromise(timeout)(dispatcher) } dispatcher dispatchMessage new MessageInvocation(this, message, future) future - } + } else throw new ActorInitializationException("Actor has not been started, you need to invoke 'actor' before using it") /** * Callback for the dispatcher. This is the single entry point to the user Actor implementation. @@ -800,8 +794,7 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: dead.restart(reason, maxRetries, within) case _ ⇒ - if (_supervisor.isDefined) notifySupervisorWithMessage(Death(this, reason)) - else dead.stop() + if (_supervisor.isDefined) throw reason else dead.stop() //Escalate problem if not handled here } } @@ -975,22 +968,8 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: } private def notifySupervisorWithMessage(notification: LifeCycleMessage) { - // FIXME to fix supervisor restart of remote actor for oneway calls, inject a supervisor proxy that can send notification back to client - _supervisor.foreach { sup ⇒ - if (sup.isShutdown) { - // if supervisor is shut down, game over for all linked actors - //Scoped stop all linked actors, to avoid leaking the 'i' val - { - val i = _linkedActors.values.iterator - while (i.hasNext) { - i.next.stop() - i.remove - } - } - //Stop the actor itself - stop - } else sup ! notification // else notify supervisor - } + val sup = _supervisor + if (sup.isDefined) sup.get ! notification } private def setActorSelfFields(actor: Actor, value: ActorRef) { @@ -1043,6 +1022,8 @@ class LocalActorRef private[akka] (private[this] val props: Props, val address: _futureTimeout = None } } + + startInternal() } /** @@ -1069,8 +1050,6 @@ private[akka] case class RemoteActorRef private[akka] ( ClusterModule.ensureEnabled() - start() - protected[akka] override def timeout: Long = _timeout def postMessageToMailbox(message: Any, channel: UntypedChannel): Unit = { @@ -1091,12 +1070,6 @@ private[akka] case class RemoteActorRef private[akka] ( else throw new IllegalActorStateException("Expected a future from remote call to actor " + toString) } - def start(): this.type = synchronized[this.type] { - if (_status == ActorRefInternals.UNSTARTED) - _status = ActorRefInternals.RUNNING - this - } - def stop() { synchronized { if (_status == ActorRefInternals.RUNNING) { @@ -1126,6 +1099,10 @@ private[akka] case class RemoteActorRef private[akka] ( } private def unsupported = throw new UnsupportedOperationException("Not supported for RemoteActorRef") + + /* If you start me up... */ + if (_status == ActorRefInternals.UNSTARTED) + _status = ActorRefInternals.RUNNING } /** @@ -1156,8 +1133,7 @@ trait ActorRefShared { * There are implicit conversions in ../actor/Implicits.scala * from ActorRef -> ScalaActorRef and back */ -trait ScalaActorRef extends ActorRefShared with ReplyChannel[Any] { - ref: ActorRef ⇒ +trait ScalaActorRef extends ActorRefShared with ReplyChannel[Any] { ref: ActorRef ⇒ /** * Sends a one-way asynchronous message. E.g. fire-and-forget semantics. @@ -1173,23 +1149,12 @@ trait ScalaActorRef extends ActorRefShared with ReplyChannel[Any] { * * */ - def !(message: Any)(implicit channel: UntypedChannel): Unit = { - if (isRunning) postMessageToMailbox(message, channel) - else throw new ActorInitializationException( - "Actor has not been started, you need to invoke 'actor.start()' before using it") - } + def !(message: Any)(implicit channel: UntypedChannel): Unit = postMessageToMailbox(message, channel) /** * Sends a message asynchronously, returning a future which may eventually hold the reply. */ - def ?(message: Any)(implicit channel: UntypedChannel, timeout: Timeout): Future[Any] = { - //FIXME: so it can happen that a message is posted after the actor has been shut down (the isRunning and postMessageToMailboxAndCreateFutureResultWithTimeout are not atomic. - if (isRunning) { - postMessageToMailboxAndCreateFutureResultWithTimeout(message, timeout, channel) - //FIXME: there is no after check if the running state is still true.. so no 'repairing' - } else throw new ActorInitializationException( - "Actor has not been started, you need to invoke 'actor.start()' before using it") - } + def ?(message: Any)(implicit channel: UntypedChannel, timeout: Timeout): Future[Any] = postMessageToMailboxAndCreateFutureResultWithTimeout(message, timeout, channel) def ?(message: Any, timeout: Timeout)(implicit channel: UntypedChannel): Future[Any] = ?(message)(channel, timeout) @@ -1198,14 +1163,7 @@ trait ScalaActorRef extends ActorRefShared with ReplyChannel[Any] { * * Works with '!' and '?'/'ask'. */ - def forward(message: Any)(implicit channel: ForwardableChannel) = { - //FIXME: so it can happen that a message is posted after the actor has been shut down (the isRunning and postMessageToMailbox are not atomic. - if (isRunning) { - postMessageToMailbox(message, channel.channel) - //FIXME: there is no after check if the running state is still true.. so no 'repairing' - } else throw new ActorInitializationException( - "Actor has not been started, you need to invoke 'actor.start()' before using it") - } + def forward(message: Any)(implicit channel: ForwardableChannel) = postMessageToMailbox(message, channel.channel) } /** diff --git a/akka-actor/src/main/scala/akka/actor/Deployer.scala b/akka-actor/src/main/scala/akka/actor/Deployer.scala index 40d00fe589..f9fbb19fb2 100644 --- a/akka-actor/src/main/scala/akka/actor/Deployer.scala +++ b/akka-actor/src/main/scala/akka/actor/Deployer.scala @@ -287,7 +287,7 @@ object LocalDeployer extends ActorDeployer { deployments.putIfAbsent(deployment.address, deployment) /* match { case null ⇒ deployment match { - case Deploy(address, Some(recipe), routing, _) ⇒ Actor.actorOf(recipe.implementationClass, address).start() //FIXME use routing? + case Deploy(address, Some(recipe), routing, _) ⇒ Actor.actorOf(recipe.implementationClass, address) //FIXME use routing? case _ ⇒ } case `deployment` ⇒ //Already deployed TODO should it be like this? diff --git a/akka-actor/src/main/scala/akka/actor/IO.scala b/akka-actor/src/main/scala/akka/actor/IO.scala index 60f8e71a99..95f08c1f1e 100644 --- a/akka-actor/src/main/scala/akka/actor/IO.scala +++ b/akka-actor/src/main/scala/akka/actor/IO.scala @@ -258,7 +258,7 @@ class IOManager(bufferSize: Int = 8192) extends Actor { override def preStart: Unit = { worker = new IOWorker(self, bufferSize) - worker.start + worker.start() } def receive = { @@ -307,7 +307,7 @@ private[akka] class IOWorker(ioManager: ActorRef, val bufferSize: Int) { addRequest(request) def start(): Unit = - thread.start + thread.start() // private diff --git a/akka-actor/src/main/scala/akka/actor/Supervisor.scala b/akka-actor/src/main/scala/akka/actor/Supervisor.scala index 3a80f66e37..d214c81e85 100644 --- a/akka-actor/src/main/scala/akka/actor/Supervisor.scala +++ b/akka-actor/src/main/scala/akka/actor/Supervisor.scala @@ -79,7 +79,7 @@ object Supervisor { * *
* val supervisor = factory.newInstance - * supervisor.start // start up all managed servers + * supervisor.start() // start up all managed servers ** * @author Jonas Bonér @@ -91,8 +91,7 @@ case class SupervisorFactory(val config: SupervisorConfig) { def newInstanceFor(config: SupervisorConfig): Supervisor = { val supervisor = new Supervisor(config.restartStrategy, config.maxRestartsHandler) supervisor.configure(config) - supervisor.start - supervisor + supervisor.start() } } @@ -147,7 +146,6 @@ sealed class Supervisor(handler: FaultHandlingStrategy, maxRestartsHandler: (Act case Supervise(actorRef, lifeCycle, registerAsRemoteService) ⇒ // actorRef.lifeCycle = lifeCycle THIS IS NOT COOL, BUT WAITING FOR https://www.assembla.com/spaces/akka/tickets/1124-supervisor-dsl-doesn-t-make-much-sense-after-the-introduction-of-props supervisor.link(actorRef) - actorRef.start() _childActors.add(actorRef) //TODO Why do we keep this here, mem leak? diff --git a/akka-actor/src/main/scala/akka/actor/UntypedActor.scala b/akka-actor/src/main/scala/akka/actor/UntypedActor.scala index 8442df5270..f5e079c071 100644 --- a/akka-actor/src/main/scala/akka/actor/UntypedActor.scala +++ b/akka-actor/src/main/scala/akka/actor/UntypedActor.scala @@ -45,7 +45,6 @@ import akka.japi.{ Creator, Procedure } * * public static void main(String[] args) { * ActorRef actor = Actors.actorOf(SampleUntypedActor.class); - * actor.start(); * actor.tell("SendToSelf"); * actor.stop(); * } @@ -86,7 +85,7 @@ abstract class UntypedActor extends Actor { /** * User overridable callback. * - * Is called when an Actor is started by invoking 'actor.start()'. + * Is called when an Actor is started, this only happens at most once in the life of an actor. */ override def preStart() {} diff --git a/akka-actor/src/main/scala/akka/cluster/RemoteInterface.scala b/akka-actor/src/main/scala/akka/cluster/RemoteInterface.scala index 2e31147424..139545f599 100644 --- a/akka-actor/src/main/scala/akka/cluster/RemoteInterface.scala +++ b/akka-actor/src/main/scala/akka/cluster/RemoteInterface.scala @@ -178,7 +178,7 @@ abstract class RemoteSupport extends ListenerManagement with RemoteServerModule val eventHandler: ActorRef = { implicit object format extends StatelessActorFormat[RemoteEventHandler] val clazz = classOf[RemoteEventHandler] - val handler = Actor.actorOf(Props(clazz).withLocalOnly(true), clazz.getName) + val handler = new LocalActorRef(Props(clazz), clazz.getName, true) // add the remote client and server listener that pipes the events to the event handler system addListener(handler) handler diff --git a/akka-actor/src/main/scala/akka/dispatch/MessageHandling.scala b/akka-actor/src/main/scala/akka/dispatch/MessageHandling.scala index f57d374d4a..164d22275c 100644 --- a/akka-actor/src/main/scala/akka/dispatch/MessageHandling.scala +++ b/akka-actor/src/main/scala/akka/dispatch/MessageHandling.scala @@ -49,7 +49,12 @@ object MessageDispatcher { /** * @author Jonas Bonér */ -abstract class MessageDispatcher { +abstract class MessageDispatcher extends Serializable { + + private def writeObject(out: java.io.ObjectOutputStream) { + (new Exception).printStackTrace() + throw new Exception("Damn you!") + } import MessageDispatcher._ diff --git a/akka-actor/src/main/scala/akka/event/EventHandler.scala b/akka-actor/src/main/scala/akka/event/EventHandler.scala index eb66a8abca..39de30e7d0 100644 --- a/akka-actor/src/main/scala/akka/event/EventHandler.scala +++ b/akka-actor/src/main/scala/akka/event/EventHandler.scala @@ -119,7 +119,7 @@ object EventHandler extends ListenerManagement { defaultListeners foreach { listenerName ⇒ try { ReflectiveAccess.getClassFor[Actor](listenerName) match { - case Right(actorClass) ⇒ addListener(Actor.actorOf(Props(actorClass).withDispatcher(EventHandlerDispatcher))) + case Right(actorClass) ⇒ addListener(new LocalActorRef(Props(actorClass).withDispatcher(EventHandlerDispatcher), newUuid.toString, systemService = true)) case Left(exception) ⇒ throw exception } } catch { diff --git a/akka-actor/src/main/scala/akka/routing/Pool.scala b/akka-actor/src/main/scala/akka/routing/Pool.scala index 053650005b..b407ec120f 100644 --- a/akka-actor/src/main/scala/akka/routing/Pool.scala +++ b/akka-actor/src/main/scala/akka/routing/Pool.scala @@ -117,7 +117,7 @@ trait DefaultActorPool extends ActorPool { this: Actor ⇒ for (i ← 0 until requestedCapacity) yield { val delegate = instance() self link delegate - delegate.start() + delegate } } case qty if qty < 0 ⇒ diff --git a/akka-actor/src/main/scala/akka/routing/Routing.scala b/akka-actor/src/main/scala/akka/routing/Routing.scala index 3e862b0950..532271cf80 100644 --- a/akka-actor/src/main/scala/akka/routing/Routing.scala +++ b/akka-actor/src/main/scala/akka/routing/Routing.scala @@ -219,17 +219,13 @@ object Routing { val clusteringEnabled = ReflectiveAccess.ClusterModule.isEnabled val localOnly = props.localOnly - if (!localOnly && !clusteringEnabled) - throw new IllegalArgumentException("Can't have clustered actor reference without the ClusterModule being enabled") - - else if (clusteringEnabled && !props.localOnly) - ReflectiveAccess.ClusterModule.newClusteredActorRef(props).start() - + if (clusteringEnabled && !props.localOnly) + ReflectiveAccess.ClusterModule.newClusteredActorRef(props) else { if (props.connections.isEmpty) throw new IllegalArgumentException("A routed actorRef can't have an empty connection set") - new RoutedActorRef(props).start() + new RoutedActorRef(props) } } @@ -270,7 +266,7 @@ object Routing { RoutedProps.defaultFailureDetectorFactory, actorAddress, connections, - RoutedProps.defaultTimeout, true)).start() + RoutedProps.defaultTimeout, true)) } } @@ -309,12 +305,6 @@ private[akka] class RoutedActorRef(val routedProps: RoutedProps) extends Abstrac router.init(new RemoveConnectionOnFirstFailureLocalFailureDetector(routedProps.connections)) - def start(): this.type = synchronized[this.type] { - if (_status == ActorRefInternals.UNSTARTED) - _status = ActorRefInternals.RUNNING - this - } - def stop() { synchronized { if (_status == ActorRefInternals.RUNNING) { @@ -323,6 +313,10 @@ private[akka] class RoutedActorRef(val routedProps: RoutedProps) extends Abstrac } } } + + /*If you start me up*/ + if (_status == ActorRefInternals.UNSTARTED) + _status = ActorRefInternals.RUNNING } /** diff --git a/akka-actor/src/main/scala/akka/util/Duration.scala b/akka-actor/src/main/scala/akka/util/Duration.scala index e9dbe62b3e..cdf4347c30 100644 --- a/akka-actor/src/main/scala/akka/util/Duration.scala +++ b/akka-actor/src/main/scala/akka/util/Duration.scala @@ -249,7 +249,7 @@ object Duration { * val d3 = d2 + 1.millisecond * */ -abstract class Duration { +abstract class Duration extends Serializable { def length: Long def unit: TimeUnit def toNanos: Long diff --git a/akka-actor/src/main/scala/akka/util/Helpers.scala b/akka-actor/src/main/scala/akka/util/Helpers.scala index 7f08280698..1436f76580 100644 --- a/akka-actor/src/main/scala/akka/util/Helpers.scala +++ b/akka-actor/src/main/scala/akka/util/Helpers.scala @@ -5,6 +5,7 @@ package akka.util import akka.event.EventHandler +import java.io.{ PrintWriter, StringWriter } /** * @author Jonas Bonér @@ -39,6 +40,11 @@ object Helpers { body } catch { case e: Throwable ⇒ + val sw = new java.io.StringWriter() + var root = e + while (root.getCause ne null) root = e.getCause + root.printStackTrace(new java.io.PrintWriter(sw)) + System.err.println(sw.toString) EventHandler.error(e, this, e.toString) throw e } diff --git a/akka-actor/src/main/scala/akka/util/ListenerManagement.scala b/akka-actor/src/main/scala/akka/util/ListenerManagement.scala index 164157f8e1..9462559ebb 100644 --- a/akka-actor/src/main/scala/akka/util/ListenerManagement.scala +++ b/akka-actor/src/main/scala/akka/util/ListenerManagement.scala @@ -26,7 +26,6 @@ trait ListenerManagement { * The
listener is started by this method if manageLifeCycleOfListeners yields true.
*/
def addListener(listener: ActorRef) {
- if (manageLifeCycleOfListeners) listener.start()
listeners add listener
}
diff --git a/akka-camel-typed/src/main/scala/akka/camel/TypedCamel.scala b/akka-camel-typed/src/main/scala/akka/camel/TypedCamel.scala
index b1014743e6..b82dfefa7d 100644
--- a/akka-camel-typed/src/main/scala/akka/camel/TypedCamel.scala
+++ b/akka-camel-typed/src/main/scala/akka/camel/TypedCamel.scala
@@ -7,8 +7,7 @@ package akka.camel
import org.apache.camel.CamelContext
import akka.actor.Actor._
-import akka.actor.Props
-import akka.actor.ActorRef
+import akka.actor._
import akka.camel.component.TypedActorComponent
/**
@@ -33,8 +32,8 @@ private[camel] object TypedCamel {
* and re-uses the activationTracker of service.
*/
def onCamelServiceStart(service: CamelService) {
- consumerPublisher = actorOf(Props(new TypedConsumerPublisher(service.activationTracker)).withLocalOnly(true))
- publishRequestor = actorOf(Props(new TypedConsumerPublishRequestor).withLocalOnly(true))
+ consumerPublisher = new LocalActorRef(Props(new TypedConsumerPublisher(service.activationTracker)), newUuid.toString, true)
+ publishRequestor = new LocalActorRef(Props(new TypedConsumerPublishRequestor), newUuid.toString, true)
registerPublishRequestor
diff --git a/akka-camel-typed/src/test/scala/akka/camel/TypedConsumerPublishRequestorTest.scala b/akka-camel-typed/src/test/scala/akka/camel/TypedConsumerPublishRequestorTest.scala
index 8d1669f9d8..5d145b0249 100644
--- a/akka-camel-typed/src/test/scala/akka/camel/TypedConsumerPublishRequestorTest.scala
+++ b/akka-camel-typed/src/test/scala/akka/camel/TypedConsumerPublishRequestorTest.scala
@@ -21,13 +21,13 @@ class TypedConsumerPublishRequestorTest extends JUnitSuite {
@Before
def setUp: Unit = {
- publisher = actorOf(new TypedConsumerPublisherMock).start
- requestor = actorOf(new TypedConsumerPublishRequestor).start
+ publisher = actorOf(new TypedConsumerPublisherMock)
+ requestor = actorOf(new TypedConsumerPublishRequestor)
requestor ! InitPublishRequestor(publisher)
consumer = actorOf(new Actor with Consumer {
def endpointUri = "mock:test"
protected def receive = null
- }).start
+ })
}
@After
diff --git a/akka-camel/src/main/scala/akka/camel/CamelContextLifecycle.scala b/akka-camel/src/main/scala/akka/camel/CamelContextLifecycle.scala
index da472da927..468a7c8ce7 100644
--- a/akka-camel/src/main/scala/akka/camel/CamelContextLifecycle.scala
+++ b/akka-camel/src/main/scala/akka/camel/CamelContextLifecycle.scala
@@ -99,8 +99,8 @@ trait CamelContextLifecycle {
c ← context
t ← template
} {
- c.start
- t.start
+ c.start()
+ t.start()
_started = true
EventHandler.info(this, "Camel context started")
}
diff --git a/akka-camel/src/main/scala/akka/camel/CamelService.scala b/akka-camel/src/main/scala/akka/camel/CamelService.scala
index abae389630..7774800d0b 100644
--- a/akka-camel/src/main/scala/akka/camel/CamelService.scala
+++ b/akka-camel/src/main/scala/akka/camel/CamelService.scala
@@ -8,8 +8,7 @@ import java.util.concurrent.TimeUnit
import org.apache.camel.CamelContext
-import akka.actor.Props
-import akka.actor.Actor._
+import akka.actor.{ newUuid, Props, LocalActorRef, Actor }
import akka.config.Config._
import akka.japi.{ SideEffect, Option ⇒ JOption }
import akka.util.Bootable
@@ -27,9 +26,9 @@ import TypedCamelAccess._
* @author Martin Krasser
*/
trait CamelService extends Bootable {
- private[camel] val activationTracker = actorOf(Props(new ActivationTracker).withLocalOnly(true))
- private[camel] val consumerPublisher = actorOf(Props(new ConsumerPublisher(activationTracker)).withLocalOnly(true))
- private[camel] val publishRequestor = actorOf(Props(new ConsumerPublishRequestor).withLocalOnly(true))
+ private[camel] val activationTracker = new LocalActorRef(Props[ActivationTracker], newUuid.toString, true)
+ private[camel] val consumerPublisher = new LocalActorRef(Props(new ConsumerPublisher(activationTracker)), newUuid.toString, true)
+ private[camel] val publishRequestor = new LocalActorRef(Props(new ConsumerPublishRequestor), newUuid.toString, true)
private val serviceEnabled = config.getList("akka.enabled-modules").exists(_ == "camel")
@@ -65,9 +64,6 @@ trait CamelService extends Bootable {
registerPublishRequestor
- activationTracker.start
- consumerPublisher.start
-
// send registration events for all (un)typed actors that have been registered in the past.
for (event ← PublishRequestor.pastActorRegisteredEvents) publishRequestor ! event
@@ -179,10 +175,10 @@ trait CamelService extends Bootable {
(activationTracker ? SetExpectedDeactivationCount(count)).as[CountDownLatch].get
private[camel] def registerPublishRequestor: Unit =
- registry.addListener(publishRequestor)
+ Actor.registry.addListener(publishRequestor)
private[camel] def unregisterPublishRequestor: Unit =
- registry.removeListener(publishRequestor)
+ Actor.registry.removeListener(publishRequestor)
}
/**
diff --git a/akka-camel/src/main/scala/akka/camel/Producer.scala b/akka-camel/src/main/scala/akka/camel/Producer.scala
index b868498cba..972d42cb21 100644
--- a/akka-camel/src/main/scala/akka/camel/Producer.scala
+++ b/akka-camel/src/main/scala/akka/camel/Producer.scala
@@ -173,7 +173,7 @@ trait ProducerSupport { this: Actor ⇒
*/
private def createSendProcessor = {
val sendProcessor = new SendProcessor(endpoint)
- sendProcessor.start
+ sendProcessor.start()
sendProcessor
}
}
diff --git a/akka-camel/src/main/scala/akka/camel/component/ActorComponent.scala b/akka-camel/src/main/scala/akka/camel/component/ActorComponent.scala
index c6f7d29695..891e809423 100644
--- a/akka-camel/src/main/scala/akka/camel/component/ActorComponent.scala
+++ b/akka-camel/src/main/scala/akka/camel/component/ActorComponent.scala
@@ -266,7 +266,8 @@ private[akka] class AsyncCallbackAdapter(exchange: Exchange, callback: AsyncCall
val address = exchange.getExchangeId
def start = {
- _status = ActorRefInternals.RUNNING
+ if (_status == ActorRefInternals.UNSTARTED)
+ _status = ActorRefInternals.RUNNING
this
}
diff --git a/akka-camel/src/test/java/akka/camel/ConsumerJavaTestBase.java b/akka-camel/src/test/java/akka/camel/ConsumerJavaTestBase.java
index 929f72ceb6..9220b47d0c 100644
--- a/akka-camel/src/test/java/akka/camel/ConsumerJavaTestBase.java
+++ b/akka-camel/src/test/java/akka/camel/ConsumerJavaTestBase.java
@@ -32,7 +32,7 @@ public class ConsumerJavaTestBase {
public void shouldHandleExceptionThrownByActorAndGenerateCustomResponse() {
getMandatoryService().awaitEndpointActivation(1, new SideEffect() {
public void apply() {
- actorOf(SampleErrorHandlingConsumer.class).start();
+ actorOf(SampleErrorHandlingConsumer.class);
}
});
String result = getMandatoryTemplate().requestBody("direct:error-handler-test-java", "hello", String.class);
diff --git a/akka-camel/src/test/scala/akka/camel/ConsumerPublishRequestorTest.scala b/akka-camel/src/test/scala/akka/camel/ConsumerPublishRequestorTest.scala
index 5e1baaf305..1fe5728327 100644
--- a/akka-camel/src/test/scala/akka/camel/ConsumerPublishRequestorTest.scala
+++ b/akka-camel/src/test/scala/akka/camel/ConsumerPublishRequestorTest.scala
@@ -18,13 +18,13 @@ class ConsumerPublishRequestorTest extends JUnitSuite {
@Before
def setUp: Unit = {
- publisher = actorOf(new ConsumerPublisherMock).start
- requestor = actorOf(new ConsumerPublishRequestor).start
+ publisher = actorOf(new ConsumerPublisherMock)
+ requestor = actorOf(new ConsumerPublishRequestor)
requestor ! InitPublishRequestor(publisher)
consumer = actorOf(new Actor with Consumer {
def endpointUri = "mock:test"
protected def receive = null
- }).start.asInstanceOf[LocalActorRef]
+ }).asInstanceOf[LocalActorRef]
}
@After
diff --git a/akka-camel/src/test/scala/akka/camel/ConsumerScalaTest.scala b/akka-camel/src/test/scala/akka/camel/ConsumerScalaTest.scala
index 56e2b7e00a..2f1da484f9 100644
--- a/akka-camel/src/test/scala/akka/camel/ConsumerScalaTest.scala
+++ b/akka-camel/src/test/scala/akka/camel/ConsumerScalaTest.scala
@@ -21,17 +21,15 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
import CamelContextManager.mandatoryTemplate
import ConsumerScalaTest._
- var service: CamelService = _
+ var service: CamelService = null
override protected def beforeAll = {
registry.local.shutdownAll
service = CamelServiceFactory.createCamelService
// register test consumer before registering the publish requestor
// and before starting the CamelService (registry is scanned for consumers)
- actorOf(new TestConsumer("direct:publish-test-1")).start
+ actorOf(new TestConsumer("direct:publish-test-1"))
service.registerPublishRequestor
- service.consumerPublisher.start
- service.activationTracker.start
service.awaitEndpointActivation(1) {
service.start
} must be(true)
@@ -43,7 +41,7 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
}
"A responding consumer" when {
- val consumer = actorOf(new TestConsumer("direct:publish-test-2"))
+ var consumer: ActorRef = null
"started before starting the CamelService" must {
"support an in-out message exchange via its endpoint" in {
mandatoryTemplate.requestBody("direct:publish-test-1", "msg1") must equal("received msg1")
@@ -57,7 +55,7 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
"started" must {
"support an in-out message exchange via its endpoint" in {
service.awaitEndpointActivation(1) {
- consumer.start
+ consumer = actorOf(new TestConsumer("direct:publish-test-2"))
} must be(true)
mandatoryTemplate.requestBody("direct:publish-test-2", "msg2") must equal("received msg2")
}
@@ -78,11 +76,11 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
}
"A responding, untyped consumer" when {
- val consumer = Actor.actorOf(classOf[SampleUntypedConsumer])
+ var consumer: ActorRef = null
"started" must {
"support an in-out message exchange via its endpoint" in {
service.awaitEndpointActivation(1) {
- consumer.start
+ consumer = Actor.actorOf(classOf[SampleUntypedConsumer])
} must be(true)
mandatoryTemplate.requestBodyAndHeader("direct:test-untyped-consumer", "x", "test", "y") must equal("x y")
}
@@ -103,7 +101,7 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
"receiving an in-out message exchange" must {
"lead to a TimeoutException" in {
service.awaitEndpointActivation(1) {
- actorOf(Props(creator = () ⇒ new TestBlocker("direct:publish-test-5"), timeout = Timeout(1000))).start
+ actorOf(Props(creator = () ⇒ new TestBlocker("direct:publish-test-5"), timeout = Timeout(1000)))
} must be(true)
try {
@@ -122,7 +120,7 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
"activated with a custom error handler" must {
"handle thrown exceptions by generating a custom response" in {
service.awaitEndpointActivation(1) {
- actorOf[ErrorHandlingConsumer].start
+ actorOf[ErrorHandlingConsumer]
} must be(true)
mandatoryTemplate.requestBody("direct:error-handler-test", "hello") must equal("error: hello")
@@ -131,7 +129,7 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
"activated with a custom redelivery handler" must {
"handle thrown exceptions by redelivering the initial message" in {
service.awaitEndpointActivation(1) {
- actorOf[RedeliveringConsumer].start
+ actorOf[RedeliveringConsumer]
} must be(true)
mandatoryTemplate.requestBody("direct:redelivery-test", "hello") must equal("accepted: hello")
@@ -140,11 +138,13 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
}
"An non auto-acknowledging consumer" when {
- val consumer = actorOf(new TestAckConsumer("direct:application-ack-test"))
"started" must {
"must support acknowledgements on application level" in {
+
+ var consumer: ActorRef = null
+
service.awaitEndpointActivation(1) {
- consumer.start
+ consumer = actorOf(new TestAckConsumer("direct:application-ack-test"))
} must be(true)
val endpoint = mandatoryContext.getEndpoint("direct:application-ack-test", classOf[DirectEndpoint])
@@ -163,13 +163,14 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
producer.process(exchange, handler)
latch.await(5, TimeUnit.SECONDS) must be(true)
+ consumer.stop
}
}
}
"A supervised consumer" must {
"be able to reply during receive" in {
- val consumer = Actor.actorOf(new SupervisedConsumer("reply-channel-test-1")).start
+ val consumer = Actor.actorOf(new SupervisedConsumer("reply-channel-test-1"))
(consumer ? "succeed").get must equal("ok")
}
@@ -181,7 +182,7 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
Supervise(consumer, Permanent) :: Nil))
val latch = new CountDownLatch(1)
- val sender = Actor.actorOf(new Sender("pr", latch)).start
+ val sender = Actor.actorOf(new Sender("pr", latch))
consumer.!("fail")(Some(sender))
latch.await(5, TimeUnit.SECONDS) must be(true)
@@ -195,7 +196,7 @@ class ConsumerScalaTest extends WordSpec with BeforeAndAfterAll with MustMatcher
Supervise(consumer, Temporary) :: Nil))
val latch = new CountDownLatch(1)
- val sender = Actor.actorOf(new Sender("ps", latch)).start
+ val sender = Actor.actorOf(new Sender("ps", latch))
consumer.!("fail")(Some(sender))
latch.await(5, TimeUnit.SECONDS) must be(true)
diff --git a/akka-camel/src/test/scala/akka/camel/ProducerFeatureTest.scala b/akka-camel/src/test/scala/akka/camel/ProducerFeatureTest.scala
index 30dd3f43a3..334e5cd338 100644
--- a/akka-camel/src/test/scala/akka/camel/ProducerFeatureTest.scala
+++ b/akka-camel/src/test/scala/akka/camel/ProducerFeatureTest.scala
@@ -32,7 +32,6 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message and receive normal response") {
given("a registered two-way producer")
val producer = actorOf(new TestProducer("direct:producer-test-2", true))
- producer.start
when("a test message is sent to the producer with ?")
val message = Message("test", Map(Message.MessageExchangeId -> "123"))
@@ -46,7 +45,6 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message and receive failure response") {
given("a registered two-way producer")
val producer = actorOf(new TestProducer("direct:producer-test-2"))
- producer.start
when("a test message causing an exception is sent to the producer with ?")
val message = Message("fail", Map(Message.MessageExchangeId -> "123"))
@@ -62,7 +60,6 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message oneway") {
given("a registered one-way producer")
val producer = actorOf(new TestProducer("direct:producer-test-1", true) with Oneway)
- producer.start
when("a test message is sent to the producer with !")
mockEndpoint.expectedBodiesReceived("TEST")
@@ -75,7 +72,6 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message twoway without sender reference") {
given("a registered two-way producer")
val producer = actorOf(new TestProducer("direct:producer-test-1"))
- producer.start
when("a test message is sent to the producer with !")
mockEndpoint.expectedBodiesReceived("test")
@@ -91,7 +87,6 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message and receive normal response") {
given("a registered two-way producer")
val producer = actorOf(new TestProducer("direct:producer-test-3"))
- producer.start
when("a test message is sent to the producer with ?")
val message = Message("test", Map(Message.MessageExchangeId -> "123"))
@@ -104,7 +99,6 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message and receive failure response") {
given("a registered two-way producer")
val producer = actorOf(new TestProducer("direct:producer-test-3"))
- producer.start
when("a test message causing an exception is sent to the producer with ?")
val message = Message("fail", Map(Message.MessageExchangeId -> "123"))
@@ -122,8 +116,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward normal response to a replying target actor and receive response") {
given("a registered two-way producer configured with a forward target")
- val target = actorOf[ReplyingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-2", target)).start
+ val target = actorOf[ReplyingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-2", target))
when("a test message is sent to the producer with ?")
val message = Message("test", Map(Message.MessageExchangeId -> "123"))
@@ -136,8 +130,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward failure response to a replying target actor and receive response") {
given("a registered two-way producer configured with a forward target")
- val target = actorOf[ReplyingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-2", target)).start
+ val target = actorOf[ReplyingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-2", target))
when("a test message causing an exception is sent to the producer with ?")
val message = Message("fail", Map(Message.MessageExchangeId -> "123"))
@@ -152,8 +146,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward normal response to a producing target actor and produce response to direct:forward-test-1") {
given("a registered one-way producer configured with a forward target")
- val target = actorOf[ProducingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-2", target)).start
+ val target = actorOf[ProducingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-2", target))
when("a test message is sent to the producer with !")
mockEndpoint.expectedBodiesReceived("received test")
@@ -165,8 +159,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward failure response to a producing target actor and produce response to direct:forward-test-1") {
given("a registered one-way producer configured with a forward target")
- val target = actorOf[ProducingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-2", target)).start
+ val target = actorOf[ProducingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-2", target))
when("a test message causing an exception is sent to the producer with !")
mockEndpoint.expectedMessageCount(1)
@@ -182,8 +176,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward normal response to a replying target actor and receive response") {
given("a registered two-way producer configured with a forward target")
- val target = actorOf[ReplyingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-3", target)).start
+ val target = actorOf[ReplyingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-3", target))
when("a test message is sent to the producer with ?")
val message = Message("test", Map(Message.MessageExchangeId -> "123"))
@@ -196,8 +190,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward failure response to a replying target actor and receive response") {
given("a registered two-way producer configured with a forward target")
- val target = actorOf[ReplyingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-3", target)).start
+ val target = actorOf[ReplyingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-3", target))
when("a test message causing an exception is sent to the producer with ?")
val message = Message("fail", Map(Message.MessageExchangeId -> "123"))
@@ -212,8 +206,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward normal response to a producing target actor and produce response to direct:forward-test-1") {
given("a registered one-way producer configured with a forward target")
- val target = actorOf[ProducingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-3", target)).start
+ val target = actorOf[ProducingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-3", target))
when("a test message is sent to the producer with !")
mockEndpoint.expectedBodiesReceived("received test")
@@ -225,8 +219,8 @@ class ProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with Before
scenario("produce message, forward failure response to a producing target actor and produce response to direct:forward-test-1") {
given("a registered one-way producer configured with a forward target")
- val target = actorOf[ProducingForwardTarget].start
- val producer = actorOf(new TestForwarder("direct:producer-test-3", target)).start
+ val target = actorOf[ProducingForwardTarget]
+ val producer = actorOf(new TestForwarder("direct:producer-test-3", target))
when("a test message causing an exception is sent to the producer with !")
mockEndpoint.expectedMessageCount(1)
@@ -279,7 +273,7 @@ object ProducerFeatureTest {
}
class TestRoute extends RouteBuilder {
- val responder = actorOf[TestResponder].start
+ val responder = actorOf[TestResponder]
def configure {
from("direct:forward-test-1").to("mock:mock")
// for one-way messaging tests
diff --git a/akka-camel/src/test/scala/akka/camel/UntypedProducerFeatureTest.scala b/akka-camel/src/test/scala/akka/camel/UntypedProducerFeatureTest.scala
index 4a5dfdca0c..6039ddcc89 100644
--- a/akka-camel/src/test/scala/akka/camel/UntypedProducerFeatureTest.scala
+++ b/akka-camel/src/test/scala/akka/camel/UntypedProducerFeatureTest.scala
@@ -31,7 +31,6 @@ class UntypedProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with
scenario("produce message and receive normal response") {
given("a registered two-way producer")
val producer = actorOf(classOf[SampleUntypedReplyingProducer])
- producer.start
when("a test message is sent to the producer with ?")
val message = Message("test", Map(Message.MessageExchangeId -> "123"))
@@ -45,7 +44,6 @@ class UntypedProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with
scenario("produce message and receive failure response") {
given("a registered two-way producer")
val producer = actorOf(classOf[SampleUntypedReplyingProducer])
- producer.start
when("a test message causing an exception is sent to the producer with ?")
val message = Message("fail", Map(Message.MessageExchangeId -> "123"))
@@ -65,7 +63,6 @@ class UntypedProducerFeatureTest extends FeatureSpec with BeforeAndAfterAll with
scenario("produce message and send normal response to direct:forward-test-1") {
given("a registered one-way producer configured with a forward target")
val producer = actorOf(classOf[SampleUntypedForwardingProducer])
- producer.start
when("a test message is sent to the producer with !")
mockEndpoint.expectedBodiesReceived("received test")
diff --git a/akka-camel/src/test/scala/akka/camel/component/ActorComponentFeatureTest.scala b/akka-camel/src/test/scala/akka/camel/component/ActorComponentFeatureTest.scala
index c837075368..5fe07b2cd4 100644
--- a/akka-camel/src/test/scala/akka/camel/component/ActorComponentFeatureTest.scala
+++ b/akka-camel/src/test/scala/akka/camel/component/ActorComponentFeatureTest.scala
@@ -33,7 +33,7 @@ class ActorComponentFeatureTest extends FeatureSpec with BeforeAndAfterAll with
import CamelContextManager.mandatoryTemplate
scenario("one-way communication") {
- val actor = actorOf[Tester1].start
+ val actor = actorOf[Tester1]
val latch = (actor ? SetExpectedMessageCount(1)).as[CountDownLatch].get
mandatoryTemplate.sendBody("actor:uuid:%s" format actor.uuid, "Martin")
assert(latch.await(5000, TimeUnit.MILLISECONDS))
@@ -42,12 +42,12 @@ class ActorComponentFeatureTest extends FeatureSpec with BeforeAndAfterAll with
}
scenario("two-way communication") {
- val actor = actorOf[Tester2].start
+ val actor = actorOf[Tester2]
assert(mandatoryTemplate.requestBody("actor:uuid:%s" format actor.uuid, "Martin") === "Hello Martin")
}
scenario("two-way communication with timeout") {
- val actor = actorOf(Props[Tester3].withTimeout(Timeout(1))).start
+ val actor = actorOf(Props[Tester3].withTimeout(Timeout(1)))
intercept[RuntimeCamelException] {
mandatoryTemplate.requestBody("actor:uuid:%s?blocking=true" format actor.uuid, "Martin")
}
@@ -70,7 +70,7 @@ class ActorComponentFeatureTest extends FeatureSpec with BeforeAndAfterAll with
import CamelContextManager.mandatoryTemplate
scenario("one-way communication") {
- val actor = actorOf[Tester1].start
+ val actor = actorOf[Tester1]
val latch = (actor ? SetExpectedMessageCount(1)).as[CountDownLatch].get
mandatoryTemplate.sendBody("actor:%s" format actor.address, "Martin")
assert(latch.await(5000, TimeUnit.MILLISECONDS))
@@ -79,12 +79,12 @@ class ActorComponentFeatureTest extends FeatureSpec with BeforeAndAfterAll with
}
scenario("two-way communication") {
- val actor = actorOf[Tester2].start
+ val actor = actorOf[Tester2]
assert(mandatoryTemplate.requestBody("actor:%s" format actor.address, "Martin") === "Hello Martin")
}
scenario("two-way communication via a custom route") {
- val actor = actorOf[CustomIdActor]("custom-id").start
+ val actor = actorOf[CustomIdActor]("custom-id")
assert(mandatoryTemplate.requestBody("direct:custom-id-test-1", "Martin") === "Received Martin")
assert(mandatoryTemplate.requestBody("direct:custom-id-test-2", "Martin") === "Received Martin")
}
@@ -113,8 +113,8 @@ object ActorComponentFeatureTest {
}
class TestRoute extends RouteBuilder {
- val failWithMessage = actorOf[FailWithMessage].start
- val failWithException = actorOf[FailWithException].start
+ val failWithMessage = actorOf[FailWithMessage]
+ val failWithException = actorOf[FailWithException]
def configure {
from("direct:custom-id-test-1").to("actor:custom-id")
from("direct:custom-id-test-2").to("actor:id:custom-id")
diff --git a/akka-camel/src/test/scala/akka/camel/component/ActorProducerTest.scala b/akka-camel/src/test/scala/akka/camel/component/ActorProducerTest.scala
index 29571a2b87..27fa751c63 100644
--- a/akka-camel/src/test/scala/akka/camel/component/ActorProducerTest.scala
+++ b/akka-camel/src/test/scala/akka/camel/component/ActorProducerTest.scala
@@ -23,7 +23,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
@Test
def shouldSendMessageToActorWithSyncProcessor = {
- val actor = actorOf[Tester1].start
+ val actor = actorOf[Tester1]
val latch = (actor ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:uuid:%s" format actor.uuid)
val exchange = endpoint.createExchange(ExchangePattern.InOnly)
@@ -38,7 +38,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
@Test
def shouldSendMessageToActorWithAsyncProcessor = {
- val actor = actorOf[Tester1].start
+ val actor = actorOf[Tester1]
val latch = (actor ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:uuid:%s" format actor.uuid)
val exchange = endpoint.createExchange(ExchangePattern.InOnly)
@@ -55,7 +55,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
def shouldSendMessageToActorAndReceiveResponseWithSyncProcessor = {
val actor = actorOf(new Tester2 {
override def response(msg: Message) = Message(super.response(msg), Map("k2" -> "v2"))
- }).start
+ })
val endpoint = actorEndpoint("actor:uuid:%s" format actor.uuid)
val exchange = endpoint.createExchange(ExchangePattern.InOut)
exchange.getIn.setBody("Martin")
@@ -69,7 +69,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
def shouldSendMessageToActorAndReceiveResponseWithAsyncProcessor = {
val actor = actorOf(new Tester2 {
override def response(msg: Message) = Message(super.response(msg), Map("k2" -> "v2"))
- }).start
+ })
val completion = expectAsyncCompletion
val endpoint = actorEndpoint("actor:uuid:%s" format actor.uuid)
val exchange = endpoint.createExchange(ExchangePattern.InOut)
@@ -85,7 +85,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
def shouldSendMessageToActorAndReceiveFailureWithAsyncProcessor = {
val actor = actorOf(new Tester2 {
override def response(msg: Message) = Failure(new Exception("testmsg"), Map("k3" -> "v3"))
- }).start
+ })
val completion = expectAsyncCompletion
val endpoint = actorEndpoint("actor:uuid:%s" format actor.uuid)
val exchange = endpoint.createExchange(ExchangePattern.InOut)
@@ -102,7 +102,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
def shouldSendMessageToActorAndReceiveAckWithAsyncProcessor = {
val actor = actorOf(new Tester2 {
override def response(msg: Message) = akka.camel.Ack
- }).start
+ })
val completion = expectAsyncCompletion
val endpoint = actorEndpoint("actor:uuid:%s?autoack=false" format actor.uuid)
val exchange = endpoint.createExchange(ExchangePattern.InOnly)
@@ -117,8 +117,8 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
def shouldDynamicallyRouteMessageToActorWithDefaultId = {
val actor1 = actorOf[Tester1]("x")
val actor2 = actorOf[Tester1]("y")
- actor1.start
- actor2.start
+ actor1
+ actor2
val latch1 = (actor1 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val latch2 = (actor2 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:id:%s" format actor1.address)
@@ -141,8 +141,8 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
def shouldDynamicallyRouteMessageToActorWithoutDefaultId = {
val actor1 = actorOf[Tester1]("x")
val actor2 = actorOf[Tester1]("y")
- actor1.start
- actor2.start
+ actor1
+ actor2
val latch1 = (actor1 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val latch2 = (actor2 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:id:")
@@ -164,8 +164,8 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
@Test
def shouldDynamicallyRouteMessageToActorWithDefaultUuid = {
- val actor1 = actorOf[Tester1].start
- val actor2 = actorOf[Tester1].start
+ val actor1 = actorOf[Tester1]
+ val actor2 = actorOf[Tester1]
val latch1 = (actor1 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val latch2 = (actor2 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:uuid:%s" format actor1.uuid)
@@ -186,8 +186,8 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
@Test
def shouldDynamicallyRouteMessageToActorWithoutDefaultUuid = {
- val actor1 = actorOf[Tester1].start
- val actor2 = actorOf[Tester1].start
+ val actor1 = actorOf[Tester1]
+ val actor2 = actorOf[Tester1]
val latch1 = (actor1 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val latch2 = (actor2 ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:uuid:")
@@ -209,7 +209,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
@Test
def shouldThrowExceptionWhenIdNotSet: Unit = {
- val actor = actorOf[Tester1].start
+ val actor = actorOf[Tester1]
val latch = (actor ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:id:")
intercept[ActorIdentifierNotSetException] {
@@ -219,7 +219,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
@Test
def shouldThrowExceptionWhenUuidNotSet: Unit = {
- val actor = actorOf[Tester1].start
+ val actor = actorOf[Tester1]
val latch = (actor ? SetExpectedMessageCount(1)).as[CountDownLatch].get
val endpoint = actorEndpoint("actor:uuid:")
intercept[ActorIdentifierNotSetException] {
@@ -229,7 +229,7 @@ class ActorProducerTest extends JUnitSuite with BeforeAndAfterAll {
@Test
def shouldSendMessageToActorAndTimeout(): Unit = {
- val actor = actorOf(Props[Tester3].withTimeout(Timeout(1))).start
+ val actor = actorOf(Props[Tester3].withTimeout(Timeout(1)))
val endpoint = actorEndpoint("actor:uuid:%s" format actor.uuid)
val exchange = endpoint.createExchange(ExchangePattern.InOut)
exchange.getIn.setBody("Martin")
diff --git a/akka-cluster/src/main/scala/akka/cluster/Cluster.scala b/akka-cluster/src/main/scala/akka/cluster/Cluster.scala
index 2dee6d94da..f7798d5401 100644
--- a/akka-cluster/src/main/scala/akka/cluster/Cluster.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/Cluster.scala
@@ -287,7 +287,7 @@ class DefaultClusterNode private[akka] (
}
}), "akka.cluster.RemoteClientLifeCycleListener")
- private[cluster] lazy val remoteDaemon = actorOf(Props(new RemoteClusterDaemon(this)).copy(dispatcher = new PinnedDispatcher(), localOnly = true), RemoteClusterDaemon.Address)
+ private[cluster] lazy val remoteDaemon = new LocalActorRef(Props(new RemoteClusterDaemon(this)).copy(dispatcher = new PinnedDispatcher()), RemoteClusterDaemon.Address, systemService = true)
private[cluster] lazy val remoteDaemonSupervisor = Supervisor(
SupervisorConfig(
@@ -295,7 +295,7 @@ class DefaultClusterNode private[akka] (
Supervise(
remoteDaemon,
Permanent)
- :: Nil))
+ :: Nil)).start()
lazy val remoteService: RemoteSupport = {
val remote = new akka.cluster.netty.NettyRemoteSupport
@@ -502,7 +502,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, 0, Transient, false, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), 0, Transient, false, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@@ -510,7 +510,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, 0, replicationScheme, false, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), 0, replicationScheme, false, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@@ -518,7 +518,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, replicationFactor, Transient, false, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), replicationFactor, Transient, false, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@@ -526,7 +526,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, replicationFactor, replicationScheme, false, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), replicationFactor, replicationScheme, false, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@@ -534,7 +534,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, 0, Transient, serializeMailbox, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), 0, Transient, serializeMailbox, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@@ -542,7 +542,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, 0, replicationScheme, serializeMailbox, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), 0, replicationScheme, serializeMailbox, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@@ -550,7 +550,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, replicationFactor, Transient, serializeMailbox, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), replicationFactor, Transient, serializeMailbox, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@@ -558,7 +558,7 @@ class DefaultClusterNode private[akka] (
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
- store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress).start, replicationFactor, replicationScheme, serializeMailbox, serializer)
+ store(actorAddress, () ⇒ Actor.actorOf(actorClass, actorAddress), replicationFactor, replicationScheme, serializeMailbox, serializer)
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
@@ -814,7 +814,6 @@ class DefaultClusterNode private[akka] (
// create ADDRESS -> NODE mapping
ignore[ZkNodeExistsException](zkClient.createPersistent(actorAddressToNodesPathFor(actorAddress, nodeName)))
- actorRef.start()
actorRef
}
}
@@ -1318,7 +1317,7 @@ class DefaultClusterNode private[akka] (
EventHandler.debug(this, "Setting up connection to node with nodename [%s] and address [%s]".format(node, address))
val clusterDaemon = remoteService.actorFor(
- RemoteClusterDaemon.Address, address.getHostName, address.getPort).start()
+ RemoteClusterDaemon.Address, address.getHostName, address.getPort)
newConnections = newConnections + (node -> (address, clusterDaemon))
change = true
}
@@ -1784,7 +1783,7 @@ class RemoteClusterDaemon(cluster: ClusterNode) extends Actor {
if (Cluster.shouldCompressData) LZF.uncompress(bytes)
else bytes
- val snapshotActorRef = fromBinary(uncompressedBytes, newActorRef.uuid).start()
+ val snapshotActorRef = fromBinary(uncompressedBytes, newActorRef.uuid)
cluster.remoteService.register(actorAddress, snapshotActorRef)
// FIXME we should call 'stop()' here (to GC the actor), but can't since that will currently
@@ -1864,25 +1863,35 @@ class RemoteClusterDaemon(cluster: ClusterNode) extends Actor {
}
def handle_fun0_unit(message: RemoteProtocol.RemoteDaemonMessageProtocol) {
- actorOf(Props(
- self ⇒ { case f: Function0[_] ⇒ try { f() } finally { self.stop() } }).copy(dispatcher = computeGridDispatcher, localOnly = true)) ! payloadFor(message, classOf[Function0[Unit]])
+ new LocalActorRef(
+ Props(
+ self ⇒ {
+ case f: Function0[_] ⇒ try { f() } finally { self.stop() }
+ }).copy(dispatcher = computeGridDispatcher), newUuid.toString, systemService = true) ! payloadFor(message, classOf[Function0[Unit]])
}
def handle_fun0_any(message: RemoteProtocol.RemoteDaemonMessageProtocol) {
- actorOf(Props(
- self ⇒ { case f: Function0[_] ⇒ try { self.reply(f()) } finally { self.stop() } }).copy(dispatcher = computeGridDispatcher, localOnly = true)) forward payloadFor(message, classOf[Function0[Any]])
+ new LocalActorRef(
+ Props(
+ self ⇒ {
+ case f: Function0[_] ⇒ try { self.reply(f()) } finally { self.stop() }
+ }).copy(dispatcher = computeGridDispatcher), newUuid.toString, systemService = true) forward payloadFor(message, classOf[Function0[Any]])
}
def handle_fun1_arg_unit(message: RemoteProtocol.RemoteDaemonMessageProtocol) {
- actorOf(Props(
- self ⇒ { case (fun: Function[_, _], param: Any) ⇒ try { fun.asInstanceOf[Any ⇒ Unit].apply(param) } finally { self.stop() } }).copy(dispatcher = computeGridDispatcher, localOnly = true)) ! payloadFor(message, classOf[Tuple2[Function1[Any, Unit], Any]])
+ new LocalActorRef(
+ Props(
+ self ⇒ {
+ case (fun: Function[_, _], param: Any) ⇒ try { fun.asInstanceOf[Any ⇒ Unit].apply(param) } finally { self.stop() }
+ }).copy(dispatcher = computeGridDispatcher), newUuid.toString, systemService = true) ! payloadFor(message, classOf[Tuple2[Function1[Any, Unit], Any]])
}
def handle_fun1_arg_any(message: RemoteProtocol.RemoteDaemonMessageProtocol) {
- actorOf(Props(
- self ⇒ {
- case (fun: Function[_, _], param: Any) ⇒ try { self.reply(fun.asInstanceOf[Any ⇒ Any](param)) } finally { self.stop() }
- }).copy(dispatcher = computeGridDispatcher, localOnly = true)) forward payloadFor(message, classOf[Tuple2[Function1[Any, Any], Any]])
+ new LocalActorRef(
+ Props(
+ self ⇒ {
+ case (fun: Function[_, _], param: Any) ⇒ try { self.reply(fun.asInstanceOf[Any ⇒ Any](param)) } finally { self.stop() }
+ }).copy(dispatcher = computeGridDispatcher), newUuid.toString, systemService = true) forward payloadFor(message, classOf[Tuple2[Function1[Any, Any], Any]])
}
def handleFailover(message: RemoteProtocol.RemoteDaemonMessageProtocol) {
diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterActorRef.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterActorRef.scala
index 626ccea760..d677f7c663 100644
--- a/akka-cluster/src/main/scala/akka/cluster/ClusterActorRef.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/ClusterActorRef.scala
@@ -60,7 +60,7 @@ object ClusterActorRef {
.withDeployId(actorAddress)
.withTimeout(timeout)
.withRouter(routerFactory)
- .withFailureDetector(failureDetectorFactory)).start()
+ .withFailureDetector(failureDetectorFactory))
}
/**
@@ -112,14 +112,6 @@ private[akka] class ClusterActorRef(props: RoutedProps) extends AbstractRoutedAc
connections.failOver(from, to)
}
- def start(): this.type = synchronized[this.type] {
- if (_status == ActorRefInternals.UNSTARTED) {
- _status = ActorRefInternals.RUNNING
- Actor.registry.local.registerClusterActorRef(this)
- }
- this
- }
-
def stop() {
synchronized {
if (_status == ActorRefInternals.RUNNING) {
@@ -132,4 +124,10 @@ private[akka] class ClusterActorRef(props: RoutedProps) extends AbstractRoutedAc
}
}
}
+
+ /* If you start me up */
+ if (_status == ActorRefInternals.UNSTARTED) {
+ _status = ActorRefInternals.RUNNING
+ Actor.registry.local.registerClusterActorRef(this)
+ }
}
diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterDeployer.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterDeployer.scala
index 3fd94c6bb3..cd2ee02b8e 100644
--- a/akka-cluster/src/main/scala/akka/cluster/ClusterDeployer.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/ClusterDeployer.scala
@@ -168,7 +168,7 @@ object ClusterDeployer extends ActorDeployer {
case Deploy(_, _, _, _, Local) | Deploy(_, _, _, _, _: Local) ⇒ //TODO LocalDeployer.deploy(deployment)??
case Deploy(address, recipe, routing, _, _) ⇒ // cluster deployment
/*TODO recipe foreach { r ⇒
- Deployer.newClusterActorRef(() ⇒ Actor.actorOf(r.implementationClass), address, deployment).start()
+ Deployer.newClusterActorRef(() ⇒ Actor.actorOf(r.implementationClass), address, deployment)
}*/
val path = deploymentAddressPath.format(address)
try {
diff --git a/akka-cluster/src/main/scala/akka/cluster/LocalCluster.scala b/akka-cluster/src/main/scala/akka/cluster/LocalCluster.scala
index e32d51c7cf..ff78d05796 100644
--- a/akka-cluster/src/main/scala/akka/cluster/LocalCluster.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/LocalCluster.scala
@@ -88,9 +88,8 @@ object LocalCluster {
*/
def shutdownLocalCluster() {
withPrintStackTraceOnError {
- EventHandler.info(this, "Shuts down local cluster")
- zkServer.get.foreach(_.shutdown())
- zkServer.set(None)
+ EventHandler.debug(this, "Shuts down local cluster")
+ zkServer.getAndSet(None).foreach(_.shutdown())
}
}
diff --git a/akka-cluster/src/main/scala/akka/cluster/RemoteFailureDetector.scala b/akka-cluster/src/main/scala/akka/cluster/RemoteFailureDetector.scala
index 2965460e91..48fd1f6da0 100644
--- a/akka-cluster/src/main/scala/akka/cluster/RemoteFailureDetector.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/RemoteFailureDetector.scala
@@ -4,8 +4,8 @@
package akka.cluster
-import akka.actor.{ Actor, ActorRef, Props }
-import Actor._
+import akka.actor.{ newUuid, Actor, ActorRef, Props, LocalActorRef }
+import akka.actor.Actor._
import akka.cluster._
import akka.routing._
import akka.event.EventHandler
@@ -24,7 +24,7 @@ object RemoteFailureDetector {
private case class Register(strategy: RemoteFailureListener, address: InetSocketAddress) extends FailureDetectorEvent
private case class Unregister(strategy: RemoteFailureListener, address: InetSocketAddress) extends FailureDetectorEvent
- private[akka] val registry = actorOf(Props(new Registry).copy(dispatcher = new PinnedDispatcher(), localOnly = true))
+ private[akka] val registry = new LocalActorRef(Props[Registry].copy(dispatcher = new PinnedDispatcher()), newUuid.toString, systemService = true)
def register(strategy: RemoteFailureListener, address: InetSocketAddress) = registry ! Register(strategy, address)
diff --git a/akka-cluster/src/main/scala/akka/cluster/netty/NettyRemoteSupport.scala b/akka-cluster/src/main/scala/akka/cluster/netty/NettyRemoteSupport.scala
index 451c43f17f..614f1c60dc 100644
--- a/akka-cluster/src/main/scala/akka/cluster/netty/NettyRemoteSupport.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/netty/NettyRemoteSupport.scala
@@ -769,10 +769,8 @@ trait NettyRemoteServerModule extends RemoteServerModule {
}
private def register[Key](id: Key, actorRef: ActorRef, registry: ConcurrentHashMap[Key, ActorRef]) {
- if (_isRunning.isOn) {
+ if (_isRunning.isOn)
registry.put(id, actorRef) //TODO change to putIfAbsent
- if (!actorRef.isRunning) actorRef.start()
- }
}
/**
@@ -1078,7 +1076,7 @@ class RemoteServerHandler(
"Looking up a remotely available actor for address [%s] on node [%s]"
.format(address, Config.nodename))
- val actorRef = Actor.createActor(address, () ⇒ createSessionActor(actorInfo, channel))
+ val actorRef = Actor.createActor(address, () ⇒ createSessionActor(actorInfo, channel), false)
if (actorRef eq null) throw new IllegalActorStateException("Could not find a remote actor with address [" + address + "] or uuid [" + uuid + "]")
@@ -1098,9 +1096,8 @@ class RemoteServerHandler(
case null ⇒ null
case factory ⇒
val actorRef = factory()
- actorRef.uuid = parseUuid(uuid) //FIXME is this sensible?
sessionActors.get(channel).put(address, actorRef)
- actorRef.start() //Start it where's it's created
+ actorRef //Start it where's it's created
}
case sessionActor ⇒ sessionActor
}
diff --git a/akka-cluster/src/main/scala/akka/serialization/SerializationProtocol.scala b/akka-cluster/src/main/scala/akka/serialization/SerializationProtocol.scala
index b7be90097c..7910ba768e 100644
--- a/akka-cluster/src/main/scala/akka/serialization/SerializationProtocol.scala
+++ b/akka-cluster/src/main/scala/akka/serialization/SerializationProtocol.scala
@@ -230,7 +230,7 @@ object ActorSerialization {
//
)
- val receiveTimeout = if (protocol.hasReceiveTimeout) Some(protocol.getReceiveTimeout) else None //TODO FIXME, I'm expensive and slow
+ val receiveTimeout = if (protocol.hasReceiveTimeout) Some(protocol.getReceiveTimeout) else None
val ar = new LocalActorRef(actorUuid, protocol.getAddress, props, receiveTimeout, storedHotswap)
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/nosnapshot/ReplicationTransactionLogWriteBehindNoSnapshotMultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/nosnapshot/ReplicationTransactionLogWriteBehindNoSnapshotMultiJvmSpec.scala
index 1cdab61e23..8c3054cf81 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/nosnapshot/ReplicationTransactionLogWriteBehindNoSnapshotMultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/nosnapshot/ReplicationTransactionLogWriteBehindNoSnapshotMultiJvmSpec.scala
@@ -42,7 +42,7 @@
// }
// barrier("create-actor-on-node1", NrOfNodes) {
-// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-behind-nosnapshot").start()
+// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-behind-nosnapshot")
// // node.isInUseOnNode("hello-world") must be(true)
// actorRef.address must be("hello-world-write-behind-nosnapshot")
// for (i ← 0 until 10) {
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/snapshot/ReplicationTransactionLogWriteBehindSnapshotMultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/snapshot/ReplicationTransactionLogWriteBehindSnapshotMultiJvmSpec.scala
index fff2897702..4415fd7c84 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/snapshot/ReplicationTransactionLogWriteBehindSnapshotMultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writebehind/snapshot/ReplicationTransactionLogWriteBehindSnapshotMultiJvmSpec.scala
@@ -44,7 +44,7 @@
// }
// barrier("create-actor-on-node1", NrOfNodes) {
-// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-behind-snapshot").start()
+// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-behind-snapshot")
// node.isInUseOnNode("hello-world-write-behind-snapshot") must be(true)
// actorRef.address must be("hello-world-write-behind-snapshot")
// var counter = 0
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/nosnapshot/ReplicationTransactionLogWriteThroughNoSnapshotMultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/nosnapshot/ReplicationTransactionLogWriteThroughNoSnapshotMultiJvmSpec.scala
index 69e5dac00d..ca925d28b3 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/nosnapshot/ReplicationTransactionLogWriteThroughNoSnapshotMultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/nosnapshot/ReplicationTransactionLogWriteThroughNoSnapshotMultiJvmSpec.scala
@@ -44,7 +44,7 @@
// }
// barrier("create-actor-on-node1", NrOfNodes) {
-// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-through-nosnapshot").start()
+// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-through-nosnapshot")
// actorRef.address must be("hello-world-write-through-nosnapshot")
// for (i ← 0 until 10)
// (actorRef ? Count(i)).as[String] must be(Some("World from node [node1]"))
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/snapshot/ReplicationTransactionLogWriteThroughSnapshotMultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/snapshot/ReplicationTransactionLogWriteThroughSnapshotMultiJvmSpec.scala
index ba514c4dac..44360ae359 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/snapshot/ReplicationTransactionLogWriteThroughSnapshotMultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/replication/transactionlog/writethrough/snapshot/ReplicationTransactionLogWriteThroughSnapshotMultiJvmSpec.scala
@@ -42,7 +42,7 @@
// }
// barrier("create-actor-on-node1", NrOfNodes) {
-// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-through-snapshot").start()
+// val actorRef = Actor.actorOf[HelloWorld]("hello-world-write-through-snapshot")
// node.isInUseOnNode("hello-world-write-through-snapshot") must be(true)
// actorRef.address must be("hello-world-write-through-snapshot")
// var counter = 0
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/homenode/HomeNode1MultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/homenode/HomeNode1MultiJvmSpec.scala
index 848b6cb959..0467a0c7a6 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/homenode/HomeNode1MultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/homenode/HomeNode1MultiJvmSpec.scala
@@ -45,11 +45,11 @@ class HomeNodeMultiJvmNode2 extends ClusterTestNode {
Cluster.node.start()
barrier("waiting-for-begin", NrOfNodes).await()
- val actorNode1 = Actor.actorOf[SomeActor]("service-node1").start()
+ val actorNode1 = Actor.actorOf[SomeActor]("service-node1")
val name1 = (actorNode1 ? "identify").get.asInstanceOf[String]
name1 must equal("node1")
- val actorNode2 = Actor.actorOf[SomeActor]("service-node2").start()
+ val actorNode2 = Actor.actorOf[SomeActor]("service-node2")
val name2 = (actorNode2 ? "identify").get.asInstanceOf[String]
name2 must equal("node2")
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/normalusage/SingleReplicaDirectRoutingMultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/normalusage/SingleReplicaDirectRoutingMultiJvmSpec.scala
index 92b0551775..cc928f999a 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/normalusage/SingleReplicaDirectRoutingMultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/direct/normalusage/SingleReplicaDirectRoutingMultiJvmSpec.scala
@@ -48,7 +48,7 @@ class SingleReplicaDirectRoutingMultiJvmNode2 extends ClusterTestNode {
Cluster.node.start()
LocalCluster.barrier("waiting-for-begin", NrOfNodes).await()
- val actor = Actor.actorOf[SomeActor]("service-hello").start().asInstanceOf[ClusterActorRef]
+ val actor = Actor.actorOf[SomeActor]("service-hello").asInstanceOf[ClusterActorRef]
actor.isRunning must be(true)
val result = (actor ? "identify").get
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/roundrobin/homenode/HomeNodeMultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/roundrobin/homenode/HomeNodeMultiJvmSpec.scala
index 3387b64c16..5ffc7348b4 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/roundrobin/homenode/HomeNodeMultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/roundrobin/homenode/HomeNodeMultiJvmSpec.scala
@@ -48,11 +48,11 @@ class HomeNodeMultiJvmNode2 extends ClusterTestNode {
Cluster.node.start()
barrier("waiting-for-begin", NrOfNodes).await()
- val actorNode1 = Actor.actorOf[SomeActor]("service-node1").start()
+ val actorNode1 = Actor.actorOf[SomeActor]("service-node1")
val name1 = (actorNode1 ? "identify").get.asInstanceOf[String]
name1 must equal("node1")
- val actorNode2 = Actor.actorOf[SomeActor]("service-node2").start()
+ val actorNode2 = Actor.actorOf[SomeActor]("service-node2")
val name2 = (actorNode2 ? "identify").get.asInstanceOf[String]
name2 must equal("node2")
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/scattergather/failover/ScatterGatherFailoverMultiJvmSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/scattergather/failover/ScatterGatherFailoverMultiJvmSpec.scala
index ea4eae2525..5e0c566b4d 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/scattergather/failover/ScatterGatherFailoverMultiJvmSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/routing/scattergather/failover/ScatterGatherFailoverMultiJvmSpec.scala
@@ -26,7 +26,7 @@ object ScatterGatherFailoverMultiJvmSpec {
Thread.sleep(2000)
Cluster.node.shutdown()
}
- }.start()
+ }
def receive = {
case Shutdown(None) ⇒ shutdownNode
diff --git a/akka-cluster/src/test/scala/akka/serialization/ActorSerializeSpec.scala b/akka-cluster/src/test/scala/akka/serialization/ActorSerializeSpec.scala
index 9d0ff3766d..f7dc33b856 100644
--- a/akka-cluster/src/test/scala/akka/serialization/ActorSerializeSpec.scala
+++ b/akka-cluster/src/test/scala/akka/serialization/ActorSerializeSpec.scala
@@ -21,12 +21,12 @@ class ActorSerializeSpec extends Spec with ShouldMatchers with BeforeAndAfterAll
describe("Serializable actor") {
it("should be able to serialize and de-serialize a stateful actor with a given serializer") {
- val actor1 = actorOf(Props[MyJavaSerializableActor].withLocalOnly(true)).asInstanceOf[LocalActorRef]
+ val actor1 = new LocalActorRef(Props[MyJavaSerializableActor], newUuid.toString, true)
(actor1 ? "hello").get should equal("world 1")
(actor1 ? "hello").get should equal("world 2")
val bytes = toBinary(actor1)
- val actor2 = fromBinary(bytes).start().asInstanceOf[LocalActorRef]
+ val actor2 = fromBinary(bytes).asInstanceOf[LocalActorRef]
(actor2 ? "hello").get should equal("world 3")
actor2.receiveTimeout should equal(Some(1000))
@@ -36,7 +36,7 @@ class ActorSerializeSpec extends Spec with ShouldMatchers with BeforeAndAfterAll
it("should be able to serialize and deserialize a MyStatelessActorWithMessagesInMailbox") {
- val actor1 = actorOf(Props[MyStatelessActorWithMessagesInMailbox].withLocalOnly(true)).asInstanceOf[LocalActorRef]
+ val actor1 = new LocalActorRef(Props[MyStatelessActorWithMessagesInMailbox], newUuid.toString, true)
for (i ← 1 to 10) actor1 ! "hello"
actor1.getDispatcher.mailboxSize(actor1) should be > (0)
@@ -54,17 +54,8 @@ class ActorSerializeSpec extends Spec with ShouldMatchers with BeforeAndAfterAll
it("should be able to serialize and deserialize a PersonActorWithMessagesInMailbox") {
val p1 = Person("debasish ghosh", 25, SerializeSpec.Address("120", "Monroe Street", "Santa Clara", "95050"))
- val actor1 = actorOf(Props[PersonActorWithMessagesInMailbox].withLocalOnly(true)).asInstanceOf[LocalActorRef]
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
- (actor1 ! p1)
+ val actor1 = new LocalActorRef(Props[PersonActorWithMessagesInMailbox], newUuid.toString, true)
+ for (i ← 1 to 10) actor1 ! p1
actor1.getDispatcher.mailboxSize(actor1) should be > (0)
val actor2 = fromBinary(toBinary(actor1)).asInstanceOf[LocalActorRef]
Thread.sleep(1000)
@@ -98,7 +89,7 @@ class ActorSerializeSpec extends Spec with ShouldMatchers with BeforeAndAfterAll
describe("serialize actor that accepts protobuf message") {
it("should serialize") {
- val actor1 = actorOf(Props[MyActorWithProtobufMessagesInMailbox].withLocalOnly(true)).asInstanceOf[LocalActorRef]
+ val actor1 = new LocalActorRef(Props[MyActorWithProtobufMessagesInMailbox], newUuid.toString, true)
val msg = MyMessage(123, "debasish ghosh", true)
val b = ProtobufProtocol.MyMessage.newBuilder.setId(msg.id).setName(msg.name).setStatus(msg.status).build
for (i ← 1 to 10) actor1 ! b
diff --git a/akka-docs/disabled/examples/Pi.scala b/akka-docs/disabled/examples/Pi.scala
index 16c781bd4d..b70eb20f37 100644
--- a/akka-docs/disabled/examples/Pi.scala
+++ b/akka-docs/disabled/examples/Pi.scala
@@ -62,7 +62,7 @@ object Pi extends App {
//#create-workers
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
val router = Routing.actorOf(
@@ -72,7 +72,7 @@ object Pi extends App {
.withDeployId("pi")
)
- loadBalancerActor(CyclicIterator(workers)).start()
+ loadBalancerActor(CyclicIterator(workers))
//#create-workers
//#master-receive
@@ -123,7 +123,7 @@ object Pi extends App {
// create the master
val master = actorOf(
- new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch)).start()
+ new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
// start the calculation
master ! Calculate
diff --git a/akka-docs/intro/getting-started-first-java.rst b/akka-docs/intro/getting-started-first-java.rst
index da606c7d84..be79581416 100644
--- a/akka-docs/intro/getting-started-first-java.rst
+++ b/akka-docs/intro/getting-started-first-java.rst
@@ -338,7 +338,7 @@ The master actor is a little bit more involved. In its constructor we need to cr
// create the workers
final ActorRef[] workers = new ActorRef[nrOfWorkers];
for (int i = 0; i < nrOfWorkers; i++) {
- workers[i] = actorOf(Worker.class).start();
+ workers[i] = actorOf(Worker.class);
}
// wrap them with a load-balancing router
@@ -346,7 +346,7 @@ The master actor is a little bit more involved. In its constructor we need to cr
public UntypedActor create() {
return new PiRouter(workers);
}
- }).start();
+ });
}
}
@@ -360,8 +360,7 @@ One thing to note is that we used two different versions of the ``actorOf`` meth
The actor's life-cycle is:
-- Created -- ``Actor.actorOf[MyActor]`` -- can **not** receive messages
-- Started -- ``actorRef.start()`` -- can receive messages
+- Created & Started -- ``Actor.actorOf[MyActor]`` -- can receive messages
- Stopped -- ``actorRef.stop()`` -- can **not** receive messages
Once the actor has been stopped it is dead and can not be started again.
@@ -406,7 +405,7 @@ Here is the master actor::
// create the workers
final ActorRef[] workers = new ActorRef[nrOfWorkers];
for (int i = 0; i < nrOfWorkers; i++) {
- workers[i] = actorOf(Worker.class).start();
+ workers[i] = actorOf(Worker.class);
}
// wrap them with a load-balancing router
@@ -414,7 +413,7 @@ Here is the master actor::
public UntypedActor create() {
return new PiRouter(workers);
}
- }).start();
+ });
}
// message handler
@@ -501,7 +500,7 @@ Now the only thing that is left to implement is the runner that should bootstrap
public UntypedActor create() {
return new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch);
}
- }).start();
+ });
// start the calculation
master.tell(new Calculate());
@@ -631,7 +630,7 @@ Before we package it up and run it, let's take a look at the full code now, with
// create the workers
final ActorRef[] workers = new ActorRef[nrOfWorkers];
for (int i = 0; i < nrOfWorkers; i++) {
- workers[i] = actorOf(Worker.class).start();
+ workers[i] = actorOf(Worker.class);
}
// wrap them with a load-balancing router
@@ -639,7 +638,7 @@ Before we package it up and run it, let's take a look at the full code now, with
public UntypedActor create() {
return new PiRouter(workers);
}
- }).start();
+ });
}
// message handler
@@ -697,7 +696,7 @@ Before we package it up and run it, let's take a look at the full code now, with
public UntypedActor create() {
return new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch);
}
- }).start();
+ });
// start the calculation
master.tell(new Calculate());
diff --git a/akka-docs/intro/getting-started-first-scala-eclipse.rst b/akka-docs/intro/getting-started-first-scala-eclipse.rst
index ebe3785828..d764e0cdae 100644
--- a/akka-docs/intro/getting-started-first-scala-eclipse.rst
+++ b/akka-docs/intro/getting-started-first-scala-eclipse.rst
@@ -254,10 +254,10 @@ Now create a new class for the master actor. The master actor is a little bit mo
and then we can create the workers::
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
- val router = Routing.loadBalancerActor(CyclicIterator(workers)).start()
+ val router = Routing.loadBalancerActor(CyclicIterator(workers))
As you can see we are using the ``actorOf`` factory method to create actors, this method returns as an ``ActorRef`` which is a reference to our newly created actor. This method is available in the ``Actor`` object but is usually imported::
@@ -268,7 +268,7 @@ There are two versions of ``actorOf``; one of them taking a actor type and the o
The actor's life-cycle is:
- Created -- ``Actor.actorOf[MyActor]`` -- can **not** receive messages
-- Started -- ``actorRef.start()`` -- can receive messages
+- Started -- ``actorRef`` -- can receive messages
- Stopped -- ``actorRef.stop()`` -- can **not** receive messages
Once the actor has been stopped it is dead and can not be started again.
@@ -290,10 +290,10 @@ Here is the master actor::
var start: Long = _
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
- val router = Routing.loadBalancerActor(CyclicIterator(workers)).start()
+ val router = Routing.loadBalancerActor(CyclicIterator(workers))
def receive = { ... }
@@ -366,7 +366,7 @@ The ``Pi`` object is a perfect container module for our actors and messages, so
// create the master
val master = actorOf(
- new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch)).start()
+ new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
// start the calculation
master ! Calculate
diff --git a/akka-docs/intro/getting-started-first-scala.rst b/akka-docs/intro/getting-started-first-scala.rst
index 71591a3134..ea57e0c57e 100644
--- a/akka-docs/intro/getting-started-first-scala.rst
+++ b/akka-docs/intro/getting-started-first-scala.rst
@@ -250,10 +250,10 @@ Creating the master
The master actor is a little bit more involved. In its constructor we need to create the workers (the ``Worker`` actors) and start them. We will also wrap them in a load-balancing router to make it easier to spread out the work evenly between the workers. Let's do that first::
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
- val router = Routing.loadBalancerActor(CyclicIterator(workers)).start()
+ val router = Routing.loadBalancerActor(CyclicIterator(workers))
As you can see we are using the ``actorOf`` factory method to create actors, this method returns as an ``ActorRef`` which is a reference to our newly created actor. This method is available in the ``Actor`` object but is usually imported::
@@ -263,8 +263,7 @@ There are two versions of ``actorOf``; one of them taking a actor type and the o
The actor's life-cycle is:
-- Created -- ``Actor.actorOf[MyActor]`` -- can **not** receive messages
-- Started -- ``actorRef.start()`` -- can receive messages
+- Created & Started -- ``Actor.actorOf[MyActor]`` -- can receive messages
- Stopped -- ``actorRef.stop()`` -- can **not** receive messages
Once the actor has been stopped it is dead and can not be started again.
@@ -286,10 +285,10 @@ Here is the master actor::
var start: Long = _
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
- val router = Routing.loadBalancerActor(CyclicIterator(workers)).start()
+ val router = Routing.loadBalancerActor(CyclicIterator(workers))
def receive = { ... }
@@ -362,7 +361,7 @@ The ``Pi`` object is a perfect container module for our actors and messages, so
// create the master
val master = actorOf(
- new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch)).start()
+ new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
// start the calculation
master ! Calculate
@@ -428,10 +427,10 @@ But before we package it up and run it, let's take a look at the full code now,
var start: Long = _
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
- val router = Routing.loadBalancerActor(CyclicIterator(workers)).start()
+ val router = Routing.loadBalancerActor(CyclicIterator(workers))
// message handler
def receive = {
@@ -476,7 +475,7 @@ But before we package it up and run it, let's take a look at the full code now,
// create the master
val master = actorOf(
- new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch)).start()
+ new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
// start the calculation
master ! Calculate
diff --git a/akka-docs/java/actor-registry.rst b/akka-docs/java/actor-registry.rst
index 41d50c05c3..4439c37468 100644
--- a/akka-docs/java/actor-registry.rst
+++ b/akka-docs/java/actor-registry.rst
@@ -77,5 +77,5 @@ The above actor can be added as listener of registry events:
import static akka.actor.Actors.*;
- ActorRef listener = actorOf(RegistryListener.class).start();
+ ActorRef listener = actorOf(RegistryListener.class);
registry().addListener(listener);
diff --git a/akka-docs/java/dispatchers.rst b/akka-docs/java/dispatchers.rst
index 3af53cfa6b..e70776840b 100644
--- a/akka-docs/java/dispatchers.rst
+++ b/akka-docs/java/dispatchers.rst
@@ -174,7 +174,6 @@ Creating a PriorityDispatcher using PriorityGenerator:
// We create a new Priority dispatcher and seed it with the priority generator
ref.setDispatcher(new PriorityDispatcher("foo", gen));
- ref.start(); // Start the actor
ref.getDispatcher().suspend(ref); // Suspending the actor so it doesn't start to treat the messages before we have enqueued all of them :-)
ref.tell("lowpriority");
ref.tell("lowpriority");
diff --git a/akka-docs/java/fault-tolerance.rst b/akka-docs/java/fault-tolerance.rst
index 7ad7553031..23a3b39bb2 100644
--- a/akka-docs/java/fault-tolerance.rst
+++ b/akka-docs/java/fault-tolerance.rst
@@ -233,7 +233,7 @@ Here is the API and how to use it from within an 'Actor':
getContext().unlink(actorRef);
// starts and links Actors atomically
- getContext().link(actorRef).start();
+ getContext().link(actorRef);
// spawns (creates and starts) actors
getContext().spawn(MyActor.class);
diff --git a/akka-docs/java/remote-actors.rst b/akka-docs/java/remote-actors.rst
index 311bc78803..3739300938 100644
--- a/akka-docs/java/remote-actors.rst
+++ b/akka-docs/java/remote-actors.rst
@@ -38,7 +38,7 @@ Here is how to start up the server and specify the hostname and port in the ‘a
import static akka.actor.Actors.*;
- remote().start();
+ remote();
// Specify the classloader to use to load the remote class (actor)
remote().start(classLoader);
diff --git a/akka-docs/java/routing.rst b/akka-docs/java/routing.rst
index 56b549a24a..cb74f2ee01 100644
--- a/akka-docs/java/routing.rst
+++ b/akka-docs/java/routing.rst
@@ -27,8 +27,8 @@ An UntypedDispatcher is an actor that routes incoming messages to outbound actor
}
public class MyRouter extends UntypedRouter {
- private ActorRef pinger = actorOf(Pinger.class).start();
- private ActorRef ponger = actorOf(Ponger.class).start();
+ private ActorRef pinger = actorOf(Pinger.class);
+ private ActorRef ponger = actorOf(Ponger.class);
//Route Ping-messages to the pinger, and Pong-messages to the ponger
public ActorRef route(Object message) {
@@ -38,7 +38,7 @@ An UntypedDispatcher is an actor that routes incoming messages to outbound actor
}
}
- ActorRef router = actorOf(MyRouter.class).start();
+ ActorRef router = actorOf(MyRouter.class);
router.tell("Ping"); //Prints "Pinger: Ping"
router.tell("Pong"); //Prints "Ponger: Pong"
@@ -71,8 +71,8 @@ An UntypedLoadBalancer is an actor that forwards messages it receives to a bound
//Our load balancer, sends messages to a pinger, then a ponger, rinse and repeat.
public class MyLoadBalancer extends UntypedLoadBalancer {
private InfiniteIterator
- * val chatService = Actor.actorOf[ChatService].start()
+ * val chatService = Actor.actorOf[ChatService]
*
*/
class ChatService extends
@@ -554,7 +552,7 @@ Run a sample chat session
import sample.chat._
import akka.actor.Actor._
- val chatService = actorOf[ChatService].start()
+ val chatService = actorOf[ChatService]
3. In the second REPL you get execute:
diff --git a/akka-durable-mailboxes/akka-mailboxes-common/src/test/scala/akka/actor/mailbox/DurableMailboxSpec.scala b/akka-durable-mailboxes/akka-mailboxes-common/src/test/scala/akka/actor/mailbox/DurableMailboxSpec.scala
index 2709c220c4..9a0342d7c8 100644
--- a/akka-durable-mailboxes/akka-mailboxes-common/src/test/scala/akka/actor/mailbox/DurableMailboxSpec.scala
+++ b/akka-durable-mailboxes/akka-mailboxes-common/src/test/scala/akka/actor/mailbox/DurableMailboxSpec.scala
@@ -32,7 +32,7 @@ abstract class DurableMailboxSpec(val backendName: String, val storage: DurableM
"should handle reply to ! for 1 message" in {
val latch = new CountDownLatch(1)
val queueActor = createMailboxTestActor(backendName + " should handle reply to !")
- val sender = actorOf(Props(self ⇒ { case "sum" ⇒ latch.countDown }).withLocalOnly(true))
+ val sender = new LocalActorRef(Props(self ⇒ { case "sum" ⇒ latch.countDown }), newUuid.toString, true)
queueActor.!("sum")(Some(sender))
latch.await(10, TimeUnit.SECONDS) must be(true)
@@ -41,7 +41,7 @@ abstract class DurableMailboxSpec(val backendName: String, val storage: DurableM
"should handle reply to ! for multiple messages" in {
val latch = new CountDownLatch(5)
val queueActor = createMailboxTestActor(backendName + " should handle reply to !")
- val sender = actorOf(Props(self ⇒ { case "sum" ⇒ latch.countDown }).withLocalOnly(true))
+ val sender = new LocalActorRef(Props(self ⇒ { case "sum" ⇒ latch.countDown }), newUuid.toString, true)
for (i ← 1 to 5) queueActor.!("sum")(Some(sender))
diff --git a/akka-durable-mailboxes/akka-mongo-mailbox/src/test/scala/akka/actor/mailbox/MongoBasedMailboxSpec.scala b/akka-durable-mailboxes/akka-mongo-mailbox/src/test/scala/akka/actor/mailbox/MongoBasedMailboxSpec.scala
index 1f1ea055af..1ebded684e 100644
--- a/akka-durable-mailboxes/akka-mongo-mailbox/src/test/scala/akka/actor/mailbox/MongoBasedMailboxSpec.scala
+++ b/akka-durable-mailboxes/akka-mongo-mailbox/src/test/scala/akka/actor/mailbox/MongoBasedMailboxSpec.scala
@@ -33,9 +33,9 @@ class MongoBasedMailboxSpec extends DurableMailboxSpec("mongodb", MongoNaiveDura
}
def createMongoMailboxTestActor(id: String)(implicit dispatcher: MessageDispatcher): ActorRef = {
- val queueActor = localActorOf[MongoMailboxTestActor]
+ val queueActor = actorOf[MongoMailboxTestActor]
queueActor.dispatcher = dispatcher
- queueActor.start
+ queueActor
}
}*/
@@ -48,7 +48,7 @@ class MongoBasedMailboxSpec extends DurableMailboxSpec("mongodb", MongoNaiveDura
"should handle reply to ! for 1 message" in {
val latch = new CountDownLatch(1)
val queueActor = createMongoMailboxTestActor("mongoDB Backend should handle Reply to !")
- val sender = localActorOf(new Actor { def receive = { case "sum" => latch.countDown } }).start
+ val sender = actorOf(new Actor { def receive = { case "sum" => latch.countDown } })
queueActor.!("sum")(Some(sender))
latch.await(10, TimeUnit.SECONDS) must be (true)
@@ -57,7 +57,7 @@ class MongoBasedMailboxSpec extends DurableMailboxSpec("mongodb", MongoNaiveDura
"should handle reply to ! for multiple messages" in {
val latch = new CountDownLatch(5)
val queueActor = createMongoMailboxTestActor("mongoDB Backend should handle reply to !")
- val sender = localActorOf( new Actor { def receive = { case "sum" => latch.countDown } } ).start
+ val sender = actorOf( new Actor { def receive = { case "sum" => latch.countDown } } )
queueActor.!("sum")(Some(sender))
queueActor.!("sum")(Some(sender))
diff --git a/akka-kernel/src/main/scala/akka/kernel/Kernel.scala b/akka-kernel/src/main/scala/akka/kernel/Kernel.scala
index b2710f5086..8bb66acca4 100644
--- a/akka-kernel/src/main/scala/akka/kernel/Kernel.scala
+++ b/akka-kernel/src/main/scala/akka/kernel/Kernel.scala
@@ -32,7 +32,7 @@ object Kernel extends AkkaLoader {
// For testing purposes only
def startRemoteService(): Unit = bundles.foreach(_ match {
- case x: BootableRemoteActorService ⇒ x.startRemoteService
+ case x: BootableRemoteActorService ⇒ x.startRemoteService()
case _ ⇒
})
}
diff --git a/akka-samples/akka-sample-ants/src/main/scala/Ants.scala b/akka-samples/akka-sample-ants/src/main/scala/Ants.scala
index 24349047ec..a6aae6f8bd 100644
--- a/akka-samples/akka-sample-ants/src/main/scala/Ants.scala
+++ b/akka-samples/akka-sample-ants/src/main/scala/Ants.scala
@@ -65,7 +65,7 @@ object World {
val homeOff = Dim / 4
lazy val places = Vector.fill(Dim, Dim)(new Place)
lazy val ants = setup
- lazy val evaporator = actorOf[Evaporator].start()
+ lazy val evaporator = actorOf[Evaporator]
private val snapshotFactory = TransactionFactory(readonly = true, familyName = "snapshot")
@@ -81,7 +81,7 @@ object World {
for (x <- homeRange; y <- homeRange) yield {
place(x, y).makeHome
place(x, y) enter Ant(randomInt(8))
- actorOf(new AntActor(x, y)).start()
+ actorOf(new AntActor(x, y))
}
}
diff --git a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Actors.scala b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Actors.scala
index 81efbd72bd..745dbcaa6c 100644
--- a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Actors.scala
+++ b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Actors.scala
@@ -75,7 +75,7 @@ class Consumer5 extends Actor with Consumer {
def receive = {
case _ ⇒ {
- Actor.actorOf[Consumer4].start
+ Actor.actorOf[Consumer4]
self.reply("Consumer4 started")
}
}
diff --git a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Boot.scala b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Boot.scala
index bb2026f7ba..a1b7e92a0a 100644
--- a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Boot.scala
+++ b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/Boot.scala
@@ -21,8 +21,8 @@ class Boot {
// Basic example
// -----------------------------------------------------------------------
- actorOf[Consumer1].start
- actorOf[Consumer2].start
+ actorOf[Consumer1]
+ actorOf[Consumer2]
// Alternatively, use a supervisor for these actors
//val supervisor = Supervisor(
@@ -47,17 +47,13 @@ class Boot {
val mediator = actorOf(new Transformer(producer))
val consumer = actorOf(new Consumer3(mediator))
- producer.start
- mediator.start
- consumer.start
-
// -----------------------------------------------------------------------
// Asynchronous consumer-producer example (Akka homepage transformation)
// -----------------------------------------------------------------------
- val httpTransformer = actorOf(new HttpTransformer).start
- val httpProducer = actorOf(new HttpProducer(httpTransformer)).start
- val httpConsumer = actorOf(new HttpConsumer(httpProducer)).start
+ val httpTransformer = actorOf(new HttpTransformer)
+ val httpProducer = actorOf(new HttpProducer(httpTransformer))
+ val httpConsumer = actorOf(new HttpConsumer(httpProducer))
// -----------------------------------------------------------------------
// Publish subscribe examples
@@ -68,23 +64,23 @@ class Boot {
//
//val cometdUri = "cometd://localhost:8111/test/abc?baseResource=file:target"
- //val cometdSubscriber = actorOf(new Subscriber("cometd-subscriber", cometdUri)).start
- //val cometdPublisher = actorOf(new Publisher("cometd-publisher", cometdUri)).start
+ //val cometdSubscriber = actorOf(new Subscriber("cometd-subscriber", cometdUri))
+ //val cometdPublisher = actorOf(new Publisher("cometd-publisher", cometdUri))
val jmsUri = "jms:topic:test"
- val jmsSubscriber1 = actorOf(new Subscriber("jms-subscriber-1", jmsUri)).start
- val jmsSubscriber2 = actorOf(new Subscriber("jms-subscriber-2", jmsUri)).start
- val jmsPublisher = actorOf(new Publisher(jmsUri), "jms-publisher").start
+ val jmsSubscriber1 = actorOf(new Subscriber("jms-subscriber-1", jmsUri))
+ val jmsSubscriber2 = actorOf(new Subscriber("jms-subscriber-2", jmsUri))
+ val jmsPublisher = actorOf(new Publisher(jmsUri), "jms-publisher")
- //val cometdPublisherBridge = actorOf(new PublisherBridge("jetty:http://0.0.0.0:8877/camel/pub/cometd", cometdPublisher)).start
- val jmsPublisherBridge = actorOf(new PublisherBridge("jetty:http://0.0.0.0:8877/camel/pub/jms", jmsPublisher)).start
+ //val cometdPublisherBridge = actorOf(new PublisherBridge("jetty:http://0.0.0.0:8877/camel/pub/cometd", cometdPublisher))
+ val jmsPublisherBridge = actorOf(new PublisherBridge("jetty:http://0.0.0.0:8877/camel/pub/jms", jmsPublisher))
// -----------------------------------------------------------------------
// Actor un-publishing and re-publishing example
// -----------------------------------------------------------------------
- actorOf[Consumer4].start // POSTing "stop" to http://0.0.0.0:8877/camel/stop stops and unpublishes this actor
- actorOf[Consumer5].start // POSTing any msg to http://0.0.0.0:8877/camel/start starts and published Consumer4 again.
+ actorOf[Consumer4] // POSTing "stop" to http://0.0.0.0:8877/camel/stop stops and unpublishes this actor
+ actorOf[Consumer5] // POSTing any msg to http://0.0.0.0:8877/camel/start starts and published Consumer4 again.
// -----------------------------------------------------------------------
// Active object example
diff --git a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ClientApplication.scala b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ClientApplication.scala
index 5c257f2ada..c5662ea3b6 100644
--- a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ClientApplication.scala
+++ b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ClientApplication.scala
@@ -11,7 +11,7 @@ object ClientApplication extends App {
/* TODO: fix remote example
- val actor1 = remote.actorOf[RemoteActor1]("localhost", 7777).start
+ val actor1 = remote.actorOf[RemoteActor1]("localhost", 7777)
val actor2 = remote.actorFor("remote2", "localhost", 7777)
val typedActor1 =
diff --git a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ServerApplication.scala b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ServerApplication.scala
index 9e943b4d30..aae7a61d99 100644
--- a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ServerApplication.scala
+++ b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/ServerApplication.scala
@@ -14,7 +14,7 @@ object ServerApplication extends App {
startCamelService
- val ua = actorOf[RemoteActor2].start
+ val ua = actorOf[RemoteActor2]
val ta = TypedActor.typedActorOf(
classOf[RemoteTypedConsumer2],
classOf[RemoteTypedConsumer2Impl], Props())
diff --git a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/StandaloneApplication.scala b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/StandaloneApplication.scala
index f922d9544f..62506c5c5a 100644
--- a/akka-samples/akka-sample-camel/src/main/scala/sample/camel/StandaloneApplication.scala
+++ b/akka-samples/akka-sample-camel/src/main/scala/sample/camel/StandaloneApplication.scala
@@ -91,11 +91,11 @@ object StandaloneJmsApplication extends App {
startCamelService
val jmsUri = "jms:topic:test"
- val jmsPublisher = Actor.actorOf(new Publisher(jmsUri), "jms-publisher").start
+ val jmsPublisher = Actor.actorOf(new Publisher(jmsUri), "jms-publisher")
mandatoryService.awaitEndpointActivation(2) {
- Actor.actorOf(new Subscriber("jms-subscriber-1", jmsUri)).start
- Actor.actorOf(new Subscriber("jms-subscriber-2", jmsUri)).start
+ Actor.actorOf(new Subscriber("jms-subscriber-1", jmsUri))
+ Actor.actorOf(new Subscriber("jms-subscriber-2", jmsUri))
}
// Send 10 messages to via publisher actor
@@ -121,7 +121,7 @@ object StandaloneFileApplication {
def main(args: Array[String]) {
startCamelService
mandatoryService.awaitEndpointActivation(1) {
- Actor.actorOf(new FileConsumer).start
+ Actor.actorOf(new FileConsumer)
}
}
}
diff --git a/akka-samples/akka-sample-camel/src/test/scala/sample/camel/HttpConcurrencyTestStress.scala b/akka-samples/akka-sample-camel/src/test/scala/sample/camel/HttpConcurrencyTestStress.scala
index cdcee916f9..9b5afd4572 100644
--- a/akka-samples/akka-sample-camel/src/test/scala/sample/camel/HttpConcurrencyTestStress.scala
+++ b/akka-samples/akka-sample-camel/src/test/scala/sample/camel/HttpConcurrencyTestStress.scala
@@ -26,9 +26,9 @@ class HttpConcurrencyTestStress extends JUnitSuite {
val latch1 = new CountDownLatch(num)
val latch2 = new CountDownLatch(num)
val latch3 = new CountDownLatch(num)
- val client1 = actorOf(new HttpClientActor("client1", latch1)).start
- val client2 = actorOf(new HttpClientActor("client2", latch2)).start
- val client3 = actorOf(new HttpClientActor("client3", latch3)).start
+ val client1 = actorOf(new HttpClientActor("client1", latch1))
+ val client2 = actorOf(new HttpClientActor("client2", latch2))
+ val client3 = actorOf(new HttpClientActor("client3", latch3))
for (i <- 1 to num) {
client1 ! Message("client1", Map(Message.MessageExchangeId -> i))
client2 ! Message("client2", Map(Message.MessageExchangeId -> i))
@@ -48,11 +48,11 @@ object HttpConcurrencyTestStress {
def beforeClass: Unit = {
startCamelService
- val workers = for (i ← 1 to 8) yield actorOf[HttpServerWorker].start
+ val workers = for (i ← 1 to 8) yield actorOf[HttpServerWorker]
val balancer = Routing.actorOf(
RoutedProps.apply.withRoundRobinRouter.withConnections(workers).withDeployId("loadbalancer"))
//service.get.awaitEndpointActivation(1) {
- // actorOf(new HttpServerActor(balancer)).start
+ // actorOf(new HttpServerActor(balancer))
//}
}
diff --git a/akka-samples/akka-sample-chat/README b/akka-samples/akka-sample-chat/README
index 3606097d6f..965d3d5f75 100644
--- a/akka-samples/akka-sample-chat/README
+++ b/akka-samples/akka-sample-chat/README
@@ -9,7 +9,7 @@ How to run the sample:
2. In the first REPL you get execute:
- scala> import sample.chat._
- scala> import akka.actor.Actor._
- - scala> val chatService = actorOf[ChatService].start()
+ - scala> val chatService = actorOf[ChatService]
3. In the second REPL you get execute:
- scala> import sample.chat._
- scala> ClientRunner.run
diff --git a/akka-samples/akka-sample-chat/src/main/scala/ChatServer.scala b/akka-samples/akka-sample-chat/src/main/scala/ChatServer.scala
index 630b45e5cf..10da961fb5 100644
--- a/akka-samples/akka-sample-chat/src/main/scala/ChatServer.scala
+++ b/akka-samples/akka-sample-chat/src/main/scala/ChatServer.scala
@@ -6,7 +6,7 @@
import scala.collection.mutable.HashMap
- import akka.actor.{Actor, ActorRef}
+ import akka.actor.{Actor, ActorRef, Props}
import akka.stm._
import akka.config.Supervision.{OneForOneStrategy,Permanent}
import Actor._
@@ -24,7 +24,7 @@
2. In the first REPL you get execute:
- scala> import sample.chat._
- scala> import akka.actor.Actor._
- - scala> val chatService = actorOf[ChatService].start()
+ - scala> val chatService = actorOf[ChatService]
3. In the second REPL you get execute:
- scala> import sample.chat._
- scala> ClientRunner.run
@@ -127,7 +127,7 @@
case Login(username) =>
EventHandler.info(this, "User [%s] has logged in".format(username))
val session = actorOf(new Session(username, storage))
- session.start()
+ session
sessions += (username -> session)
case Logout(username) =>
@@ -169,7 +169,7 @@
* Creates and links a MemoryChatStorage.
*/
trait MemoryChatStorageFactory { this: Actor =>
- val storage = this.self.link(actorOf[MemoryChatStorage]).start() // starts and links ChatStorage
+ val storage = actorOf(Props[MemoryChatStorage].withSupervisor(this.self)) // starts and links ChatStorage
}
/**
@@ -201,7 +201,7 @@
* Class encapsulating the full Chat Service.
* Start service by invoking:
*
- * val chatService = Actor.actorOf[ChatService].start()
+ * val chatService = Actor.actorOf[ChatService]
*
*/
class ChatService extends
@@ -223,7 +223,7 @@
def main(args: Array[String]) { ServerRunner.run() }
def run() {
- actorOf[ChatService].start()
+ actorOf[ChatService]
}
}
diff --git a/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnBecome.scala b/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnBecome.scala
index 7e313eb369..f552702c86 100644
--- a/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnBecome.scala
+++ b/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnBecome.scala
@@ -125,11 +125,11 @@ class Hakker(name: String, left: ActorRef, right: ActorRef) extends Actor {
object DiningHakkers {
def run {
//Create 5 chopsticks
- val chopsticks = for (i ← 1 to 5) yield actorOf(new Chopstick("Chopstick " + i)).start()
+ val chopsticks = for (i ← 1 to 5) yield actorOf(new Chopstick("Chopstick " + i))
//Create 5 awesome hakkers and assign them their left and right chopstick
val hakkers = for {
(name, i) ← List("Ghosh", "Bonér", "Klang", "Krasser", "Manie").zipWithIndex
- } yield actorOf(new Hakker(name, chopsticks(i), chopsticks((i + 1) % 5))).start()
+ } yield actorOf(new Hakker(name, chopsticks(i), chopsticks((i + 1) % 5)))
//Signal all hakkers that they should start thinking, and watch the show
hakkers.foreach(_ ! Think)
diff --git a/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnFsm.scala b/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnFsm.scala
index f6c894ccba..014afec2c4 100644
--- a/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnFsm.scala
+++ b/akka-samples/akka-sample-fsm/src/main/scala/DiningHakkersOnFsm.scala
@@ -166,11 +166,11 @@ object DiningHakkersOnFsm {
def run = {
// Create 5 chopsticks
- val chopsticks = for (i ← 1 to 5) yield actorOf(new Chopstick("Chopstick " + i)).start()
+ val chopsticks = for (i ← 1 to 5) yield actorOf(new Chopstick("Chopstick " + i))
// Create 5 awesome fsm hakkers and assign them their left and right chopstick
val hakkers = for {
(name, i) ← List("Ghosh", "Bonér", "Klang", "Krasser", "Manie").zipWithIndex
- } yield actorOf(new FSMHakker(name, chopsticks(i), chopsticks((i + 1) % 5))).start()
+ } yield actorOf(new FSMHakker(name, chopsticks(i), chopsticks((i + 1) % 5)))
hakkers.foreach(_ ! Think)
}
diff --git a/akka-samples/akka-sample-hello/src/main/scala/sample/hello/Boot.scala b/akka-samples/akka-sample-hello/src/main/scala/sample/hello/Boot.scala
index c1769bff04..823ae024ee 100644
--- a/akka-samples/akka-sample-hello/src/main/scala/sample/hello/Boot.scala
+++ b/akka-samples/akka-sample-hello/src/main/scala/sample/hello/Boot.scala
@@ -16,5 +16,5 @@ class Boot {
Supervise(Actor.actorOf[RootEndpoint], Permanent) ::
Supervise(Actor.actorOf[HelloEndpoint], Permanent) :: Nil))
- factory.newInstance.start
+ factory.newInstance.start()
}
diff --git a/akka-samples/akka-sample-hello/src/main/scala/sample/hello/HelloEndpoint.scala b/akka-samples/akka-sample-hello/src/main/scala/sample/hello/HelloEndpoint.scala
index c08aee2618..2ea8c1fe83 100644
--- a/akka-samples/akka-sample-hello/src/main/scala/sample/hello/HelloEndpoint.scala
+++ b/akka-samples/akka-sample-hello/src/main/scala/sample/hello/HelloEndpoint.scala
@@ -19,7 +19,7 @@ class HelloEndpoint extends Actor with Endpoint {
def receive = {
case get: Get => get OK "Hello at " + time
}
- }).start
+ })
def hook: Endpoint.Hook = { case _ => hello }
diff --git a/akka-samples/akka-sample-osgi/src/main/scala/OsgiExample.scala b/akka-samples/akka-sample-osgi/src/main/scala/OsgiExample.scala
index 0e9fda067a..3744061e23 100644
--- a/akka-samples/akka-sample-osgi/src/main/scala/OsgiExample.scala
+++ b/akka-samples/akka-sample-osgi/src/main/scala/OsgiExample.scala
@@ -12,7 +12,7 @@ class Activator extends BundleActivator {
def start(context: BundleContext) {
println("Starting the OSGi example ...")
- val echo = actorOf[EchoActor].start()
+ val echo = actorOf[EchoActor]
val answer = (echo ? "OSGi example").as[String]
println(answer getOrElse "No answer!")
}
diff --git a/akka-spring/src/main/scala/akka/spring/ActorFactoryBean.scala b/akka-spring/src/main/scala/akka/spring/ActorFactoryBean.scala
index 1ffb4e0c3d..41d46a4118 100644
--- a/akka-spring/src/main/scala/akka/spring/ActorFactoryBean.scala
+++ b/akka-spring/src/main/scala/akka/spring/ActorFactoryBean.scala
@@ -107,8 +107,9 @@ class ActorFactoryBean extends AbstractFactoryBean[AnyRef] with ApplicationConte
case UNTYPED_ACTOR_TAG ⇒
val untypedActor = createUntypedInstance()
setProperties(untypedActor.actor)
- if (autostart)
- untypedActor.start
+ //Will always auto start
+ //if (autostart)
+ // untypedActor
untypedActor
case _ ⇒ throw new IllegalArgumentException("Unknown actor type")
}
diff --git a/akka-spring/src/main/scala/akka/spring/CamelServiceFactoryBean.scala b/akka-spring/src/main/scala/akka/spring/CamelServiceFactoryBean.scala
index 37a47c3490..af49617f51 100644
--- a/akka-spring/src/main/scala/akka/spring/CamelServiceFactoryBean.scala
+++ b/akka-spring/src/main/scala/akka/spring/CamelServiceFactoryBean.scala
@@ -34,7 +34,7 @@ class CamelServiceFactoryBean extends FactoryBean[CamelService] with Initializin
CamelContextManager.init(camelContext)
}
instance = CamelServiceFactory.createCamelService
- instance.start
+ instance.start()
}
/**
diff --git a/akka-spring/src/test/java/akka/spring/foo/PingActor.java b/akka-spring/src/test/java/akka/spring/foo/PingActor.java
index 6c4d185860..3a04ba77cf 100644
--- a/akka-spring/src/test/java/akka/spring/foo/PingActor.java
+++ b/akka-spring/src/test/java/akka/spring/foo/PingActor.java
@@ -60,7 +60,7 @@ public class PingActor extends UntypedActor implements ApplicationContextAware {
if (message instanceof String) {
lastMessage = (String) message;
if (message.equals("longRunning")) {
- ActorRef pongActor = actorOf(PongActor.class).start();
+ ActorRef pongActor = actorOf(PongActor.class);
pongActor.sendRequestReply("longRunning", getContext());
}
latch.countDown();
diff --git a/akka-spring/src/test/scala/ActorFactoryBeanTest.scala b/akka-spring/src/test/scala/ActorFactoryBeanTest.scala
index 9345d881e0..27d7a3954c 100644
--- a/akka-spring/src/test/scala/ActorFactoryBeanTest.scala
+++ b/akka-spring/src/test/scala/ActorFactoryBeanTest.scala
@@ -88,7 +88,6 @@ class ActorFactoryBeanTest extends Spec with ShouldMatchers with BeforeAndAfterA
it("should stop the created untyped actor when scope is singleton and the context is closed") {
var ctx = new ClassPathXmlApplicationContext("appContext.xml");
val target = ctx.getBean("untypedActor").asInstanceOf[ActorRef]
- target.start
assert(target.isRunning)
ctx.close
assert(!target.isRunning)
diff --git a/akka-spring/src/test/scala/DispatcherSpringFeatureTest.scala b/akka-spring/src/test/scala/DispatcherSpringFeatureTest.scala
index 002ebbe123..2f37bb3e5e 100644
--- a/akka-spring/src/test/scala/DispatcherSpringFeatureTest.scala
+++ b/akka-spring/src/test/scala/DispatcherSpringFeatureTest.scala
@@ -53,18 +53,7 @@ class DispatcherSpringFeatureTest extends FeatureSpec with ShouldMatchers {
assert(executor.getQueue().remainingCapacity() === Integer.MAX_VALUE)
assert(dispatcher.name === EVENT_DRIVEN_PREFIX + "dispatcher-2")
}
- /*
- scenario("get a executor-event-driven-dispatcher with bounded-blocking-queue and with bounded mailbox capacity") {
- val context = new ClassPathXmlApplicationContext("/dispatcher-config.xml")
- val dispatcher = context.getBean("executor-event-driven-dispatcher-mc").asInstanceOf[ExecutorBasedEventDrivenDispatcher]
- assert(dispatcher.name === EVENT_DRIVEN_PREFIX + "dispatcher-mc")
- val actorRef = UntypedActor.actorOf(classOf[PingActor])
- actorRef.dispatcher = dispatcher
- actorRef.start
- assert(actorRef.mailbox.isInstanceOf[BlockingQueue[MessageInvocation]])
- assert((actorRef.mailbox.asInstanceOf[BlockingQueue[MessageInvocation]]).remainingCapacity === 1000)
- }
-*/
+
scenario("get a executor-event-driven-dispatcher with unbounded-linked-blocking-queue with bounded capacity from context") {
val context = new ClassPathXmlApplicationContext("/dispatcher-config.xml")
val dispatcher = context.getBean("executor-event-driven-dispatcher-4").asInstanceOf[ExecutorBasedEventDrivenDispatcher]
@@ -110,7 +99,6 @@ class DispatcherSpringFeatureTest extends FeatureSpec with ShouldMatchers {
val context = new ClassPathXmlApplicationContext("/dispatcher-config.xml")
val actorRef = context.getBean("untyped-actor-with-thread-based-dispatcher").asInstanceOf[ActorRef]
assert(actorRef.getActorClassName() === "akka.spring.foo.PingActor")
- actorRef.start()
actorRef.tell("Hello")
assert(actorRef.getDispatcher.isInstanceOf[ThreadBasedDispatcher])
}
diff --git a/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala b/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala
index 8dacdb6905..d552a2bb83 100644
--- a/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala
+++ b/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala
@@ -42,7 +42,7 @@ class UntypedActorSpringFeatureTest extends FeatureSpec with ShouldMatchers with
val context = new ClassPathXmlApplicationContext(config)
val pingActor = context.getBean(id).asInstanceOf[ActorRef]
assert(pingActor.getActorClassName() === "akka.spring.foo.PingActor")
- pingActor.start()
+ pingActor
}
feature("parse Spring application context") {
@@ -133,7 +133,7 @@ class UntypedActorSpringFeatureTest extends FeatureSpec with ShouldMatchers with
val context = new ClassPathXmlApplicationContext("/server-managed-config.xml")
val pingActor = context.getBean("server-managed-remote-untyped-actor-custom-id").asInstanceOf[ActorRef]
assert(pingActor.getActorClassName() === "akka.spring.foo.PingActor")
- pingActor.start()
+ pingActor
val nrOfActors = Actor.registry.actors.length
// get client actor ref from spring context
val actorRef = context.getBean("client-1").asInstanceOf[ActorRef]
diff --git a/akka-stm/src/main/scala/akka/agent/Agent.scala b/akka-stm/src/main/scala/akka/agent/Agent.scala
index a07fd1cde3..9d6b5b6df9 100644
--- a/akka-stm/src/main/scala/akka/agent/Agent.scala
+++ b/akka-stm/src/main/scala/akka/agent/Agent.scala
@@ -168,7 +168,7 @@ class Agent[T](initialValue: T) {
val result = new DefaultPromise[T](timeout)
send((value: T) ⇒ {
suspend
- val threadBased = Actor.actorOf(new ThreadBasedAgentUpdater(this)).start()
+ val threadBased = Actor.actorOf(new ThreadBasedAgentUpdater(this))
result completeWith threadBased.?(Update(f), timeout).asInstanceOf[Future[T]]
value
})
diff --git a/akka-stm/src/test/java/akka/stm/example/EitherOrElseExample.java b/akka-stm/src/test/java/akka/stm/example/EitherOrElseExample.java
index 75ce56c4fe..b3c386fb09 100644
--- a/akka-stm/src/test/java/akka/stm/example/EitherOrElseExample.java
+++ b/akka-stm/src/test/java/akka/stm/example/EitherOrElseExample.java
@@ -12,7 +12,7 @@ public class EitherOrElseExample {
final Ref
* class Test extends TestKit {
- * val test = actorOf[SomeActor].start()
+ * val test = actorOf[SomeActor]
*
* within (1 second) {
* test ! SomeWork
@@ -98,7 +98,7 @@ trait TestKitLight {
* ActorRef of the test actor. Access is provided to enable e.g.
* registration as message target.
*/
- val testActor = actorOf(Props(new TestActor(queue)).copy(dispatcher = CallingThreadDispatcher.global, localOnly = true), "testActor" + TestKit.testActorId.incrementAndGet())
+ val testActor = new LocalActorRef(Props(new TestActor(queue)).copy(dispatcher = CallingThreadDispatcher.global), "testActor" + TestKit.testActorId.incrementAndGet(), true)
/**
* Implicit sender reference so that replies are possible for messages sent
diff --git a/akka-testkit/src/test/scala/akka/testkit/TestActorRefSpec.scala b/akka-testkit/src/test/scala/akka/testkit/TestActorRefSpec.scala
index 856298bd28..9d1ae8fab7 100644
--- a/akka-testkit/src/test/scala/akka/testkit/TestActorRefSpec.scala
+++ b/akka-testkit/src/test/scala/akka/testkit/TestActorRefSpec.scala
@@ -192,7 +192,7 @@ class TestActorRefSpec extends WordSpec with MustMatchers with BeforeAndAfterEac
}
"support futures" in {
- val a = TestActorRef[WorkerActor].start()
+ val a = TestActorRef[WorkerActor]
val f = a ? "work" mapTo manifest[String]
f must be('completed)
f.get must equal("workDone")
@@ -208,7 +208,7 @@ class TestActorRefSpec extends WordSpec with MustMatchers with BeforeAndAfterEac
def receiveT = {
case x: String ⇒ s = x
}
- }).start()
+ })
ref ! "hallo"
val actor = ref.underlyingActor
actor.s must equal("hallo")
@@ -225,8 +225,8 @@ class TestActorRefSpec extends WordSpec with MustMatchers with BeforeAndAfterEac
}
"warn about scheduled supervisor" in {
- val boss = Actor.actorOf(new Actor { def receive = { case _ ⇒ } }).start()
- val ref = TestActorRef[WorkerActor].start()
+ val boss = Actor.actorOf(new Actor { def receive = { case _ ⇒ } })
+ val ref = TestActorRef[WorkerActor]
val filter = EventFilter.custom(_ ⇒ true)
EventHandler.notify(TestEvent.Mute(filter))
@@ -241,7 +241,7 @@ class TestActorRefSpec extends WordSpec with MustMatchers with BeforeAndAfterEac
}
"proxy apply for the underlying actor" in {
- val ref = TestActorRef[WorkerActor].start()
+ val ref = TestActorRef[WorkerActor]
intercept[IllegalActorStateException] { ref("work") }
val ch = Promise.channel()
ref ! ch
diff --git a/akka-testkit/src/test/scala/akka/testkit/TestFSMRefSpec.scala b/akka-testkit/src/test/scala/akka/testkit/TestFSMRefSpec.scala
index d65850b050..016dbc35e5 100644
--- a/akka-testkit/src/test/scala/akka/testkit/TestFSMRefSpec.scala
+++ b/akka-testkit/src/test/scala/akka/testkit/TestFSMRefSpec.scala
@@ -25,7 +25,7 @@ class TestFSMRefSpec extends WordSpec with MustMatchers with TestKit {
when(2) {
case Ev("back") ⇒ goto(1) using "back"
}
- }).start()
+ })
fsm.stateName must be(1)
fsm.stateData must be("")
fsm ! "go"
diff --git a/akka-tutorials/akka-tutorial-first/src/main/java/akka/tutorial/first/java/Pi.java b/akka-tutorials/akka-tutorial-first/src/main/java/akka/tutorial/first/java/Pi.java
index ce1c0dd1f7..ab29cf5f52 100644
--- a/akka-tutorials/akka-tutorial-first/src/main/java/akka/tutorial/first/java/Pi.java
+++ b/akka-tutorials/akka-tutorial-first/src/main/java/akka/tutorial/first/java/Pi.java
@@ -105,7 +105,7 @@ public class Pi {
LinkedList workers = new LinkedList();
for (int i = 0; i < nrOfWorkers; i++) {
- ActorRef worker = actorOf(Worker.class, "worker").start();
+ ActorRef worker = actorOf(Worker.class, "worker");
workers.add(worker);
}
@@ -172,7 +172,7 @@ public class Pi {
public UntypedActor create() {
return new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch);
}
- }, "master").start();
+ }, "master");
// start the calculation
master.tell(new Calculate());
diff --git a/akka-tutorials/akka-tutorial-first/src/main/scala/Pi.scala b/akka-tutorials/akka-tutorial-first/src/main/scala/Pi.scala
index 9b05194ead..c5a10e41a3 100644
--- a/akka-tutorials/akka-tutorial-first/src/main/scala/Pi.scala
+++ b/akka-tutorials/akka-tutorial-first/src/main/scala/Pi.scala
@@ -55,7 +55,7 @@ object Pi extends App {
var start: Long = _
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
val router = Routing.actorOf(
@@ -105,7 +105,7 @@ object Pi extends App {
val latch = new CountDownLatch(1)
// create the master
- val master = actorOf(new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch)).start()
+ val master = actorOf(new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
// start the calculation
master ! Calculate
diff --git a/akka-tutorials/akka-tutorial-second/src/main/java/akka/tutorial/java/second/Pi.java b/akka-tutorials/akka-tutorial-second/src/main/java/akka/tutorial/java/second/Pi.java
index 1a6986c22c..51ec382fd5 100644
--- a/akka-tutorials/akka-tutorial-second/src/main/java/akka/tutorial/java/second/Pi.java
+++ b/akka-tutorials/akka-tutorial-second/src/main/java/akka/tutorial/java/second/Pi.java
@@ -99,7 +99,7 @@ public class Pi {
LinkedList workers = new LinkedList();
for (int i = 0; i < nrOfWorkers; i++) {
- ActorRef worker = actorOf(Worker.class, "worker").start();
+ ActorRef worker = actorOf(Worker.class, "worker");
workers.add(worker);
}
@@ -168,7 +168,7 @@ public class Pi {
public UntypedActor create() {
return new Master(nrOfWorkers, nrOfMessages, nrOfElements);
}
- }, "worker").start();
+ }, "worker");
// start the calculation
long start = currentTimeMillis();
diff --git a/akka-tutorials/akka-tutorial-second/src/main/scala/Pi.scala b/akka-tutorials/akka-tutorial-second/src/main/scala/Pi.scala
index 6057ef57d1..2f3717227a 100644
--- a/akka-tutorials/akka-tutorial-second/src/main/scala/Pi.scala
+++ b/akka-tutorials/akka-tutorial-second/src/main/scala/Pi.scala
@@ -50,7 +50,7 @@ object Pi extends App {
var nrOfResults: Int = _
// create the workers
- val workers = Vector.fill(nrOfWorkers)(actorOf[Worker].start())
+ val workers = Vector.fill(nrOfWorkers)(actorOf[Worker])
// wrap them with a load-balancing router
val router = Routing.actorOf(
@@ -100,7 +100,7 @@ object Pi extends App {
// ==================
def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) {
// create the master
- val master = actorOf(new Master(nrOfWorkers, nrOfElements, nrOfMessages)).start()
+ val master = actorOf(new Master(nrOfWorkers, nrOfElements, nrOfMessages))
//start the calculation
val start = now