remove all but one occurrence of single-arg tell()
This commit is contained in:
Roland
parent
2502919c6e
commit
ce49ffe3c6
71 changed files with 550 additions and 538 deletions
|
|
@ -2,7 +2,7 @@ package akka.actor;
|
|||
|
||||
public class JavaAPITestActor extends UntypedActor {
|
||||
public void onReceive(Object msg) {
|
||||
getSender().tell("got it!");
|
||||
getSender().tell("got it!", getSelf());
|
||||
getContext().getChildren();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,6 +12,6 @@ public class NonPublicClass {
|
|||
|
||||
class MyNonPublicActorClass extends UntypedActor {
|
||||
@Override public void onReceive(Object msg) {
|
||||
getSender().tell(msg);
|
||||
getSender().tell(msg, getSelf());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,14 +1,10 @@
|
|||
package akka.actor;
|
||||
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.japi.Creator;
|
||||
import akka.testkit.AkkaSpec;
|
||||
import com.typesafe.config.ConfigFactory;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
import com.typesafe.config.ConfigFactory;
|
||||
|
||||
public class StashJavaAPI {
|
||||
|
||||
|
|
@ -16,7 +12,8 @@ public class StashJavaAPI {
|
|||
|
||||
@BeforeClass
|
||||
public static void beforeAll() {
|
||||
system = ActorSystem.create("StashJavaAPI", ConfigFactory.parseString(ActorWithStashSpec.testConf()));
|
||||
system = ActorSystem.create("StashJavaAPI",
|
||||
ConfigFactory.parseString(ActorWithStashSpec.testConf()));
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
|
|
@ -27,10 +24,11 @@ public class StashJavaAPI {
|
|||
|
||||
@Test
|
||||
public void mustBeAbleToUseStash() {
|
||||
ActorRef ref = system.actorOf(new Props(StashJavaAPITestActor.class).withDispatcher("my-dispatcher"));
|
||||
ActorRef ref = system.actorOf(new Props(StashJavaAPITestActor.class)
|
||||
.withDispatcher("my-dispatcher"));
|
||||
ref.tell("Hello", ref);
|
||||
ref.tell("Hello", ref);
|
||||
ref.tell(new Object());
|
||||
ref.tell(new Object(), null);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,10 +4,11 @@ import static org.junit.Assert.*;
|
|||
|
||||
public class StashJavaAPITestActor extends UntypedActorWithStash {
|
||||
int count = 0;
|
||||
|
||||
public void onReceive(Object msg) {
|
||||
if (msg instanceof String) {
|
||||
if (count < 0) {
|
||||
getSender().tell(new Integer(((String) msg).length()));
|
||||
getSender().tell(new Integer(((String) msg).length()), getSelf());
|
||||
} else if (count == 2) {
|
||||
count = -1;
|
||||
unstashAll();
|
||||
|
|
|
|||
|
|
@ -3,14 +3,9 @@
|
|||
*/
|
||||
package akka.routing;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.actor.Props;
|
||||
import akka.routing.RoundRobinRouter;
|
||||
import akka.testkit.ExtractRoute;
|
||||
|
||||
public class CustomRouteTest {
|
||||
|
|
|
|||
|
|
@ -371,8 +371,8 @@ class ActorRefSpec extends AkkaSpec with DefaultTimeout {
|
|||
val timeout = Timeout(20000)
|
||||
val ref = system.actorOf(Props(new Actor {
|
||||
def receive = {
|
||||
case 5 ⇒ sender.tell("five")
|
||||
case 0 ⇒ sender.tell("null")
|
||||
case 5 ⇒ sender ! "five"
|
||||
case 0 ⇒ sender ! "null"
|
||||
}
|
||||
}))
|
||||
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ object ConsistencySpec {
|
|||
case step: Long ⇒
|
||||
|
||||
if (lastStep != (step - 1))
|
||||
sender.tell("Test failed: Last step %s, this step %s".format(lastStep, step))
|
||||
sender ! "Test failed: Last step %s, this step %s".format(lastStep, step)
|
||||
|
||||
var shouldBeFortyTwo = left.value + right.value
|
||||
if (shouldBeFortyTwo != 42)
|
||||
|
|
|
|||
|
|
@ -59,11 +59,11 @@ object Ticket669Spec {
|
|||
}
|
||||
|
||||
override def preRestart(reason: scala.Throwable, msg: Option[Any]) {
|
||||
sender.tell("failure1")
|
||||
sender ! "failure1"
|
||||
}
|
||||
|
||||
override def postStop() {
|
||||
sender.tell("failure2")
|
||||
sender ! "failure2"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -86,7 +86,7 @@ object ActorModelSpec {
|
|||
case Wait(time) ⇒ ack; Thread.sleep(time); busy.switchOff()
|
||||
case WaitAck(time, l) ⇒ ack; Thread.sleep(time); l.countDown(); busy.switchOff()
|
||||
case Reply(msg) ⇒ ack; sender ! msg; busy.switchOff()
|
||||
case TryReply(msg) ⇒ ack; sender.tell(msg); busy.switchOff()
|
||||
case TryReply(msg) ⇒ ack; sender.tell(msg, null); busy.switchOff()
|
||||
case Forward(to, msg) ⇒ ack; to.forward(msg); busy.switchOff()
|
||||
case CountDown(latch) ⇒ ack; latch.countDown(); busy.switchOff()
|
||||
case Increment(count) ⇒ ack; count.incrementAndGet(); busy.switchOff()
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ class PriorityDispatcherSpec extends AkkaSpec(PriorityDispatcherSpec.config) wit
|
|||
|
||||
def receive = {
|
||||
case i: Int ⇒ acc = i :: acc
|
||||
case 'Result ⇒ sender.tell(acc)
|
||||
case 'Result ⇒ sender ! acc
|
||||
}
|
||||
}).withDispatcher(dispatcherKey)).asInstanceOf[InternalActorRef]
|
||||
|
||||
|
|
|
|||
|
|
@ -537,7 +537,7 @@ class RoutingSpec extends AkkaSpec(RoutingSpec.config) with DefaultTimeout with
|
|||
case _id: Int if (_id == id) ⇒
|
||||
case x ⇒ {
|
||||
Thread sleep 100 * id
|
||||
sender.tell(id)
|
||||
sender ! id
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -6,9 +6,6 @@ package akka.dispatch;
|
|||
|
||||
import akka.util.Unsafe;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
|
||||
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
|
||||
|
||||
abstract class AbstractMessageDispatcher {
|
||||
final static long shutdownScheduleOffset;
|
||||
final static long inhabitantsOffset;
|
||||
|
|
|
|||
|
|
@ -92,13 +92,14 @@ abstract class ActorRef extends java.lang.Comparable[ActorRef] with Serializable
|
|||
* actor.tell(message);
|
||||
* </pre>
|
||||
*/
|
||||
@deprecated("use the two-arg variant (typically getSelf() as second arg)", "2.1")
|
||||
final def tell(msg: Any): Unit = this.!(msg)(null: ActorRef)
|
||||
|
||||
/**
|
||||
* Java API. <p/>
|
||||
* Sends the specified message to the sender, i.e. fire-and-forget
|
||||
* semantics, including the sender reference if possible (not supported on
|
||||
* all senders).<p/>
|
||||
* semantics, including the sender reference if possible (pass `null` if
|
||||
* there is nobody to reply to).<p/>
|
||||
* <pre>
|
||||
* actor.tell(message, context);
|
||||
* </pre>
|
||||
|
|
|
|||
|
|
@ -195,7 +195,7 @@ private[akka] class UnstartedCell(val systemImpl: ActorSystemImpl, val self: Rep
|
|||
lock.unlock()
|
||||
}
|
||||
} else {
|
||||
system.deadLetters.tell(DeadLetter(message, sender, self))
|
||||
system.deadLetters ! DeadLetter(message, sender, self)
|
||||
}
|
||||
}
|
||||
def sendSystemMessage(msg: SystemMessage): Unit = {
|
||||
|
|
@ -209,7 +209,7 @@ private[akka] class UnstartedCell(val systemImpl: ActorSystemImpl, val self: Rep
|
|||
} else {
|
||||
// FIXME: once we have guaranteed delivery of system messages, hook this in!
|
||||
system.eventStream.publish(Warning(self.path.toString, getClass, "dropping system message " + msg + " due to lock timeout"))
|
||||
system.deadLetters.tell(DeadLetter(msg, self, self))
|
||||
system.deadLetters ! DeadLetter(msg, self, self)
|
||||
}
|
||||
}
|
||||
def isLocal = true
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ private[akka] trait Dispatch { this: ActorCell ⇒
|
|||
if (sendSupervise) {
|
||||
// ➡➡➡ NEVER SEND THE SAME SYSTEM MESSAGE OBJECT TO TWO ACTORS ⬅⬅⬅
|
||||
parent.sendSystemMessage(akka.dispatch.Supervise(self, uid))
|
||||
parent.tell(NullMessage) // read ScalaDoc of NullMessage to see why
|
||||
parent ! NullMessage // read ScalaDoc of NullMessage to see why
|
||||
}
|
||||
|
||||
// This call is expected to start off the actor by scheduling its mailbox.
|
||||
|
|
|
|||
|
|
@ -194,7 +194,7 @@ private[akka] trait FaultHandling { this: ActorCell ⇒
|
|||
try if (a ne null) a.postStop()
|
||||
finally try dispatcher.detach(this)
|
||||
finally try parent.sendSystemMessage(ChildTerminated(self))
|
||||
finally try parent.tell(NullMessage) // read ScalaDoc of NullMessage to see why
|
||||
finally try parent ! NullMessage // read ScalaDoc of NullMessage to see why
|
||||
finally try tellWatchersWeDied(a)
|
||||
finally try unwatchWatchedActors(a)
|
||||
finally {
|
||||
|
|
|
|||
|
|
@ -75,7 +75,7 @@ trait AskSupport {
|
|||
*/
|
||||
def ask(actorRef: ActorRef, message: Any)(implicit timeout: Timeout): Future[Any] = actorRef match {
|
||||
case ref: InternalActorRef if ref.isTerminated ⇒
|
||||
actorRef.tell(message)
|
||||
actorRef ! message
|
||||
Future.failed[Any](new AskTimeoutException("Recipient[%s] had already been terminated." format actorRef))
|
||||
case ref: InternalActorRef ⇒
|
||||
if (timeout.duration.length <= 0) Future.failed[Any](new IllegalArgumentException("Timeout length for an `ask` must be greater or equal to 1. Question not sent to [%s]" format actorRef))
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@ import akka.actor.Props;
|
|||
import akka.testkit.AkkaSpec;
|
||||
import akka.testkit.JavaTestKit;
|
||||
|
||||
|
||||
/**
|
||||
* @author Martin Krasser
|
||||
*/
|
||||
|
|
@ -35,25 +34,29 @@ public class ConsumerJavaTestBase {
|
|||
}
|
||||
|
||||
@Test
|
||||
public void shouldHandleExceptionThrownByActorAndGenerateCustomResponse() throws Exception {
|
||||
new JavaTestKit(system) {{
|
||||
public void shouldHandleExceptionThrownByActorAndGenerateCustomResponse()
|
||||
throws Exception {
|
||||
new JavaTestKit(system) {
|
||||
{
|
||||
String result = new EventFilter<String>(Exception.class) {
|
||||
protected String run() {
|
||||
FiniteDuration timeout = Duration.create(1, TimeUnit.SECONDS);
|
||||
Camel camel = CamelExtension.get(system);
|
||||
ExecutionContext executionContext = system.dispatcher();
|
||||
try {
|
||||
ActorRef ref = Await.result(
|
||||
camel.activationFutureFor(system.actorOf(new Props(SampleErrorHandlingConsumer.class)), timeout, executionContext),
|
||||
timeout);
|
||||
return camel.template().requestBody("direct:error-handler-test-java", "hello", String.class);
|
||||
}
|
||||
catch (Exception e) {
|
||||
@SuppressWarnings("unused")
|
||||
ActorRef ref = Await.result(camel.activationFutureFor(
|
||||
system.actorOf(new Props(SampleErrorHandlingConsumer.class)),
|
||||
timeout, executionContext), timeout);
|
||||
return camel.template().requestBody(
|
||||
"direct:error-handler-test-java", "hello", String.class);
|
||||
} catch (Exception e) {
|
||||
return e.getMessage();
|
||||
}
|
||||
}
|
||||
}.occurrences(1).exec();
|
||||
assertEquals("error: hello", result);
|
||||
}};
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -208,7 +208,7 @@ public class CustomRouteTestBase {
|
|||
@Override
|
||||
public void onReceive(Object message) {
|
||||
this.getProducerTemplate().sendBody(to, "test");
|
||||
getSender().tell(Ack.getInstance());
|
||||
getSender().tell(Ack.getInstance(), getSelf());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ public class SampleErrorHandlingConsumer extends UntypedConsumerActor {
|
|||
|
||||
@Override
|
||||
public void preRestart(Throwable reason, Option<Object> message){
|
||||
getSender().tell(new Status.Failure(reason));
|
||||
getSender().tell(new Status.Failure(reason), getSelf());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ public class SampleUntypedConsumer extends UntypedConsumerActor {
|
|||
CamelMessage msg = (CamelMessage)message;
|
||||
String body = msg.getBodyAs(String.class, getCamelContext());
|
||||
String header = msg.getHeaderAs("test", String.class,getCamelContext());
|
||||
sender().tell(String.format("%s %s", body, header));
|
||||
sender().tell(String.format("%s %s", body, header), getSelf());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ public class DangerousJavaActor extends UntypedActor {
|
|||
public Future<String> call() throws Exception {
|
||||
return f;
|
||||
}
|
||||
}));
|
||||
}), getSelf());
|
||||
}
|
||||
if ("block for me".equals(m)) {
|
||||
getSender().tell(breaker
|
||||
|
|
@ -74,7 +74,7 @@ public class DangerousJavaActor extends UntypedActor {
|
|||
public String call() throws Exception {
|
||||
return dangerousCall();
|
||||
}
|
||||
}));
|
||||
}), getSelf());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,5 +22,6 @@ class Java {
|
|||
final Deadline deadline = Duration.create(10, "seconds").fromNow();
|
||||
final Duration rest = deadline.timeLeft();
|
||||
//#deadline
|
||||
rest.toString();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -147,7 +147,7 @@ public class FSMDocTestBase {
|
|||
@Override
|
||||
public void transition(State old, State next) {
|
||||
if (old == State.ACTIVE) {
|
||||
getTarget().tell(new Batch(drainQueue()));
|
||||
getTarget().tell(new Batch(drainQueue()), getSelf());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -175,11 +175,11 @@ public class FSMDocTestBase {
|
|||
public void mustBunch() {
|
||||
final ActorRef buncher = system.actorOf(new Props(MyFSM.class));
|
||||
final TestProbe probe = new TestProbe(system);
|
||||
buncher.tell(new SetTarget(probe.ref()));
|
||||
buncher.tell(new Queue(1));
|
||||
buncher.tell(new Queue(2));
|
||||
buncher.tell(flush);
|
||||
buncher.tell(new Queue(3));
|
||||
buncher.tell(new SetTarget(probe.ref()), null);
|
||||
buncher.tell(new Queue(1), null);
|
||||
buncher.tell(new Queue(2), null);
|
||||
buncher.tell(flush, null);
|
||||
buncher.tell(new Queue(3), null);
|
||||
final Batch b = probe.expectMsgClass(Batch.class);
|
||||
assert b.objects.size() == 2;
|
||||
assert b.objects.contains(1);
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ public class FaultHandlingTestBase {
|
|||
|
||||
public void onReceive(Object o) {
|
||||
if (o instanceof Props) {
|
||||
getSender().tell(getContext().actorOf((Props) o));
|
||||
getSender().tell(getContext().actorOf((Props) o), getSelf());
|
||||
} else {
|
||||
unhandled(o);
|
||||
}
|
||||
|
|
@ -102,7 +102,7 @@ public class FaultHandlingTestBase {
|
|||
|
||||
public void onReceive(Object o) {
|
||||
if (o instanceof Props) {
|
||||
getSender().tell(getContext().actorOf((Props) o));
|
||||
getSender().tell(getContext().actorOf((Props) o), getSelf());
|
||||
} else {
|
||||
unhandled(o);
|
||||
}
|
||||
|
|
@ -126,7 +126,7 @@ public class FaultHandlingTestBase {
|
|||
} else if (o instanceof Integer) {
|
||||
state = (Integer) o;
|
||||
} else if (o.equals("get")) {
|
||||
getSender().tell(state);
|
||||
getSender().tell(state, getSelf());
|
||||
} else {
|
||||
unhandled(o);
|
||||
}
|
||||
|
|
@ -167,21 +167,21 @@ public class FaultHandlingTestBase {
|
|||
//#create
|
||||
|
||||
//#resume
|
||||
child.tell(42);
|
||||
child.tell(42, null);
|
||||
assert Await.result(ask(child, "get", 5000), timeout).equals(42);
|
||||
child.tell(new ArithmeticException());
|
||||
child.tell(new ArithmeticException(), null);
|
||||
assert Await.result(ask(child, "get", 5000), timeout).equals(42);
|
||||
//#resume
|
||||
|
||||
//#restart
|
||||
child.tell(new NullPointerException());
|
||||
child.tell(new NullPointerException(), null);
|
||||
assert Await.result(ask(child, "get", 5000), timeout).equals(0);
|
||||
//#restart
|
||||
|
||||
//#stop
|
||||
final TestProbe probe = new TestProbe(system);
|
||||
probe.watch(child);
|
||||
child.tell(new IllegalArgumentException());
|
||||
child.tell(new IllegalArgumentException(), null);
|
||||
probe.expectMsgClass(Terminated.class);
|
||||
//#stop
|
||||
|
||||
|
|
@ -189,7 +189,7 @@ public class FaultHandlingTestBase {
|
|||
child = (ActorRef) Await.result(ask(supervisor, new Props(Child.class), 5000), timeout);
|
||||
probe.watch(child);
|
||||
assert Await.result(ask(child, "get", 5000), timeout).equals(0);
|
||||
child.tell(new Exception());
|
||||
child.tell(new Exception(), null);
|
||||
probe.expectMsgClass(Terminated.class);
|
||||
//#escalate-kill
|
||||
|
||||
|
|
@ -197,9 +197,9 @@ public class FaultHandlingTestBase {
|
|||
superprops = new Props(Supervisor2.class);
|
||||
supervisor = system.actorOf(superprops);
|
||||
child = (ActorRef) Await.result(ask(supervisor, new Props(Child.class), 5000), timeout);
|
||||
child.tell(23);
|
||||
child.tell(23, null);
|
||||
assert Await.result(ask(child, "get", 5000), timeout).equals(23);
|
||||
child.tell(new Exception());
|
||||
child.tell(new Exception(), null);
|
||||
assert Await.result(ask(child, "get", 5000), timeout).equals(0);
|
||||
//#escalate-restart
|
||||
//#testkit
|
||||
|
|
|
|||
|
|
@ -16,6 +16,6 @@ public class FirstUntypedActor extends UntypedActor {
|
|||
|
||||
public void onReceive(Object message) {
|
||||
myActor.forward(message, getContext());
|
||||
myActor.tell(PoisonPill.getInstance());
|
||||
myActor.tell(PoisonPill.getInstance(), null);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ public class MyReceivedTimeoutUntypedActor extends UntypedActor {
|
|||
|
||||
public void onReceive(Object message) {
|
||||
if (message.equals("Hello")) {
|
||||
getSender().tell("Hello world");
|
||||
getSender().tell("Hello world", getSelf());
|
||||
} else if (message == ReceiveTimeout.getInstance()) {
|
||||
throw new RuntimeException("received timeout");
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ import akka.testkit.AkkaSpec;
|
|||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
public class SchedulerDocTestBase {
|
||||
|
||||
|
|
@ -54,7 +53,7 @@ public class SchedulerDocTestBase {
|
|||
system.scheduler().scheduleOnce(Duration.create(50, TimeUnit.MILLISECONDS), new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
testActor.tell(System.currentTimeMillis());
|
||||
testActor.tell(System.currentTimeMillis(), null);
|
||||
}
|
||||
}, system.dispatcher());
|
||||
//#schedule-one-off-thunk
|
||||
|
|
|
|||
|
|
@ -54,21 +54,15 @@ import java.util.ArrayList;
|
|||
import akka.actor.UntypedActorWithStash;
|
||||
//#import-stash
|
||||
|
||||
import akka.actor.Props;
|
||||
import akka.actor.UntypedActor;
|
||||
import akka.actor.UntypedActorFactory;
|
||||
import akka.dispatch.MessageDispatcher;
|
||||
|
||||
import org.junit.Test;
|
||||
import scala.Option;
|
||||
import java.lang.Object;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import akka.pattern.Patterns;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
public class UntypedActorDocTestBase {
|
||||
|
||||
@Test
|
||||
|
|
@ -95,7 +89,7 @@ public class UntypedActorDocTestBase {
|
|||
ActorSystem system = ActorSystem.create("MySystem");
|
||||
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class), "myactor");
|
||||
//#system-actorOf
|
||||
myActor.tell("test");
|
||||
myActor.tell("test", null);
|
||||
system.shutdown();
|
||||
}
|
||||
|
||||
|
|
@ -105,7 +99,7 @@ public class UntypedActorDocTestBase {
|
|||
ActorSystem system = ActorSystem.create("MySystem");
|
||||
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class), "myactor");
|
||||
//#context-actorOf
|
||||
myActor.tell("test");
|
||||
myActor.tell("test", null);
|
||||
system.shutdown();
|
||||
}
|
||||
|
||||
|
|
@ -120,7 +114,7 @@ public class UntypedActorDocTestBase {
|
|||
}
|
||||
}), "myactor");
|
||||
//#creating-constructor
|
||||
myActor.tell("test");
|
||||
myActor.tell("test", null);
|
||||
system.shutdown();
|
||||
}
|
||||
|
||||
|
|
@ -130,7 +124,7 @@ public class UntypedActorDocTestBase {
|
|||
//#creating-props
|
||||
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class).withDispatcher("my-dispatcher"), "myactor");
|
||||
//#creating-props
|
||||
myActor.tell("test");
|
||||
myActor.tell("test", null);
|
||||
system.shutdown();
|
||||
}
|
||||
|
||||
|
|
@ -154,7 +148,7 @@ public class UntypedActorDocTestBase {
|
|||
public void receiveTimeout() {
|
||||
ActorSystem system = ActorSystem.create("MySystem");
|
||||
ActorRef myActor = system.actorOf(new Props(MyReceivedTimeoutUntypedActor.class));
|
||||
myActor.tell("Hello");
|
||||
myActor.tell("Hello", null);
|
||||
system.shutdown();
|
||||
}
|
||||
|
||||
|
|
@ -163,7 +157,7 @@ public class UntypedActorDocTestBase {
|
|||
ActorSystem system = ActorSystem.create("MySystem");
|
||||
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class));
|
||||
//#poison-pill
|
||||
myActor.tell(PoisonPill.getInstance());
|
||||
myActor.tell(PoisonPill.getInstance(), null);
|
||||
//#poison-pill
|
||||
system.shutdown();
|
||||
}
|
||||
|
|
@ -173,7 +167,7 @@ public class UntypedActorDocTestBase {
|
|||
ActorSystem system = ActorSystem.create("MySystem");
|
||||
ActorRef victim = system.actorOf(new Props(MyUntypedActor.class));
|
||||
//#kill
|
||||
victim.tell(Kill.getInstance());
|
||||
victim.tell(Kill.getInstance(), null);
|
||||
//#kill
|
||||
system.shutdown();
|
||||
}
|
||||
|
|
@ -186,9 +180,9 @@ public class UntypedActorDocTestBase {
|
|||
return new HotSwapActor();
|
||||
}
|
||||
}));
|
||||
myActor.tell("foo");
|
||||
myActor.tell("bar");
|
||||
myActor.tell("bar");
|
||||
myActor.tell("foo", null);
|
||||
myActor.tell("bar", null);
|
||||
myActor.tell("bar", null);
|
||||
system.shutdown();
|
||||
}
|
||||
|
||||
|
|
@ -265,7 +259,7 @@ public class UntypedActorDocTestBase {
|
|||
try {
|
||||
operation();
|
||||
} catch (Exception e) {
|
||||
getSender().tell(new akka.actor.Status.Failure(e));
|
||||
getSender().tell(new akka.actor.Status.Failure(e), getSelf());
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
|
@ -298,9 +292,9 @@ public class UntypedActorDocTestBase {
|
|||
//#reply-exception
|
||||
try {
|
||||
String result = operation();
|
||||
getSender().tell(result);
|
||||
getSender().tell(result, getSelf());
|
||||
} catch (Exception e) {
|
||||
getSender().tell(new akka.actor.Status.Failure(e));
|
||||
getSender().tell(new akka.actor.Status.Failure(e), getSelf());
|
||||
throw e;
|
||||
}
|
||||
//#reply-exception
|
||||
|
|
@ -318,7 +312,7 @@ public class UntypedActorDocTestBase {
|
|||
@Override
|
||||
public void apply(Object message) {
|
||||
if (message.equals("bar")) {
|
||||
getSender().tell("I am already angry?");
|
||||
getSender().tell("I am already angry?", getSelf());
|
||||
} else if (message.equals("foo")) {
|
||||
getContext().become(happy);
|
||||
}
|
||||
|
|
@ -329,7 +323,7 @@ public class UntypedActorDocTestBase {
|
|||
@Override
|
||||
public void apply(Object message) {
|
||||
if (message.equals("bar")) {
|
||||
getSender().tell("I am already happy :-)");
|
||||
getSender().tell("I am already happy :-)", getSelf());
|
||||
} else if (message.equals("foo")) {
|
||||
getContext().become(angry);
|
||||
}
|
||||
|
|
@ -390,7 +384,7 @@ public class UntypedActorDocTestBase {
|
|||
} else if (message instanceof Terminated) {
|
||||
final Terminated t = (Terminated) message;
|
||||
if (t.getActor() == child) {
|
||||
lastSender.tell("finished");
|
||||
lastSender.tell("finished", getSelf());
|
||||
}
|
||||
} else {
|
||||
unhandled(message);
|
||||
|
|
|
|||
|
|
@ -44,12 +44,12 @@ public class UntypedActorSwapper {
|
|||
public static void main(String... args) {
|
||||
ActorSystem system = ActorSystem.create("MySystem");
|
||||
ActorRef swap = system.actorOf(new Props(Swapper.class));
|
||||
swap.tell(SWAP); // logs Hi
|
||||
swap.tell(SWAP); // logs Ho
|
||||
swap.tell(SWAP); // logs Hi
|
||||
swap.tell(SWAP); // logs Ho
|
||||
swap.tell(SWAP); // logs Hi
|
||||
swap.tell(SWAP); // logs Ho
|
||||
swap.tell(SWAP, null); // logs Hi
|
||||
swap.tell(SWAP, null); // logs Ho
|
||||
swap.tell(SWAP, null); // logs Hi
|
||||
swap.tell(SWAP, null); // logs Ho
|
||||
swap.tell(SWAP, null); // logs Hi
|
||||
swap.tell(SWAP, null); // logs Ho
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,8 +14,6 @@ package docs.camel;
|
|||
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
public class ActivationTestBase {
|
||||
|
||||
@Test
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ public class Consumer2 extends UntypedConsumerActor {
|
|||
if (message instanceof CamelMessage) {
|
||||
CamelMessage camelMessage = (CamelMessage) message;
|
||||
String body = camelMessage.getBodyAs(String.class, getCamelContext());
|
||||
getSender().tell(String.format("Received message: %s",body));
|
||||
getSender().tell(String.format("Received message: %s",body), getSelf());
|
||||
} else
|
||||
unhandled(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,12 +18,12 @@ public class Consumer3 extends UntypedConsumerActor{
|
|||
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof CamelMessage) {
|
||||
getSender().tell(Ack.getInstance());
|
||||
getSender().tell(Ack.getInstance(), getSelf());
|
||||
// on success
|
||||
// ..
|
||||
Exception someException = new Exception("e1");
|
||||
// on failure
|
||||
getSender().tell(new Status.Failure(someException));
|
||||
getSender().tell(new Status.Failure(someException), getSelf());
|
||||
} else
|
||||
unhandled(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ public class Consumer4 extends UntypedConsumerActor {
|
|||
if (message instanceof CamelMessage) {
|
||||
CamelMessage camelMessage = (CamelMessage) message;
|
||||
String body = camelMessage.getBodyAs(String.class, getCamelContext());
|
||||
getSender().tell(String.format("Hello %s",body));
|
||||
getSender().tell(String.format("Hello %s",body), getSelf());
|
||||
} else
|
||||
unhandled(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ public class ErrorThrowingConsumer extends UntypedConsumerActor{
|
|||
|
||||
@Override
|
||||
public void preRestart(Throwable reason, Option<Object> message) {
|
||||
getSender().tell(new Status.Failure(reason));
|
||||
getSender().tell(new Status.Failure(reason), getSelf());
|
||||
}
|
||||
}
|
||||
//#ErrorThrowingConsumer
|
||||
|
|
@ -1,7 +1,6 @@
|
|||
package docs.camel;
|
||||
|
||||
import akka.actor.*;
|
||||
import org.junit.Test;
|
||||
|
||||
public class OnRouteResponseTestBase {
|
||||
|
||||
|
|
@ -18,7 +17,7 @@ public class OnRouteResponseTestBase {
|
|||
ActorRef forwardResponse = system.actorOf(new Props(factory));
|
||||
// the Forwarder sends out a request to the web page and forwards the response to
|
||||
// the ResponseReceiver
|
||||
forwardResponse.tell("some request");
|
||||
forwardResponse.tell("some request", null);
|
||||
//#RouteResponse
|
||||
system.stop(receiver);
|
||||
system.stop(forwardResponse);
|
||||
|
|
|
|||
|
|
@ -1,20 +1,14 @@
|
|||
package docs.camel;
|
||||
|
||||
import akka.actor.*;
|
||||
import akka.camel.Camel;
|
||||
import akka.camel.CamelExtension;
|
||||
import akka.camel.CamelMessage;
|
||||
import akka.pattern.Patterns;
|
||||
import scala.concurrent.Future;
|
||||
import scala.concurrent.util.Duration;
|
||||
import scala.concurrent.util.FiniteDuration;
|
||||
import org.apache.camel.CamelContext;
|
||||
import org.apache.camel.ProducerTemplate;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import scala.concurrent.Future;
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.actor.Props;
|
||||
import akka.camel.CamelMessage;
|
||||
import akka.pattern.Patterns;
|
||||
|
||||
public class ProducerTestBase {
|
||||
public void tellJmsProducer() {
|
||||
|
|
@ -22,7 +16,7 @@ public class ProducerTestBase {
|
|||
ActorSystem system = ActorSystem.create("some-system");
|
||||
Props props = new Props(Orders.class);
|
||||
ActorRef producer = system.actorOf(props, "jmsproducer");
|
||||
producer.tell("<order amount=\"100\" currency=\"PLN\" itemId=\"12345\"/>");
|
||||
producer.tell("<order amount=\"100\" currency=\"PLN\" itemId=\"12345\"/>", null);
|
||||
//#TellProducer
|
||||
system.shutdown();
|
||||
}
|
||||
|
|
@ -45,7 +39,7 @@ public class ProducerTestBase {
|
|||
ActorRef producer = system.actorOf(props,"jmsproducer");
|
||||
Map<String,Object> headers = new HashMap<String, Object>();
|
||||
headers.put(CamelMessage.MessageExchangeId(),"123");
|
||||
producer.tell(new CamelMessage("<order amount=\"100\" currency=\"PLN\" itemId=\"12345\"/>",headers));
|
||||
producer.tell(new CamelMessage("<order amount=\"100\" currency=\"PLN\" itemId=\"12345\"/>",headers), null);
|
||||
//#Correlate
|
||||
system.stop(producer);
|
||||
system.shutdown();
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ public class RequestBodyActor extends UntypedActor {
|
|||
public void onReceive(Object message) {
|
||||
Camel camel = CamelExtension.get(getContext().system());
|
||||
ProducerTemplate template = camel.template();
|
||||
getSender().tell(template.requestBody("direct:news", message));
|
||||
getSender().tell(template.requestBody("direct:news", message), getSelf());
|
||||
}
|
||||
}
|
||||
//#RequestProducerTemplate
|
||||
|
|
@ -9,7 +9,7 @@ public class Responder extends UntypedActor{
|
|||
public void onReceive(Object message) {
|
||||
if (message instanceof CamelMessage) {
|
||||
CamelMessage camelMessage = (CamelMessage) message;
|
||||
getSender().tell(createResponse(camelMessage));
|
||||
getSender().tell(createResponse(camelMessage), getSelf());
|
||||
} else
|
||||
unhandled(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,9 +16,9 @@ public class HttpTransformer extends UntypedActor{
|
|||
return text.replaceAll("Akka ", "AKKA ");
|
||||
}
|
||||
});
|
||||
getSender().tell(replacedMessage);
|
||||
getSender().tell(replacedMessage, getSelf());
|
||||
} else if (message instanceof Status.Failure) {
|
||||
getSender().tell(message);
|
||||
getSender().tell(message, getSelf());
|
||||
} else
|
||||
unhandled(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,12 +37,10 @@ import org.junit.After;
|
|||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import scala.Option;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
import com.typesafe.config.ConfigFactory;
|
||||
|
||||
import docs.actor.MyUntypedActor;
|
||||
import docs.actor.UntypedActorDocTestBase.MyActor;
|
||||
import akka.testkit.AkkaSpec;
|
||||
|
||||
public class DispatcherDocTestBase {
|
||||
|
|
@ -89,14 +87,14 @@ public class DispatcherDocTestBase {
|
|||
LoggingAdapter log =
|
||||
Logging.getLogger(getContext().system(), this);
|
||||
{
|
||||
getSelf().tell("lowpriority");
|
||||
getSelf().tell("lowpriority");
|
||||
getSelf().tell("highpriority");
|
||||
getSelf().tell("pigdog");
|
||||
getSelf().tell("pigdog2");
|
||||
getSelf().tell("pigdog3");
|
||||
getSelf().tell("highpriority");
|
||||
getSelf().tell(PoisonPill.getInstance());
|
||||
getSelf().tell("lowpriority", getSelf());
|
||||
getSelf().tell("lowpriority", getSelf());
|
||||
getSelf().tell("highpriority", getSelf());
|
||||
getSelf().tell("pigdog", getSelf());
|
||||
getSelf().tell("pigdog2", getSelf());
|
||||
getSelf().tell("pigdog3", getSelf());
|
||||
getSelf().tell("highpriority", getSelf());
|
||||
getSelf().tell(PoisonPill.getInstance(), getSelf());
|
||||
}
|
||||
|
||||
public void onReceive(Object message) {
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@ import akka.event.Logging.Debug;
|
|||
import org.junit.Test;
|
||||
|
||||
import scala.Option;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
import akka.actor.UntypedActorFactory;
|
||||
//#imports-deadletter
|
||||
|
|
@ -42,7 +41,7 @@ public class LoggingDocTestBase {
|
|||
return new MyActor();
|
||||
}
|
||||
}));
|
||||
myActor.tell("test");
|
||||
myActor.tell("test", null);
|
||||
system.shutdown();
|
||||
}
|
||||
|
||||
|
|
@ -96,7 +95,7 @@ public class LoggingDocTestBase {
|
|||
class MyEventListener extends UntypedActor {
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof InitializeLogger) {
|
||||
getSender().tell(Logging.loggerInitialized());
|
||||
getSender().tell(Logging.loggerInitialized(), getSelf());
|
||||
} else if (message instanceof Error) {
|
||||
// ...
|
||||
} else if (message instanceof Warning) {
|
||||
|
|
|
|||
|
|
@ -534,13 +534,13 @@ public class FutureDocTestBase {
|
|||
public static class MyActor extends UntypedActor {
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof String) {
|
||||
getSender().tell(((String) message).toUpperCase());
|
||||
getSender().tell(((String) message).toUpperCase(), getSelf());
|
||||
} else if (message instanceof Integer) {
|
||||
int i = ((Integer) message).intValue();
|
||||
if (i < 0) {
|
||||
getSender().tell(new Failure(new ArithmeticException("Negative values not supported")));
|
||||
getSender().tell(new Failure(new ArithmeticException("Negative values not supported")), getSelf());
|
||||
} else {
|
||||
getSender().tell(i);
|
||||
getSender().tell(i, getSelf());
|
||||
}
|
||||
} else {
|
||||
unhandled(message);
|
||||
|
|
|
|||
|
|
@ -3,28 +3,37 @@
|
|||
*/
|
||||
package docs.jrouting;
|
||||
|
||||
import java.util.List;
|
||||
import static akka.pattern.Patterns.ask;
|
||||
import static docs.jrouting.CustomRouterDocTestBase.Message.DemocratCountResult;
|
||||
import static docs.jrouting.CustomRouterDocTestBase.Message.DemocratVote;
|
||||
import static docs.jrouting.CustomRouterDocTestBase.Message.RepublicanCountResult;
|
||||
import static docs.jrouting.CustomRouterDocTestBase.Message.RepublicanVote;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import akka.actor.*;
|
||||
import akka.routing.*;
|
||||
import scala.concurrent.util.Duration;
|
||||
import akka.util.Timeout;
|
||||
import scala.concurrent.Await;
|
||||
import scala.concurrent.Future;
|
||||
import scala.concurrent.util.Duration;
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.actor.OneForOneStrategy;
|
||||
import akka.actor.Props;
|
||||
import akka.actor.SupervisorStrategy;
|
||||
import akka.actor.UntypedActor;
|
||||
import akka.dispatch.Dispatchers;
|
||||
import akka.routing.CustomRoute;
|
||||
import akka.routing.CustomRouterConfig;
|
||||
import akka.routing.Destination;
|
||||
import akka.routing.RoundRobinRouter;
|
||||
import akka.routing.RouteeProvider;
|
||||
import akka.testkit.AkkaSpec;
|
||||
import com.typesafe.config.ConfigFactory;
|
||||
import static akka.pattern.Patterns.ask;
|
||||
|
||||
import static docs.jrouting.CustomRouterDocTestBase.DemocratActor;
|
||||
import static docs.jrouting.CustomRouterDocTestBase.RepublicanActor;
|
||||
import static docs.jrouting.CustomRouterDocTestBase.Message.*;
|
||||
import akka.util.Timeout;
|
||||
|
||||
public class CustomRouterDocTestBase {
|
||||
|
||||
|
|
@ -67,11 +76,11 @@ public class CustomRouterDocTestBase {
|
|||
@Test
|
||||
public void countVotesAsIntendedNotAsInFlorida() throws Exception {
|
||||
ActorRef routedActor = system.actorOf(new Props().withRouter(new VoteCountRouter()));
|
||||
routedActor.tell(DemocratVote);
|
||||
routedActor.tell(DemocratVote);
|
||||
routedActor.tell(RepublicanVote);
|
||||
routedActor.tell(DemocratVote);
|
||||
routedActor.tell(RepublicanVote);
|
||||
routedActor.tell(DemocratVote, null);
|
||||
routedActor.tell(DemocratVote, null);
|
||||
routedActor.tell(RepublicanVote, null);
|
||||
routedActor.tell(DemocratVote, null);
|
||||
routedActor.tell(RepublicanVote, null);
|
||||
Timeout timeout = new Timeout(Duration.create(1, "seconds"));
|
||||
Future<Object> democratsResult = ask(routedActor, DemocratCountResult, timeout);
|
||||
Future<Object> republicansResult = ask(routedActor, RepublicanCountResult, timeout);
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ public class FibonacciActor extends UntypedActor {
|
|||
public void onReceive(Object msg) {
|
||||
if (msg instanceof FibonacciNumber) {
|
||||
FibonacciNumber fibonacciNumber = (FibonacciNumber) msg;
|
||||
getSender().tell(fibonacci(fibonacciNumber.getNbr()));
|
||||
getSender().tell(fibonacci(fibonacciNumber.getNbr()), getSelf());
|
||||
} else {
|
||||
unhandled(msg);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -45,14 +45,14 @@ public class RouterViaConfigExample {
|
|||
ActorRef router = system.actorOf(new Props(ExampleActor.class).withRouter(new FromConfig()), "router");
|
||||
//#configurableRouting
|
||||
for (int i = 1; i <= 10; i++) {
|
||||
router.tell(new ExampleActor.Message(i));
|
||||
router.tell(new ExampleActor.Message(i), null);
|
||||
}
|
||||
|
||||
//#configurableRoutingWithResizer
|
||||
ActorRef router2 = system.actorOf(new Props(ExampleActor.class).withRouter(new FromConfig()), "router2");
|
||||
//#configurableRoutingWithResizer
|
||||
for (int i = 1; i <= 10; i++) {
|
||||
router2.tell(new ExampleActor.Message(i));
|
||||
router2.tell(new ExampleActor.Message(i), null);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -47,7 +47,7 @@ public class RouterViaProgramExample {
|
|||
ActorRef router1 = system.actorOf(new Props(ExampleActor.class).withRouter(new RoundRobinRouter(nrOfInstances)));
|
||||
//#programmaticRoutingNrOfInstances
|
||||
for (int i = 1; i <= 6; i++) {
|
||||
router1.tell(new ExampleActor.Message(i));
|
||||
router1.tell(new ExampleActor.Message(i), null);
|
||||
}
|
||||
|
||||
//#programmaticRoutingRoutees
|
||||
|
|
@ -58,7 +58,7 @@ public class RouterViaProgramExample {
|
|||
ActorRef router2 = system.actorOf(new Props().withRouter(RoundRobinRouter.create(routees)));
|
||||
//#programmaticRoutingRoutees
|
||||
for (int i = 1; i <= 6; i++) {
|
||||
router2.tell(new ExampleActor.Message(i));
|
||||
router2.tell(new ExampleActor.Message(i), null);
|
||||
}
|
||||
|
||||
//#programmaticRoutingWithResizer
|
||||
|
|
@ -68,7 +68,7 @@ public class RouterViaProgramExample {
|
|||
ActorRef router3 = system.actorOf(new Props(ExampleActor.class).withRouter(new RoundRobinRouter(nrOfInstances)));
|
||||
//#programmaticRoutingWithResizer
|
||||
for (int i = 1; i <= 6; i++) {
|
||||
router3.tell(new ExampleActor.Message(i));
|
||||
router3.tell(new ExampleActor.Message(i), null);
|
||||
}
|
||||
|
||||
//#remoteRoutees
|
||||
|
|
|
|||
|
|
@ -9,7 +9,6 @@ import static org.junit.Assert.*;
|
|||
import akka.actor.*;
|
||||
import akka.remote.RemoteActorRefProvider;
|
||||
import akka.serialization.*;
|
||||
import com.typesafe.config.*;
|
||||
|
||||
//#imports
|
||||
|
||||
|
|
|
|||
|
|
@ -97,7 +97,7 @@ public class TestKitDocTest {
|
|||
public void demonstrateWithin() {
|
||||
//#test-within
|
||||
new JavaTestKit(system) {{
|
||||
getRef().tell(42);
|
||||
getRef().tell(42, null);
|
||||
new Within(Duration.Zero(), Duration.create(1, "second")) {
|
||||
// do not put code outside this method, will run afterwards
|
||||
public void run() {
|
||||
|
|
@ -112,7 +112,7 @@ public class TestKitDocTest {
|
|||
public void demonstrateExpectMsg() {
|
||||
//#test-expectmsg
|
||||
new JavaTestKit(system) {{
|
||||
getRef().tell(42);
|
||||
getRef().tell(42, null);
|
||||
final String out = new ExpectMsg<String>("match hint") {
|
||||
// do not put code outside this method, will run afterwards
|
||||
protected String match(Object in) {
|
||||
|
|
@ -132,9 +132,9 @@ public class TestKitDocTest {
|
|||
public void demonstrateReceiveWhile() {
|
||||
//#test-receivewhile
|
||||
new JavaTestKit(system) {{
|
||||
getRef().tell(42);
|
||||
getRef().tell(43);
|
||||
getRef().tell("hello");
|
||||
getRef().tell(42, null);
|
||||
getRef().tell(43, null);
|
||||
getRef().tell("hello", null);
|
||||
final String[] out =
|
||||
new ReceiveWhile<String>(String.class, duration("1 second")) {
|
||||
// do not put code outside this method, will run afterwards
|
||||
|
|
@ -172,7 +172,7 @@ public class TestKitDocTest {
|
|||
public void demonstrateAwaitCond() {
|
||||
//#test-awaitCond
|
||||
new JavaTestKit(system) {{
|
||||
getRef().tell(42);
|
||||
getRef().tell(42, null);
|
||||
new AwaitCond(
|
||||
duration("1 second"), // maximum wait time
|
||||
duration("100 millis") // interval at which to check the condition
|
||||
|
|
@ -191,12 +191,12 @@ public class TestKitDocTest {
|
|||
@SuppressWarnings("unchecked") // due to generic varargs
|
||||
public void demonstrateExpect() {
|
||||
new JavaTestKit(system) {{
|
||||
getRef().tell("hello");
|
||||
getRef().tell("hello");
|
||||
getRef().tell("hello");
|
||||
getRef().tell("world");
|
||||
getRef().tell(42);
|
||||
getRef().tell(42);
|
||||
getRef().tell("hello", null);
|
||||
getRef().tell("hello", null);
|
||||
getRef().tell("hello", null);
|
||||
getRef().tell("world", null);
|
||||
getRef().tell(42, null);
|
||||
getRef().tell(42, null);
|
||||
//#test-expect
|
||||
final String hello = expectMsgEquals("hello");
|
||||
final Object any = expectMsgAnyOf("hello", "world");
|
||||
|
|
@ -223,12 +223,12 @@ public class TestKitDocTest {
|
|||
return msg instanceof String;
|
||||
}
|
||||
};
|
||||
getRef().tell("hello");
|
||||
getRef().tell(42);
|
||||
getRef().tell("hello", null);
|
||||
getRef().tell(42, null);
|
||||
expectMsgEquals(42);
|
||||
// remove message filter
|
||||
ignoreNoMsg();
|
||||
getRef().tell("hello");
|
||||
getRef().tell("hello", null);
|
||||
expectMsgEquals("hello");
|
||||
}};
|
||||
//#test-ignoreMsg
|
||||
|
|
@ -294,7 +294,7 @@ public class TestKitDocTest {
|
|||
}
|
||||
|
||||
final MyProbe probe = new MyProbe();
|
||||
probe.getRef().tell("hello");
|
||||
probe.getRef().tell("hello", null);
|
||||
probe.assertHello();
|
||||
}};
|
||||
//#test-special-probe
|
||||
|
|
@ -354,7 +354,7 @@ public class TestKitDocTest {
|
|||
// install auto-pilot
|
||||
probe.setAutoPilot(new TestActor.AutoPilot() {
|
||||
public AutoPilot run(ActorRef sender, Object msg) {
|
||||
sender.tell(msg);
|
||||
sender.tell(msg, null);
|
||||
return noAutoPilot();
|
||||
}
|
||||
});
|
||||
|
|
@ -386,7 +386,7 @@ public class TestKitDocTest {
|
|||
|
||||
final int result = new EventFilter<Integer>(ActorKilledException.class) {
|
||||
protected Integer run() {
|
||||
victim.tell(Kill.getInstance());
|
||||
victim.tell(Kill.getInstance(), null);
|
||||
return 42;
|
||||
}
|
||||
}.from("akka://demoSystem/user/victim").occurrences(1).exec();
|
||||
|
|
|
|||
|
|
@ -24,12 +24,12 @@ public class TestKitSampleTest {
|
|||
public void onReceive(Object msg) {
|
||||
|
||||
if (msg.equals("hello")) {
|
||||
getSender().tell("world");
|
||||
getSender().tell("world", getSelf());
|
||||
if (target != null) target.forward(msg, getContext());
|
||||
|
||||
} else if (msg instanceof ActorRef) {
|
||||
target = (ActorRef) msg;
|
||||
getSender().tell("done");
|
||||
getSender().tell("done", getSelf());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ public class CoordinatedCounter extends UntypedActor {
|
|||
if (message instanceof Increment) {
|
||||
Increment increment = (Increment) message;
|
||||
if (increment.hasFriend()) {
|
||||
increment.getFriend().tell(coordinated.coordinate(new Increment()));
|
||||
increment.getFriend().tell(coordinated.coordinate(new Increment()), getSelf());
|
||||
}
|
||||
coordinated.atomic(new Runnable() {
|
||||
public void run() {
|
||||
|
|
@ -29,7 +29,7 @@ public class CoordinatedCounter extends UntypedActor {
|
|||
});
|
||||
}
|
||||
} else if ("GetCount".equals(incoming)) {
|
||||
getSender().tell(count.get());
|
||||
getSender().tell(count.get(), getSelf());
|
||||
} else {
|
||||
unhandled(incoming);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ public class Counter extends UntypedTransactor {
|
|||
|
||||
@Override public boolean normally(Object message) {
|
||||
if ("GetCount".equals(message)) {
|
||||
getSender().tell(count.get());
|
||||
getSender().tell(count.get(), getSelf());
|
||||
return true;
|
||||
} else return false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@
|
|||
package docs.transactor;
|
||||
|
||||
//#class
|
||||
import akka.actor.*;
|
||||
import akka.transactor.*;
|
||||
import java.util.Set;
|
||||
import scala.concurrent.stm.Ref;
|
||||
|
|
@ -31,7 +30,7 @@ public class FriendlyCounter extends UntypedTransactor {
|
|||
|
||||
@Override public boolean normally(Object message) {
|
||||
if ("GetCount".equals(message)) {
|
||||
getSender().tell(count.get());
|
||||
getSender().tell(count.get(), getSelf());
|
||||
return true;
|
||||
} else return false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,7 +12,6 @@ import akka.actor.*;
|
|||
import scala.concurrent.Await;
|
||||
import static akka.pattern.Patterns.ask;
|
||||
import akka.transactor.Coordinated;
|
||||
import scala.concurrent.util.Duration;
|
||||
import akka.util.Timeout;
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
//#imports
|
||||
|
|
@ -29,7 +28,7 @@ public class TransactorDocTest {
|
|||
|
||||
Timeout timeout = new Timeout(5, SECONDS);
|
||||
|
||||
counter1.tell(new Coordinated(new Increment(counter2), timeout));
|
||||
counter1.tell(new Coordinated(new Increment(counter2), timeout), null);
|
||||
|
||||
Integer count = (Integer) Await.result(ask(counter1, "GetCount", timeout), timeout.duration());
|
||||
//#coordinated-example
|
||||
|
|
@ -50,11 +49,11 @@ public class TransactorDocTest {
|
|||
ActorRef actor = system.actorOf(new Props(Coordinator.class));
|
||||
|
||||
//#send-coordinated
|
||||
actor.tell(new Coordinated(new Message(), timeout));
|
||||
actor.tell(new Coordinated(new Message(), timeout), null);
|
||||
//#send-coordinated
|
||||
|
||||
//#include-coordinated
|
||||
actor.tell(coordinated.coordinate(new Message()));
|
||||
actor.tell(coordinated.coordinate(new Message()), null);
|
||||
//#include-coordinated
|
||||
|
||||
coordinated.await();
|
||||
|
|
@ -69,7 +68,7 @@ public class TransactorDocTest {
|
|||
|
||||
Timeout timeout = new Timeout(5, SECONDS);
|
||||
Coordinated coordinated = new Coordinated(timeout);
|
||||
counter.tell(coordinated.coordinate(new Increment()));
|
||||
counter.tell(coordinated.coordinate(new Increment()), null);
|
||||
coordinated.await();
|
||||
|
||||
Integer count = (Integer) Await.result(ask(counter, "GetCount", timeout), timeout.duration());
|
||||
|
|
@ -86,7 +85,7 @@ public class TransactorDocTest {
|
|||
|
||||
Timeout timeout = new Timeout(5, SECONDS);
|
||||
Coordinated coordinated = new Coordinated(timeout);
|
||||
friendlyCounter.tell(coordinated.coordinate(new Increment(friend)));
|
||||
friendlyCounter.tell(coordinated.coordinate(new Increment(friend)), null);
|
||||
coordinated.await();
|
||||
|
||||
Integer count1 = (Integer) Await.result(ask(friendlyCounter, "GetCount", timeout), timeout.duration());
|
||||
|
|
|
|||
|
|
@ -46,7 +46,6 @@ import com.typesafe.config.ConfigFactory;
|
|||
import java.lang.management.MemoryMXBean;
|
||||
import java.lang.management.MemoryUsage;
|
||||
import java.lang.management.OperatingSystemMXBean;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.Date;
|
||||
import java.text.SimpleDateFormat;
|
||||
|
||||
|
|
@ -58,8 +57,6 @@ import org.junit.Before;
|
|||
import org.junit.Test;
|
||||
import org.junit.Assume;
|
||||
|
||||
import akka.zeromq.SocketType;
|
||||
|
||||
public class ZeromqDocTestBase {
|
||||
|
||||
ActorSystem system;
|
||||
|
|
@ -95,12 +92,12 @@ public class ZeromqDocTestBase {
|
|||
//#sub-topic-socket
|
||||
|
||||
//#unsub-topic-socket
|
||||
subTopicSocket.tell(new Unsubscribe("foo.bar"));
|
||||
subTopicSocket.tell(new Unsubscribe("foo.bar"), null);
|
||||
//#unsub-topic-socket
|
||||
|
||||
byte[] payload = new byte[0];
|
||||
//#pub-topic
|
||||
pubSocket.tell(new ZMQMessage(new Frame("foo.bar"), new Frame(payload)));
|
||||
pubSocket.tell(new ZMQMessage(new Frame("foo.bar"), new Frame(payload)), null);
|
||||
//#pub-topic
|
||||
|
||||
//#high-watermark
|
||||
|
|
@ -205,12 +202,12 @@ public class ZeromqDocTestBase {
|
|||
byte[] heapPayload = ser.serializerFor(Heap.class).toBinary(
|
||||
new Heap(timestamp, currentHeap.getUsed(), currentHeap.getMax()));
|
||||
// the first frame is the topic, second is the message
|
||||
pubSocket.tell(new ZMQMessage(new Frame("health.heap"), new Frame(heapPayload)));
|
||||
pubSocket.tell(new ZMQMessage(new Frame("health.heap"), new Frame(heapPayload)), getSelf());
|
||||
|
||||
// use akka SerializationExtension to convert to bytes
|
||||
byte[] loadPayload = ser.serializerFor(Load.class).toBinary(new Load(timestamp, os.getSystemLoadAverage()));
|
||||
// the first frame is the topic, second is the message
|
||||
pubSocket.tell(new ZMQMessage(new Frame("health.load"), new Frame(loadPayload)));
|
||||
pubSocket.tell(new ZMQMessage(new Frame("health.load"), new Frame(loadPayload)), getSelf());
|
||||
} else {
|
||||
unhandled(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,8 +18,6 @@ import com.typesafe.config.ConfigFactory;
|
|||
import akka.actor.ActorSystem;
|
||||
import akka.actor.UntypedActor;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
public class DurableMailboxDocTestBase {
|
||||
|
||||
ActorSystem system;
|
||||
|
|
@ -41,7 +39,7 @@ public class DurableMailboxDocTestBase {
|
|||
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class).
|
||||
withDispatcher("my-dispatcher"), "myactor");
|
||||
//#dispatcher-config-use
|
||||
myActor.tell("test");
|
||||
myActor.tell("test", null);
|
||||
}
|
||||
|
||||
public static class MyUntypedActor extends UntypedActor {
|
||||
|
|
|
|||
|
|
@ -29,7 +29,7 @@ class PrintlnActor extends Actor {
|
|||
//#fibonacciActor
|
||||
class FibonacciActor extends Actor {
|
||||
def receive = {
|
||||
case FibonacciNumber(nbr) ⇒ sender tell fibonacci(nbr)
|
||||
case FibonacciNumber(nbr) ⇒ sender ! fibonacci(nbr)
|
||||
}
|
||||
|
||||
private def fibonacci(n: Int): Int = {
|
||||
|
|
|
|||
|
|
@ -13,28 +13,31 @@ public class HelloKernel implements Bootable {
|
|||
final ActorSystem system = ActorSystem.create("hellokernel");
|
||||
|
||||
static class HelloActor extends UntypedActor {
|
||||
final ActorRef worldActor =
|
||||
getContext().actorOf(new Props(WorldActor.class));
|
||||
final ActorRef worldActor = getContext().actorOf(
|
||||
new Props(WorldActor.class));
|
||||
|
||||
public void onReceive(Object message) {
|
||||
if (message == "start")
|
||||
worldActor.tell("Hello");
|
||||
worldActor.tell("Hello", getSelf());
|
||||
else if (message instanceof String)
|
||||
System.out.println("Received message '%s'".format((String)message));
|
||||
else unhandled(message);
|
||||
System.out.println(String.format("Received message '%s'", message));
|
||||
else
|
||||
unhandled(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static class WorldActor extends UntypedActor {
|
||||
static class WorldActor extends UntypedActor {
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof String)
|
||||
getSender().tell(((String)message).toUpperCase() + " world!");
|
||||
else unhandled(message);
|
||||
getSender().tell(((String) message).toUpperCase() + " world!",
|
||||
getSelf());
|
||||
else
|
||||
unhandled(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void startup() {
|
||||
system.actorOf(new Props(HelloActor.class)).tell("start");
|
||||
system.actorOf(new Props(HelloActor.class)).tell("start", null);
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
|
|
|
|||
|
|
@ -12,17 +12,23 @@ public class JAdvancedCalculatorActor extends UntypedActor {
|
|||
|
||||
if (message instanceof Op.Multiply) {
|
||||
Op.Multiply multiply = (Op.Multiply) message;
|
||||
System.out.println("Calculating " + multiply.getN1() + " * " + multiply.getN2());
|
||||
getSender().tell(new Op.MultiplicationResult(multiply.getN1(), multiply.getN2(), multiply.getN1() * multiply.getN2()));
|
||||
System.out.println("Calculating " + multiply.getN1() + " * "
|
||||
+ multiply.getN2());
|
||||
getSender().tell(
|
||||
new Op.MultiplicationResult(multiply.getN1(), multiply.getN2(),
|
||||
multiply.getN1() * multiply.getN2()), getSelf());
|
||||
|
||||
} else if (message instanceof Op.Divide) {
|
||||
Op.Divide divide = (Op.Divide) message;
|
||||
System.out.println("Calculating " + divide.getN1() + " / " + divide.getN2());
|
||||
getSender().tell(new Op.DivisionResult(divide.getN1(), divide.getN2(), divide.getN1() / divide.getN2()));
|
||||
System.out.println("Calculating " + divide.getN1() + " / "
|
||||
+ divide.getN2());
|
||||
getSender().tell(
|
||||
new Op.DivisionResult(divide.getN1(), divide.getN2(), divide.getN1()
|
||||
/ divide.getN2()), getSelf());
|
||||
|
||||
} else {
|
||||
unhandled(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
//#actor
|
||||
// #actor
|
||||
|
|
|
|||
|
|
@ -14,8 +14,10 @@ public class JCalculatorApplication implements Bootable {
|
|||
private ActorSystem system;
|
||||
|
||||
public JCalculatorApplication() {
|
||||
system = ActorSystem.create("CalculatorApplication", ConfigFactory.load().getConfig("calculator"));
|
||||
ActorRef actor = system.actorOf(new Props(JSimpleCalculatorActor.class), "simpleCalculator");
|
||||
system = ActorSystem.create("CalculatorApplication", ConfigFactory.load()
|
||||
.getConfig("calculator"));
|
||||
ActorRef actor = system.actorOf(new Props(JSimpleCalculatorActor.class),
|
||||
"simpleCalculator");
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -27,4 +29,4 @@ public class JCalculatorApplication implements Bootable {
|
|||
system.shutdown();
|
||||
}
|
||||
}
|
||||
//#setup
|
||||
// #setup
|
||||
|
|
@ -16,13 +16,15 @@ public class JCreationApplication implements Bootable {
|
|||
private ActorRef remoteActor;
|
||||
|
||||
public JCreationApplication() {
|
||||
system = ActorSystem.create("CreationApplication", ConfigFactory.load().getConfig("remotecreation"));
|
||||
system = ActorSystem.create("CreationApplication", ConfigFactory.load()
|
||||
.getConfig("remotecreation"));
|
||||
actor = system.actorOf(new Props(JCreationActor.class));
|
||||
remoteActor = system.actorOf(new Props(JAdvancedCalculatorActor.class), "advancedCalculator");
|
||||
remoteActor = system.actorOf(new Props(JAdvancedCalculatorActor.class),
|
||||
"advancedCalculator");
|
||||
}
|
||||
|
||||
public void doSomething(Op.MathOp mathOp) {
|
||||
actor.tell(new InternalMsg.MathOpMsg(remoteActor, mathOp));
|
||||
actor.tell(new InternalMsg.MathOpMsg(remoteActor, mathOp), null);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -34,4 +36,4 @@ public class JCreationApplication implements Bootable {
|
|||
system.shutdown();
|
||||
}
|
||||
}
|
||||
//#setup
|
||||
// #setup
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ package sample.remote.calculator.java;
|
|||
import akka.actor.ActorRef;
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.actor.Props;
|
||||
import akka.actor.UntypedActor;
|
||||
import akka.kernel.Bootable;
|
||||
import com.typesafe.config.ConfigFactory;
|
||||
//#imports
|
||||
|
|
@ -19,13 +18,15 @@ public class JLookupApplication implements Bootable {
|
|||
private ActorRef remoteActor;
|
||||
|
||||
public JLookupApplication() {
|
||||
system = ActorSystem.create("LookupApplication", ConfigFactory.load().getConfig("remotelookup"));
|
||||
system = ActorSystem.create("LookupApplication", ConfigFactory.load()
|
||||
.getConfig("remotelookup"));
|
||||
actor = system.actorOf(new Props(JLookupActor.class));
|
||||
remoteActor = system.actorFor("akka://CalculatorApplication@127.0.0.1:2552/user/simpleCalculator");
|
||||
remoteActor = system
|
||||
.actorFor("akka://CalculatorApplication@127.0.0.1:2552/user/simpleCalculator");
|
||||
}
|
||||
|
||||
public void doSomething(Op.MathOp mathOp) {
|
||||
actor.tell(new InternalMsg.MathOpMsg(remoteActor, mathOp));
|
||||
actor.tell(new InternalMsg.MathOpMsg(remoteActor, mathOp), null);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -37,4 +38,4 @@ public class JLookupApplication implements Bootable {
|
|||
system.shutdown();
|
||||
}
|
||||
}
|
||||
//#setup
|
||||
// #setup
|
||||
|
|
|
|||
|
|
@ -13,16 +13,22 @@ public class JSimpleCalculatorActor extends UntypedActor {
|
|||
if (message instanceof Op.Add) {
|
||||
Op.Add add = (Op.Add) message;
|
||||
System.out.println("Calculating " + add.getN1() + " + " + add.getN2());
|
||||
getSender().tell(new Op.AddResult(add.getN1(), add.getN2(), add.getN1() + add.getN2()));
|
||||
getSender()
|
||||
.tell(
|
||||
new Op.AddResult(add.getN1(), add.getN2(), add.getN1()
|
||||
+ add.getN2()), getSelf());
|
||||
|
||||
} else if (message instanceof Op.Subtract) {
|
||||
Op.Subtract subtract = (Op.Subtract) message;
|
||||
System.out.println("Calculating " + subtract.getN1() + " - " + subtract.getN2());
|
||||
getSender().tell(new Op.SubtractResult(subtract.getN1(), subtract.getN2(), subtract.getN1() - subtract.getN2()));
|
||||
System.out.println("Calculating " + subtract.getN1() + " - "
|
||||
+ subtract.getN2());
|
||||
getSender().tell(
|
||||
new Op.SubtractResult(subtract.getN1(), subtract.getN2(),
|
||||
subtract.getN1() - subtract.getN2()), getSelf());
|
||||
|
||||
} else {
|
||||
unhandled(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
//#actor
|
||||
// #actor
|
||||
|
|
|
|||
|
|
@ -7,11 +7,14 @@ import java.io.Serializable;
|
|||
|
||||
public class Op {
|
||||
|
||||
public interface MathOp extends Serializable {}
|
||||
public interface MathOp extends Serializable {
|
||||
}
|
||||
|
||||
public interface MathResult extends Serializable {}
|
||||
public interface MathResult extends Serializable {
|
||||
}
|
||||
|
||||
static class Add implements MathOp {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final int n1;
|
||||
private final int n2;
|
||||
|
||||
|
|
@ -30,6 +33,7 @@ public class Op {
|
|||
}
|
||||
|
||||
static class AddResult implements MathResult {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final int n1;
|
||||
private final int n2;
|
||||
private final int result;
|
||||
|
|
@ -54,6 +58,7 @@ public class Op {
|
|||
}
|
||||
|
||||
static class Subtract implements MathOp {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final int n1;
|
||||
private final int n2;
|
||||
|
||||
|
|
@ -72,6 +77,7 @@ public class Op {
|
|||
}
|
||||
|
||||
static class SubtractResult implements MathResult {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final int n1;
|
||||
private final int n2;
|
||||
private final int result;
|
||||
|
|
@ -96,6 +102,7 @@ public class Op {
|
|||
}
|
||||
|
||||
static class Multiply implements MathOp {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final int n1;
|
||||
private final int n2;
|
||||
|
||||
|
|
@ -114,6 +121,7 @@ public class Op {
|
|||
}
|
||||
|
||||
static class MultiplicationResult implements MathResult {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final int n1;
|
||||
private final int n2;
|
||||
private final int result;
|
||||
|
|
@ -138,6 +146,7 @@ public class Op {
|
|||
}
|
||||
|
||||
static class Divide implements MathOp {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final double n1;
|
||||
private final int n2;
|
||||
|
||||
|
|
@ -156,6 +165,7 @@ public class Op {
|
|||
}
|
||||
|
||||
static class DivisionResult implements MathResult {
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final double n1;
|
||||
private final int n2;
|
||||
private final double result;
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ abstract class UntypedTransactor extends UntypedActor {
|
|||
case coordinated @ Coordinated(message) ⇒ {
|
||||
val others = coordinate(message)
|
||||
for (sendTo ← others) {
|
||||
sendTo.actor.tell(coordinated(sendTo.message.getOrElse(message)))
|
||||
sendTo.actor ! coordinated(sendTo.message.getOrElse(message))
|
||||
}
|
||||
before(message)
|
||||
coordinated.atomic { txn ⇒ atomically(message) }
|
||||
|
|
|
|||
|
|
@ -4,13 +4,13 @@
|
|||
|
||||
package akka.transactor;
|
||||
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.UntypedActor;
|
||||
import scala.concurrent.stm.Ref;
|
||||
import scala.concurrent.stm.japi.STM;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import scala.concurrent.stm.Ref;
|
||||
import scala.concurrent.stm.japi.STM;
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.UntypedActor;
|
||||
|
||||
public class UntypedCoordinatedCounter extends UntypedActor {
|
||||
private String name;
|
||||
|
|
@ -35,7 +35,7 @@ public class UntypedCoordinatedCounter extends UntypedActor {
|
|||
};
|
||||
if (!friends.isEmpty()) {
|
||||
Increment coordMessage = new Increment(friends.subList(1, friends.size()), latch);
|
||||
friends.get(0).tell(coordinated.coordinate(coordMessage));
|
||||
friends.get(0).tell(coordinated.coordinate(coordMessage), getSelf());
|
||||
}
|
||||
coordinated.atomic(new Runnable() {
|
||||
public void run() {
|
||||
|
|
@ -45,7 +45,7 @@ public class UntypedCoordinatedCounter extends UntypedActor {
|
|||
});
|
||||
}
|
||||
} else if ("GetCount".equals(incoming)) {
|
||||
getSender().tell(count.get());
|
||||
getSender().tell(count.get(), getSelf());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -75,7 +75,7 @@ public class UntypedCoordinatedIncrementTest {
|
|||
public void incrementAllCountersWithSuccessfulTransaction() throws Exception {
|
||||
CountDownLatch incrementLatch = new CountDownLatch(numCounters);
|
||||
Increment message = new Increment(counters.subList(1, counters.size()), incrementLatch);
|
||||
counters.get(0).tell(new Coordinated(message, timeout));
|
||||
counters.get(0).tell(new Coordinated(message, timeout), null);
|
||||
try {
|
||||
incrementLatch.await(timeoutSeconds, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException exception) {
|
||||
|
|
@ -97,7 +97,7 @@ public class UntypedCoordinatedIncrementTest {
|
|||
List<ActorRef> actors = new ArrayList<ActorRef>(counters);
|
||||
actors.add(failer);
|
||||
Increment message = new Increment(actors.subList(1, actors.size()), incrementLatch);
|
||||
actors.get(0).tell(new Coordinated(message, timeout));
|
||||
actors.get(0).tell(new Coordinated(message, timeout), null);
|
||||
try {
|
||||
incrementLatch.await(timeoutSeconds, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException exception) {
|
||||
|
|
|
|||
|
|
@ -4,15 +4,12 @@
|
|||
|
||||
package akka.transactor;
|
||||
|
||||
import akka.actor.ActorRef;
|
||||
import akka.transactor.UntypedTransactor;
|
||||
import akka.transactor.SendTo;
|
||||
import scala.concurrent.stm.Ref;
|
||||
import scala.concurrent.stm.japi.STM;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import scala.concurrent.stm.Ref;
|
||||
import scala.concurrent.stm.japi.STM;
|
||||
import akka.actor.ActorRef;
|
||||
|
||||
public class UntypedCounter extends UntypedTransactor {
|
||||
private String name;
|
||||
|
|
@ -52,7 +49,7 @@ public class UntypedCounter extends UntypedTransactor {
|
|||
|
||||
@Override public boolean normally(Object message) {
|
||||
if ("GetCount".equals(message)) {
|
||||
getSender().tell(count.get());
|
||||
getSender().tell(count.get(), getSelf());
|
||||
return true;
|
||||
} else return false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,8 +4,6 @@
|
|||
|
||||
package akka.transactor;
|
||||
|
||||
import scala.concurrent.stm.InTxn;
|
||||
|
||||
public class UntypedFailer extends UntypedTransactor {
|
||||
public void atomically(Object message) throws Exception {
|
||||
throw new ExpectedFailureException();
|
||||
|
|
|
|||
|
|
@ -63,6 +63,8 @@ public class UntypedTransactorTest {
|
|||
for (int i = 1; i <= numCounters; i++) {
|
||||
final String name = "counter" + i;
|
||||
ActorRef counter = system.actorOf(new Props(new UntypedActorFactory() {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
public UntypedActor create() {
|
||||
return new UntypedCounter(name);
|
||||
}
|
||||
|
|
@ -75,8 +77,9 @@ public class UntypedTransactorTest {
|
|||
@Test
|
||||
public void incrementAllCountersWithSuccessfulTransaction() throws Exception {
|
||||
CountDownLatch incrementLatch = new CountDownLatch(numCounters);
|
||||
Increment message = new Increment(counters.subList(1, counters.size()), incrementLatch);
|
||||
counters.get(0).tell(message);
|
||||
Increment message = new Increment(counters.subList(1, counters.size()),
|
||||
incrementLatch);
|
||||
counters.get(0).tell(message, null);
|
||||
try {
|
||||
incrementLatch.await(timeoutSeconds, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException exception) {
|
||||
|
|
@ -90,15 +93,19 @@ public class UntypedTransactorTest {
|
|||
|
||||
@Test
|
||||
public void incrementNoCountersWithFailingTransaction() throws Exception {
|
||||
EventFilter expectedFailureFilter = (EventFilter) new ErrorFilter(ExpectedFailureException.class);
|
||||
EventFilter coordinatedFilter = (EventFilter) new ErrorFilter(CoordinatedTransactionException.class);
|
||||
Seq<EventFilter> ignoreExceptions = seq(expectedFailureFilter, coordinatedFilter);
|
||||
EventFilter expectedFailureFilter = (EventFilter) new ErrorFilter(
|
||||
ExpectedFailureException.class);
|
||||
EventFilter coordinatedFilter = (EventFilter) new ErrorFilter(
|
||||
CoordinatedTransactionException.class);
|
||||
Seq<EventFilter> ignoreExceptions = seq(expectedFailureFilter,
|
||||
coordinatedFilter);
|
||||
system.eventStream().publish(new TestEvent.Mute(ignoreExceptions));
|
||||
CountDownLatch incrementLatch = new CountDownLatch(numCounters);
|
||||
List<ActorRef> actors = new ArrayList<ActorRef>(counters);
|
||||
actors.add(failer);
|
||||
Increment message = new Increment(actors.subList(1, actors.size()), incrementLatch);
|
||||
actors.get(0).tell(message);
|
||||
Increment message = new Increment(actors.subList(1, actors.size()),
|
||||
incrementLatch);
|
||||
actors.get(0).tell(message, null);
|
||||
try {
|
||||
incrementLatch.await(timeoutSeconds, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException exception) {
|
||||
|
|
@ -111,6 +118,8 @@ public class UntypedTransactorTest {
|
|||
}
|
||||
|
||||
public <A> Seq<A> seq(A... args) {
|
||||
return JavaConverters.collectionAsScalaIterableConverter(Arrays.asList(args)).asScala().toSeq();
|
||||
return JavaConverters
|
||||
.collectionAsScalaIterableConverter(Arrays.asList(args)).asScala()
|
||||
.toSeq();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue