add preprocessor for RST docs, see #2461 and #2431

The idea is to filter the sources, replacing @<var>@ occurrences with
the mapping for <var> (which is currently hard-coded). @@ -> @. In order
to make this work, I had to move the doc sources one directory down
(into akka-docs/rst) so that the filtered result could be in a sibling
directory so that relative links (to _sphinx plugins or real code) would
continue to work.

While I was at it I also changed it so that WARNINGs and ERRORs are not
swallowed into the debug dump anymore but printed at [warn] level
(minimum).

One piece of fallout is that the (online) html build is now run after
the normal one, not in parallel.
This commit is contained in:
Roland 2012-09-21 10:47:58 +02:00
parent c0f60da8cc
commit 9bc01ae265
266 changed files with 270 additions and 182 deletions

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import org.scalatest.junit.JUnitSuite
class FSMDocTest extends FSMDocTestBase with JUnitSuite

View file

@ -0,0 +1,194 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
//#imports-data
import java.util.ArrayList;
import java.util.List;
import akka.actor.ActorRef;
//#imports-data
//#imports-actor
import akka.event.LoggingAdapter;
import akka.event.Logging;
import akka.actor.UntypedActor;
//#imports-actor
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.testkit.TestProbe;
import akka.testkit.AkkaSpec;
public class FSMDocTestBase {
//#data
public static final class SetTarget {
final ActorRef ref;
public SetTarget(ActorRef ref) {
this.ref = ref;
}
}
public static final class Queue {
final Object o;
public Queue(Object o) {
this.o = o;
}
}
public static final Object flush = new Object();
public static final class Batch {
final List<Object> objects;
public Batch(List<Object> objects) {
this.objects = objects;
}
}
//#data
//#base
static abstract class MyFSMBase extends UntypedActor {
/*
* This is the mutable state of this state machine.
*/
protected enum State {
IDLE, ACTIVE;
}
private State state = State.IDLE;
private ActorRef target;
private List<Object> queue;
/*
* Then come all the mutator methods:
*/
protected void init(ActorRef target) {
this.target = target;
queue = new ArrayList<Object>();
}
protected void setState(State s) {
if (state != s) {
transition(state, s);
state = s;
}
}
protected void enqueue(Object o) {
if (queue != null)
queue.add(o);
}
protected List<Object> drainQueue() {
final List<Object> q = queue;
if (q == null)
throw new IllegalStateException("drainQueue(): not yet initialized");
queue = new ArrayList<Object>();
return q;
}
/*
* Here are the interrogation methods:
*/
protected boolean isInitialized() {
return target != null;
}
protected State getState() {
return state;
}
protected ActorRef getTarget() {
if (target == null)
throw new IllegalStateException("getTarget(): not yet initialized");
return target;
}
/*
* And finally the callbacks (only one in this example: react to state change)
*/
abstract protected void transition(State old, State next);
}
//#base
//#actor
static public class MyFSM extends MyFSMBase {
private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
@Override
public void onReceive(Object o) {
if (getState() == State.IDLE) {
if (o instanceof SetTarget)
init(((SetTarget) o).ref);
else
whenUnhandled(o);
} else if (getState() == State.ACTIVE) {
if (o == flush)
setState(State.IDLE);
else
whenUnhandled(o);
}
}
@Override
public void transition(State old, State next) {
if (old == State.ACTIVE) {
getTarget().tell(new Batch(drainQueue()), getSelf());
}
}
private void whenUnhandled(Object o) {
if (o instanceof Queue && isInitialized()) {
enqueue(((Queue) o).o);
setState(State.ACTIVE);
} else {
log.warning("received unknown message {} in state {}", o, getState());
}
}
}
//#actor
ActorSystem system;
@org.junit.Before
public void setUp() {
system = ActorSystem.create("FSMSystem", AkkaSpec.testConf());
}
@org.junit.Test
public void mustBunch() {
final ActorRef buncher = system.actorOf(new Props(MyFSM.class));
final TestProbe probe = new TestProbe(system);
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);
assert b.objects.contains(2);
}
@org.junit.After
public void cleanup() {
system.shutdown();
}
}

View file

@ -0,0 +1,7 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import org.scalatest.junit.JUnitSuite
class FaultHandlingTest extends FaultHandlingTestBase with JUnitSuite

View file

@ -0,0 +1,214 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
//#testkit
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.SupervisorStrategy;
import static akka.actor.SupervisorStrategy.*;
import akka.actor.OneForOneStrategy;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import scala.concurrent.Await;
import static akka.pattern.Patterns.ask;
import scala.concurrent.util.Duration;
import akka.testkit.AkkaSpec;
import akka.testkit.TestProbe;
//#testkit
import akka.testkit.ErrorFilter;
import akka.testkit.EventFilter;
import akka.testkit.TestEvent;
import static java.util.concurrent.TimeUnit.SECONDS;
import akka.japi.Function;
import scala.Option;
import scala.collection.JavaConverters;
import scala.collection.Seq;
import org.junit.Test;
import org.junit.BeforeClass;
import org.junit.AfterClass;
//#testkit
public class FaultHandlingTestBase {
//#testkit
//#supervisor
static public class Supervisor extends UntypedActor {
//#strategy
private static SupervisorStrategy strategy = new OneForOneStrategy(10, Duration.parse("1 minute"),
new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof ArithmeticException) {
return resume();
} else if (t instanceof NullPointerException) {
return restart();
} else if (t instanceof IllegalArgumentException) {
return stop();
} else {
return escalate();
}
}
});
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
//#strategy
public void onReceive(Object o) {
if (o instanceof Props) {
getSender().tell(getContext().actorOf((Props) o), getSelf());
} else {
unhandled(o);
}
}
}
//#supervisor
//#supervisor2
static public class Supervisor2 extends UntypedActor {
//#strategy2
private static SupervisorStrategy strategy = new OneForOneStrategy(10, Duration.parse("1 minute"),
new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof ArithmeticException) {
return resume();
} else if (t instanceof NullPointerException) {
return restart();
} else if (t instanceof IllegalArgumentException) {
return stop();
} else {
return escalate();
}
}
});
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
//#strategy2
public void onReceive(Object o) {
if (o instanceof Props) {
getSender().tell(getContext().actorOf((Props) o), getSelf());
} else {
unhandled(o);
}
}
@Override
public void preRestart(Throwable cause, Option<Object> msg) {
// do not kill all children, which is the default here
}
}
//#supervisor2
//#child
static public class Child extends UntypedActor {
int state = 0;
public void onReceive(Object o) throws Exception {
if (o instanceof Exception) {
throw (Exception) o;
} else if (o instanceof Integer) {
state = (Integer) o;
} else if (o.equals("get")) {
getSender().tell(state, getSelf());
} else {
unhandled(o);
}
}
}
//#child
//#testkit
static ActorSystem system;
Duration timeout = Duration.create(5, SECONDS);
@BeforeClass
public static void start() {
system = ActorSystem.create("test", AkkaSpec.testConf());
}
@AfterClass
public static void cleanup() {
system.shutdown();
}
@Test
public void mustEmploySupervisorStrategy() throws Exception {
// code here
//#testkit
EventFilter ex1 = (EventFilter) new ErrorFilter(ArithmeticException.class);
EventFilter ex2 = (EventFilter) new ErrorFilter(NullPointerException.class);
EventFilter ex3 = (EventFilter) new ErrorFilter(IllegalArgumentException.class);
EventFilter ex4 = (EventFilter) new ErrorFilter(Exception.class);
Seq<EventFilter> ignoreExceptions = seq(ex1, ex2, ex3, ex4);
system.eventStream().publish(new TestEvent.Mute(ignoreExceptions));
//#create
Props superprops = new Props(Supervisor.class);
ActorRef supervisor = system.actorOf(superprops, "supervisor");
ActorRef child = (ActorRef) Await.result(ask(supervisor, new Props(Child.class), 5000), timeout);
//#create
//#resume
child.tell(42, null);
assert Await.result(ask(child, "get", 5000), timeout).equals(42);
child.tell(new ArithmeticException(), null);
assert Await.result(ask(child, "get", 5000), timeout).equals(42);
//#resume
//#restart
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(), null);
probe.expectMsgClass(Terminated.class);
//#stop
//#escalate-kill
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(), null);
probe.expectMsgClass(Terminated.class);
//#escalate-kill
//#escalate-restart
superprops = new Props(Supervisor2.class);
supervisor = system.actorOf(superprops);
child = (ActorRef) Await.result(ask(supervisor, new Props(Child.class), 5000), timeout);
child.tell(23, null);
assert Await.result(ask(child, "get", 5000), timeout).equals(23);
child.tell(new Exception(), null);
assert Await.result(ask(child, "get", 5000), timeout).equals(0);
//#escalate-restart
//#testkit
}
//#testkit
public <A> Seq<A> seq(A... args) {
return JavaConverters.collectionAsScalaIterableConverter(java.util.Arrays.asList(args)).asScala().toSeq();
}
//#testkit
}
//#testkit

View file

@ -0,0 +1,21 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.PoisonPill;
import akka.actor.UntypedActor;
//#context-actorOf
public class FirstUntypedActor extends UntypedActor {
ActorRef myActor = getContext().actorOf(new Props(MyActor.class), "myactor");
//#context-actorOf
public void onReceive(Object message) {
myActor.forward(message, getContext());
myActor.tell(PoisonPill.getInstance(), null);
}
}

View file

@ -0,0 +1,28 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
//#immutable-message
public class ImmutableMessage {
private final int sequenceNumber;
private final List<String> values;
public ImmutableMessage(int sequenceNumber, List<String> values) {
this.sequenceNumber = sequenceNumber;
this.values = Collections.unmodifiableList(new ArrayList<String>(values));
}
public int getSequenceNumber() {
return sequenceNumber;
}
public List<String> getValues() {
return values;
}
}
//#immutable-message

View file

@ -0,0 +1,27 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
//#receive-timeout
import akka.actor.ReceiveTimeout;
import akka.actor.UntypedActor;
import scala.concurrent.util.Duration;
public class MyReceivedTimeoutUntypedActor extends UntypedActor {
public MyReceivedTimeoutUntypedActor() {
getContext().setReceiveTimeout(Duration.parse("30 seconds"));
}
public void onReceive(Object message) {
if (message.equals("Hello")) {
getSender().tell("Hello world", getSelf());
} else if (message == ReceiveTimeout.getInstance()) {
throw new RuntimeException("received timeout");
} else {
unhandled(message);
}
}
}
//#receive-timeout

View file

@ -0,0 +1,22 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
//#my-untyped-actor
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
public class MyUntypedActor extends UntypedActor {
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
public void onReceive(Object message) throws Exception {
if (message instanceof String)
log.info("Received String message: {}", message);
else
unhandled(message);
}
}
//#my-untyped-actor

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import org.scalatest.junit.JUnitSuite
class SchedulerDocTest extends SchedulerDocTestBase with JUnitSuite

View file

@ -0,0 +1,89 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
//#imports1
import akka.actor.Props;
import scala.concurrent.util.Duration;
import java.util.concurrent.TimeUnit;
//#imports1
//#imports2
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
import akka.actor.Cancellable;
//#imports2
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.testkit.AkkaSpec;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
public class SchedulerDocTestBase {
ActorSystem system;
ActorRef testActor;
@Before
public void setUp() {
system = ActorSystem.create("MySystem", AkkaSpec.testConf());
testActor = system.actorOf(new Props(MyUntypedActor.class));
}
@After
public void tearDown() {
system.shutdown();
}
@Test
public void scheduleOneOffTask() {
//#schedule-one-off-message
//Schedules to send the "foo"-message to the testActor after 50ms
system.scheduler().scheduleOnce(Duration.create(50, TimeUnit.MILLISECONDS), testActor, "foo", system.dispatcher());
//#schedule-one-off-message
//#schedule-one-off-thunk
//Schedules a Runnable to be executed (send the current time) to the testActor after 50ms
system.scheduler().scheduleOnce(Duration.create(50, TimeUnit.MILLISECONDS), new Runnable() {
@Override
public void run() {
testActor.tell(System.currentTimeMillis(), null);
}
}, system.dispatcher());
//#schedule-one-off-thunk
}
@Test
public void scheduleRecurringTask() {
//#schedule-recurring
ActorRef tickActor = system.actorOf(new Props().withCreator(new UntypedActorFactory() {
public UntypedActor create() {
return new UntypedActor() {
public void onReceive(Object message) {
if (message.equals("Tick")) {
// Do someting
} else {
unhandled(message);
}
}
};
}
}));
//This will schedule to send the Tick-message
//to the tickActor after 0ms repeating every 50ms
Cancellable cancellable = system.scheduler().schedule(Duration.Zero(), Duration.create(50, TimeUnit.MILLISECONDS),
tickActor, "Tick", system.dispatcher());
//This cancels further Ticks to be sent
cancellable.cancel();
//#schedule-recurring
system.stop(tickActor);
}
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import org.scalatest.junit.JUnitSuite
class TypedActorDocTest extends TypedActorDocTestBase with JUnitSuite

View file

@ -0,0 +1,187 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
//#imports
import akka.actor.TypedActor;
import akka.actor.*;
import akka.japi.*;
import akka.dispatch.Futures;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.util.Duration;
import java.util.concurrent.TimeUnit;
//#imports
import java.lang.Exception;
import org.junit.Test;
import static org.junit.Assert.*;
public class TypedActorDocTestBase {
Object someReference = null;
ActorSystem system = null;
//#typed-actor-iface
public static interface Squarer {
//#typed-actor-iface-methods
void squareDontCare(int i); //fire-forget
Future<Integer> square(int i); //non-blocking send-request-reply
Option<Integer> squareNowPlease(int i);//blocking send-request-reply
int squareNow(int i); //blocking send-request-reply
//#typed-actor-iface-methods
}
//#typed-actor-iface
//#typed-actor-impl
static class SquarerImpl implements Squarer {
private String name;
public SquarerImpl() {
this.name = "default";
}
public SquarerImpl(String name) {
this.name = name;
}
//#typed-actor-impl-methods
public void squareDontCare(int i) {
int sq = i * i; //Nobody cares :(
}
public Future<Integer> square(int i) {
return Futures.successful(i * i);
}
public Option<Integer> squareNowPlease(int i) {
return Option.some(i * i);
}
public int squareNow(int i) {
return i * i;
}
//#typed-actor-impl-methods
}
//#typed-actor-impl
@Test public void mustGetTheTypedActorExtension() {
try {
//#typed-actor-extension-tools
//Returns the Typed Actor Extension
TypedActorExtension extension =
TypedActor.get(system); //system is an instance of ActorSystem
//Returns whether the reference is a Typed Actor Proxy or not
TypedActor.get(system).isTypedActor(someReference);
//Returns the backing Akka Actor behind an external Typed Actor Proxy
TypedActor.get(system).getActorRefFor(someReference);
//Returns the current ActorContext,
// method only valid within methods of a TypedActor implementation
ActorContext context = TypedActor.context();
//Returns the external proxy of the current Typed Actor,
// method only valid within methods of a TypedActor implementation
Squarer sq = TypedActor.<Squarer>self();
//Returns a contextual instance of the Typed Actor Extension
//this means that if you create other Typed Actors with this,
//they will become children to the current Typed Actor.
TypedActor.get(TypedActor.context());
//#typed-actor-extension-tools
} catch (Exception e) {
//dun care
}
}
@Test public void createATypedActor() {
try {
//#typed-actor-create1
Squarer mySquarer =
TypedActor.get(system).typedActorOf(new TypedProps<SquarerImpl>(Squarer.class, SquarerImpl.class));
//#typed-actor-create1
//#typed-actor-create2
Squarer otherSquarer =
TypedActor.get(system).typedActorOf(new TypedProps<SquarerImpl>(Squarer.class,
new Creator<SquarerImpl>() {
public SquarerImpl create() { return new SquarerImpl("foo"); }
}),
"name");
//#typed-actor-create2
//#typed-actor-calls
//#typed-actor-call-oneway
mySquarer.squareDontCare(10);
//#typed-actor-call-oneway
//#typed-actor-call-future
Future<Integer> fSquare = mySquarer.square(10); //A Future[Int]
//#typed-actor-call-future
//#typed-actor-call-option
Option<Integer> oSquare = mySquarer.squareNowPlease(10); //Option[Int]
//#typed-actor-call-option
//#typed-actor-call-strict
int iSquare = mySquarer.squareNow(10); //Int
//#typed-actor-call-strict
//#typed-actor-calls
assertEquals(100, Await.result(fSquare, Duration.create(3, TimeUnit.SECONDS)).intValue());
assertEquals(100, oSquare.get().intValue());
assertEquals(100, iSquare);
//#typed-actor-stop
TypedActor.get(system).stop(mySquarer);
//#typed-actor-stop
//#typed-actor-poisonpill
TypedActor.get(system).poisonPill(otherSquarer);
//#typed-actor-poisonpill
} catch(Exception e) {
//Ignore
}
}
@Test public void createHierarchies() {
try {
//#typed-actor-hierarchy
Squarer childSquarer =
TypedActor.get(TypedActor.context()).
typedActorOf(
new TypedProps<SquarerImpl>(Squarer.class, SquarerImpl.class)
);
//Use "childSquarer" as a Squarer
//#typed-actor-hierarchy
} catch (Exception e) {
//dun care
}
}
@Test public void proxyAnyActorRef() {
try {
//#typed-actor-remote
Squarer typedActor =
TypedActor.get(system).
typedActorOf(
new TypedProps<Squarer>(Squarer.class),
system.actorFor("akka://SomeSystem@somehost:2552/user/some/foobar")
);
//Use "typedActor" as a FooBar
//#typed-actor-remote
} catch (Exception e) {
//dun care
}
}
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import org.scalatest.junit.JUnitSuite
class UntypedActorDocTest extends UntypedActorDocTestBase with JUnitSuite

View file

@ -0,0 +1,396 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
//#imports
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
//#imports
//#import-future
import scala.concurrent.Future;
import akka.dispatch.Futures;
import akka.dispatch.Mapper;
import scala.concurrent.Await;
import scala.concurrent.util.Duration;
import akka.util.Timeout;
//#import-future
//#import-actors
import akka.actor.PoisonPill;
import akka.actor.Kill;
//#import-actors
//#import-procedure
import akka.japi.Procedure;
//#import-procedure
//#import-watch
import akka.actor.Terminated;
//#import-watch
//#import-gracefulStop
import static akka.pattern.Patterns.gracefulStop;
import scala.concurrent.Future;
import scala.concurrent.Await;
import scala.concurrent.util.Duration;
import akka.pattern.AskTimeoutException;
//#import-gracefulStop
//#import-askPipe
import static akka.pattern.Patterns.ask;
import static akka.pattern.Patterns.pipe;
import scala.concurrent.Future;
import akka.dispatch.Futures;
import scala.concurrent.util.Duration;
import akka.util.Timeout;
import java.util.concurrent.TimeUnit;
import java.util.ArrayList;
//#import-askPipe
//#import-stash
import akka.actor.UntypedActorWithStash;
//#import-stash
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
import org.junit.Test;
import scala.Option;
import java.lang.Object;
import java.util.Iterator;
import akka.pattern.Patterns;
public class UntypedActorDocTestBase {
@Test
public void createProps() {
//#creating-props-config
Props props1 = new Props();
Props props2 = new Props(MyUntypedActor.class);
Props props3 = new Props(new UntypedActorFactory() {
public UntypedActor create() {
return new MyUntypedActor();
}
});
Props props4 = props1.withCreator(new UntypedActorFactory() {
public UntypedActor create() {
return new MyUntypedActor();
}
});
//#creating-props-config
}
@Test
public void systemActorOf() {
//#system-actorOf
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class), "myactor");
//#system-actorOf
myActor.tell("test", null);
system.shutdown();
}
@Test
public void contextActorOf() {
//#context-actorOf
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class), "myactor");
//#context-actorOf
myActor.tell("test", null);
system.shutdown();
}
@Test
public void constructorActorOf() {
ActorSystem system = ActorSystem.create("MySystem");
//#creating-constructor
// allows passing in arguments to the MyActor constructor
ActorRef myActor = system.actorOf(new Props(new UntypedActorFactory() {
public UntypedActor create() {
return new MyActor("...");
}
}), "myactor");
//#creating-constructor
myActor.tell("test", null);
system.shutdown();
}
@Test
public void propsActorOf() {
ActorSystem system = ActorSystem.create("MySystem");
//#creating-props
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class).withDispatcher("my-dispatcher"), "myactor");
//#creating-props
myActor.tell("test", null);
system.shutdown();
}
@Test
public void usingAsk() throws Exception {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(new UntypedActorFactory() {
public UntypedActor create() {
return new MyAskActor();
}
}), "myactor");
//#using-ask
Future<Object> future = Patterns.ask(myActor, "Hello", 1000);
Object result = Await.result(future, Duration.create(1, TimeUnit.SECONDS));
//#using-ask
system.shutdown();
}
@Test
public void receiveTimeout() {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(MyReceivedTimeoutUntypedActor.class));
myActor.tell("Hello", null);
system.shutdown();
}
@Test
public void usePoisonPill() {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class));
//#poison-pill
myActor.tell(PoisonPill.getInstance(), null);
//#poison-pill
system.shutdown();
}
@Test
public void useKill() {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef victim = system.actorOf(new Props(MyUntypedActor.class));
//#kill
victim.tell(Kill.getInstance(), null);
//#kill
system.shutdown();
}
@Test
public void useBecome() {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(new UntypedActorFactory() {
public UntypedActor create() {
return new HotSwapActor();
}
}));
myActor.tell("foo", null);
myActor.tell("bar", null);
myActor.tell("bar", null);
system.shutdown();
}
@Test
public void useWatch() throws Exception {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(WatchActor.class));
Future<Object> future = Patterns.ask(myActor, "kill", 1000);
assert Await.result(future, Duration.parse("1 second")).equals("finished");
system.shutdown();
}
@Test
public void usePatternsGracefulStop() throws Exception {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef actorRef = system.actorOf(new Props(MyUntypedActor.class));
//#gracefulStop
try {
Future<Boolean> stopped = gracefulStop(actorRef, Duration.create(5, TimeUnit.SECONDS), system);
Await.result(stopped, Duration.create(6, TimeUnit.SECONDS));
// the actor has been stopped
} catch (AskTimeoutException e) {
// the actor wasn't stopped within 5 seconds
}
//#gracefulStop
system.shutdown();
}
class Result {
final int x;
final String s;
public Result(int x, String s) {
this.x = x;
this.s = s;
}
}
@Test
public void usePatternsAskPipe() {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef actorA = system.actorOf(new Props(MyUntypedActor.class));
ActorRef actorB = system.actorOf(new Props(MyUntypedActor.class));
ActorRef actorC = system.actorOf(new Props(MyUntypedActor.class));
//#ask-pipe
final Timeout t = new Timeout(Duration.create(5, TimeUnit.SECONDS));
final ArrayList<Future<Object>> futures = new ArrayList<Future<Object>>();
futures.add(ask(actorA, "request", 1000)); // using 1000ms timeout
futures.add(ask(actorB, "another request", t)); // using timeout from above
final Future<Iterable<Object>> aggregate = Futures.sequence(futures, system.dispatcher());
final Future<Result> transformed = aggregate.map(new Mapper<Iterable<Object>, Result>() {
public Result apply(Iterable<Object> coll) {
final Iterator<Object> it = coll.iterator();
final String s = (String) it.next();
final int x = (Integer) it.next();
return new Result(x, s);
}
}, system.dispatcher());
pipe(transformed, system.dispatcher()).to(actorC);
//#ask-pipe
system.shutdown();
}
public static class MyActor extends UntypedActor {
public MyActor(String s) {
}
public void onReceive(Object message) throws Exception {
try {
operation();
} catch (Exception e) {
getSender().tell(new akka.actor.Status.Failure(e), getSelf());
throw e;
}
}
private void operation() {
}
//#lifecycle-callbacks
public void preStart() {
}
public void preRestart(Throwable reason, Option<Object> message) {
for (ActorRef each : getContext().getChildren())
getContext().stop(each);
postStop();
}
public void postRestart(Throwable reason) {
preStart();
}
public void postStop() {
}
//#lifecycle-callbacks
}
public static class MyAskActor extends UntypedActor {
public void onReceive(Object message) throws Exception {
//#reply-exception
try {
String result = operation();
getSender().tell(result, getSelf());
} catch (Exception e) {
getSender().tell(new akka.actor.Status.Failure(e), getSelf());
throw e;
}
//#reply-exception
}
private String operation() {
return "Hi";
}
}
//#hot-swap-actor
public static class HotSwapActor extends UntypedActor {
Procedure<Object> angry = new Procedure<Object>() {
@Override
public void apply(Object message) {
if (message.equals("bar")) {
getSender().tell("I am already angry?", getSelf());
} else if (message.equals("foo")) {
getContext().become(happy);
}
}
};
Procedure<Object> happy = new Procedure<Object>() {
@Override
public void apply(Object message) {
if (message.equals("bar")) {
getSender().tell("I am already happy :-)", getSelf());
} else if (message.equals("foo")) {
getContext().become(angry);
}
}
};
public void onReceive(Object message) {
if (message.equals("bar")) {
getContext().become(angry);
} else if (message.equals("foo")) {
getContext().become(happy);
} else {
unhandled(message);
}
}
}
//#hot-swap-actor
//#stash
public static class ActorWithProtocol extends UntypedActorWithStash {
private Boolean isOpen = false;
public void onReceive(Object msg) {
if (isOpen) {
if (msg.equals("write")) {
// do writing...
} else if (msg.equals("close")) {
unstashAll();
isOpen = false;
} else {
stash();
}
} else {
if (msg.equals("open")) {
unstashAll();
isOpen = true;
} else {
stash();
}
}
}
}
//#stash
//#watch
public static class WatchActor extends UntypedActor {
final ActorRef child = this.getContext().actorOf(Props.empty(), "child");
{
this.getContext().watch(child); // <-- this is the only call needed for registration
}
ActorRef lastSender = getContext().system().deadLetters();
@Override
public void onReceive(Object message) {
if (message.equals("kill")) {
getContext().stop(child);
lastSender = getSender();
} else if (message instanceof Terminated) {
final Terminated t = (Terminated) message;
if (t.getActor() == child) {
lastSender.tell("finished", getSelf());
}
} else {
unhandled(message);
}
}
}
//#watch
}

View file

@ -0,0 +1,56 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor;
import static docs.actor.UntypedActorSwapper.Swap.SWAP;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.ActorSystem;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.japi.Procedure;
//#swapper
public class UntypedActorSwapper {
public static class Swap {
public static Swap SWAP = new Swap();
private Swap() {
}
}
public static class Swapper extends UntypedActor {
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
public void onReceive(Object message) {
if (message == SWAP) {
log.info("Hi");
getContext().become(new Procedure<Object>() {
@Override
public void apply(Object message) {
log.info("Ho");
getContext().unbecome(); // resets the latest 'become' (just for fun)
}
});
} else {
unhandled(message);
}
}
}
public static void main(String... args) {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef swap = system.actorOf(new Props(Swapper.class));
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
}
}
//#swapper

View file

@ -0,0 +1,479 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor.japi;
//#all
//#imports
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import akka.actor.*;
import akka.dispatch.Mapper;
import akka.japi.Function;
import scala.concurrent.util.Duration;
import akka.util.Timeout;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import static akka.japi.Util.classTag;
import static akka.actor.SupervisorStrategy.*;
import static akka.pattern.Patterns.ask;
import static akka.pattern.Patterns.pipe;
import static docs.actor.japi.FaultHandlingDocSample.WorkerApi.*;
import static docs.actor.japi.FaultHandlingDocSample.CounterServiceApi.*;
import static docs.actor.japi.FaultHandlingDocSample.CounterApi.*;
import static docs.actor.japi.FaultHandlingDocSample.StorageApi.*;
//#imports
public class FaultHandlingDocSample {
/**
* Runs the sample
*/
public static void main(String[] args) {
Config config = ConfigFactory.parseString("akka.loglevel = DEBUG \n" + "akka.actor.debug.lifecycle = on");
ActorSystem system = ActorSystem.create("FaultToleranceSample", config);
ActorRef worker = system.actorOf(new Props(Worker.class), "worker");
ActorRef listener = system.actorOf(new Props(Listener.class), "listener");
// start the work and listen on progress
// note that the listener is used as sender of the tell,
// i.e. it will receive replies from the worker
worker.tell(Start, listener);
}
/**
* Listens on progress from the worker and shuts down the system when enough
* work has been done.
*/
public static class Listener extends UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
@Override
public void preStart() {
// If we don't get any progress within 15 seconds then the service is unavailable
getContext().setReceiveTimeout(Duration.parse("15 seconds"));
}
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof Progress) {
Progress progress = (Progress) msg;
log.info("Current progress: {} %", progress.percent);
if (progress.percent >= 100.0) {
log.info("That's all, shutting down");
getContext().system().shutdown();
}
} else if (msg == ReceiveTimeout.getInstance()) {
// No progress within 15 seconds, ServiceUnavailable
log.error("Shutting down due to unavailable service");
getContext().system().shutdown();
} else {
unhandled(msg);
}
}
}
//#messages
public interface WorkerApi {
public static final Object Start = "Start";
public static final Object Do = "Do";
public static class Progress {
public final double percent;
public Progress(double percent) {
this.percent = percent;
}
public String toString() {
return String.format("%s(%s)", getClass().getSimpleName(), percent);
}
}
}
//#messages
/**
* Worker performs some work when it receives the Start message. It will
* continuously notify the sender of the Start message of current Progress.
* The Worker supervise the CounterService.
*/
public static class Worker extends UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
final Timeout askTimeout = new Timeout(Duration.create(5, "seconds"));
// The sender of the initial Start message will continuously be notified about progress
ActorRef progressListener;
final ActorRef counterService = getContext().actorOf(new Props(CounterService.class), "counter");
final int totalCount = 51;
// Stop the CounterService child if it throws ServiceUnavailable
private static SupervisorStrategy strategy = new OneForOneStrategy(-1, Duration.Inf(),
new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof ServiceUnavailable) {
return stop();
} else {
return escalate();
}
}
});
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg.equals(Start) && progressListener == null) {
progressListener = getSender();
getContext().system().scheduler().schedule(
Duration.Zero(), Duration.create(1, "second"), getSelf(), Do, getContext().dispatcher()
);
} else if (msg.equals(Do)) {
counterService.tell(new Increment(1), getSelf());
counterService.tell(new Increment(1), getSelf());
counterService.tell(new Increment(1), getSelf());
// Send current progress to the initial sender
pipe(ask(counterService, GetCurrentCount, askTimeout)
.mapTo(classTag(CurrentCount.class))
.map(new Mapper<CurrentCount, Progress>() {
public Progress apply(CurrentCount c) {
return new Progress(100.0 * c.count / totalCount);
}
}, getContext().dispatcher()), getContext().dispatcher())
.to(progressListener);
} else {
unhandled(msg);
}
}
}
//#messages
public interface CounterServiceApi {
public static final Object GetCurrentCount = "GetCurrentCount";
public static class CurrentCount {
public final String key;
public final long count;
public CurrentCount(String key, long count) {
this.key = key;
this.count = count;
}
public String toString() {
return String.format("%s(%s, %s)", getClass().getSimpleName(), key, count);
}
}
public static class Increment {
public final long n;
public Increment(long n) {
this.n = n;
}
public String toString() {
return String.format("%s(%s)", getClass().getSimpleName(), n);
}
}
public static class ServiceUnavailable extends RuntimeException {
public ServiceUnavailable(String msg) {
super(msg);
}
}
}
//#messages
/**
* Adds the value received in Increment message to a persistent counter.
* Replies with CurrentCount when it is asked for CurrentCount. CounterService
* supervise Storage and Counter.
*/
public static class CounterService extends UntypedActor {
// Reconnect message
static final Object Reconnect = "Reconnect";
private static class SenderMsgPair {
final ActorRef sender;
final Object msg;
SenderMsgPair(ActorRef sender, Object msg) {
this.msg = msg;
this.sender = sender;
}
}
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
final String key = getSelf().path().name();
ActorRef storage;
ActorRef counter;
final List<SenderMsgPair> backlog = new ArrayList<SenderMsgPair>();
final int MAX_BACKLOG = 10000;
// Restart the storage child when StorageException is thrown.
// After 3 restarts within 5 seconds it will be stopped.
private static SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.parse("5 seconds"),
new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof StorageException) {
return restart();
} else {
return escalate();
}
}
});
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
@Override
public void preStart() {
initStorage();
}
/**
* The child storage is restarted in case of failure, but after 3 restarts,
* and still failing it will be stopped. Better to back-off than
* continuously failing. When it has been stopped we will schedule a
* Reconnect after a delay. Watch the child so we receive Terminated message
* when it has been terminated.
*/
void initStorage() {
storage = getContext().watch(getContext().actorOf(new Props(Storage.class), "storage"));
// Tell the counter, if any, to use the new storage
if (counter != null)
counter.tell(new UseStorage(storage), getSelf());
// We need the initial value to be able to operate
storage.tell(new Get(key), getSelf());
}
@Override
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof Entry && ((Entry) msg).key.equals(key) && counter == null) {
// Reply from Storage of the initial value, now we can create the Counter
final long value = ((Entry) msg).value;
counter = getContext().actorOf(new Props().withCreator(new UntypedActorFactory() {
public Actor create() {
return new Counter(key, value);
}
}));
// Tell the counter to use current storage
counter.tell(new UseStorage(storage), getSelf());
// and send the buffered backlog to the counter
for (SenderMsgPair each : backlog) {
counter.tell(each.msg, each.sender);
}
backlog.clear();
} else if (msg instanceof Increment) {
forwardOrPlaceInBacklog(msg);
} else if (msg.equals(GetCurrentCount)) {
forwardOrPlaceInBacklog(msg);
} else if (msg instanceof Terminated) {
// After 3 restarts the storage child is stopped.
// We receive Terminated because we watch the child, see initStorage.
storage = null;
// Tell the counter that there is no storage for the moment
counter.tell(new UseStorage(null), getSelf());
// Try to re-establish storage after while
getContext().system().scheduler().scheduleOnce(
Duration.create(10, "seconds"), getSelf(), Reconnect, getContext().dispatcher()
);
} else if (msg.equals(Reconnect)) {
// Re-establish storage after the scheduled delay
initStorage();
} else {
unhandled(msg);
}
}
void forwardOrPlaceInBacklog(Object msg) {
// We need the initial value from storage before we can start delegate to the counter.
// Before that we place the messages in a backlog, to be sent to the counter when
// it is initialized.
if (counter == null) {
if (backlog.size() >= MAX_BACKLOG)
throw new ServiceUnavailable("CounterService not available, lack of initial value");
backlog.add(new SenderMsgPair(getSender(), msg));
} else {
counter.forward(msg, getContext());
}
}
}
//#messages
public interface CounterApi {
public static class UseStorage {
public final ActorRef storage;
public UseStorage(ActorRef storage) {
this.storage = storage;
}
public String toString() {
return String.format("%s(%s)", getClass().getSimpleName(), storage);
}
}
}
//#messages
/**
* The in memory count variable that will send current value to the Storage,
* if there is any storage available at the moment.
*/
public static class Counter extends UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
final String key;
long count;
ActorRef storage;
public Counter(String key, long initialValue) {
this.key = key;
this.count = initialValue;
}
@Override
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof UseStorage) {
storage = ((UseStorage) msg).storage;
storeCount();
} else if (msg instanceof Increment) {
count += ((Increment) msg).n;
storeCount();
} else if (msg.equals(GetCurrentCount)) {
getSender().tell(new CurrentCount(key, count), getSelf());
} else {
unhandled(msg);
}
}
void storeCount() {
// Delegate dangerous work, to protect our valuable state.
// We can continue without storage.
if (storage != null) {
storage.tell(new Store(new Entry(key, count)), getSelf());
}
}
}
//#messages
public interface StorageApi {
public static class Store {
public final Entry entry;
public Store(Entry entry) {
this.entry = entry;
}
public String toString() {
return String.format("%s(%s)", getClass().getSimpleName(), entry);
}
}
public static class Entry {
public final String key;
public final long value;
public Entry(String key, long value) {
this.key = key;
this.value = value;
}
public String toString() {
return String.format("%s(%s, %s)", getClass().getSimpleName(), key, value);
}
}
public static class Get {
public final String key;
public Get(String key) {
this.key = key;
}
public String toString() {
return String.format("%s(%s)", getClass().getSimpleName(), key);
}
}
public static class StorageException extends RuntimeException {
public StorageException(String msg) {
super(msg);
}
}
}
//#messages
/**
* Saves key/value pairs to persistent storage when receiving Store message.
* Replies with current value when receiving Get message. Will throw
* StorageException if the underlying data store is out of order.
*/
public static class Storage extends UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
final DummyDB db = DummyDB.instance;
@Override
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof Store) {
Store store = (Store) msg;
db.save(store.entry.key, store.entry.value);
} else if (msg instanceof Get) {
Get get = (Get) msg;
Long value = db.load(get.key);
getSender().tell(new Entry(get.key, value == null ? Long.valueOf(0L) : value), getSelf());
} else {
unhandled(msg);
}
}
}
//#dummydb
public static class DummyDB {
public static final DummyDB instance = new DummyDB();
private final Map<String, Long> db = new HashMap<String, Long>();
private DummyDB() {
}
public synchronized void save(String key, Long value) throws StorageException {
if (11 <= value && value <= 14)
throw new StorageException("Simulated store failure " + value);
db.put(key, value);
}
public synchronized Long load(String key) throws StorageException {
return db.get(key);
}
}
//#dummydb
}
//#all

View file

@ -0,0 +1,10 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.agent
import org.scalatest.junit.JUnitWrapperSuite
class AgentDocJavaSpec extends JUnitWrapperSuite(
"docs.agent.AgentDocTest",
Thread.currentThread.getContextClassLoader)

View file

@ -0,0 +1,111 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.agent;
import static org.junit.Assert.*;
import scala.concurrent.ExecutionContext;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import akka.testkit.AkkaSpec;
//#import-system
import akka.actor.ActorSystem;
//#import-system
//#import-agent
import akka.agent.Agent;
//#import-agent
//#import-function
import akka.japi.Function;
//#import-function
//#import-timeout
import akka.util.Timeout;
import static java.util.concurrent.TimeUnit.SECONDS;
//#import-timeout
public class AgentDocTest {
private static ActorSystem testSystem;
private static ExecutionContext ec;
@BeforeClass
public static void beforeAll() {
testSystem = ActorSystem.create("AgentDocTest", AkkaSpec.testConf());
ec = testSystem.dispatcher();
}
@AfterClass
public static void afterAll() {
testSystem.shutdown();
testSystem = null;
}
@Test
public void createAndClose() {
//#create
ActorSystem system = ActorSystem.create("app");
Agent<Integer> agent = new Agent<Integer>(5, system);
//#create
//#close
agent.close();
//#close
system.shutdown();
}
@Test
public void sendAndSendOffAndReadAwait() {
Agent<Integer> agent = new Agent<Integer>(5, testSystem);
//#send
// send a value
agent.send(7);
// send a function
agent.send(new Function<Integer, Integer>() {
public Integer apply(Integer i) {
return i * 2;
}
});
//#send
Function<Integer, Integer> longRunningOrBlockingFunction = new Function<Integer, Integer>() {
public Integer apply(Integer i) {
return i * 1;
}
};
//#send-off
// sendOff a function
agent.sendOff(longRunningOrBlockingFunction, ec);
//#send-off
//#read-await
Integer result = agent.await(new Timeout(5, SECONDS));
//#read-await
assertEquals(result, new Integer(14));
agent.close();
}
@Test
public void readWithGet() {
Agent<Integer> agent = new Agent<Integer>(5, testSystem);
//#read-get
Integer result = agent.get();
//#read-get
assertEquals(result, new Integer(5));
agent.close();
}
}

View file

@ -0,0 +1,49 @@
package docs.camel;
//#CamelActivation
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.camel.Camel;
import akka.camel.CamelExtension;
import akka.camel.javaapi.UntypedConsumerActor;
import scala.concurrent.Future;
import scala.concurrent.util.Duration;
import scala.concurrent.util.FiniteDuration;
import static java.util.concurrent.TimeUnit.SECONDS;
//#CamelActivation
import org.junit.Test;
public class ActivationTestBase {
@Test
public void testActivation() {
//#CamelActivation
// ..
ActorSystem system = ActorSystem.create("some-system");
Props props = new Props(MyConsumer.class);
ActorRef producer = system.actorOf(props,"myproducer");
Camel camel = CamelExtension.get(system);
// get a future reference to the activation of the endpoint of the Consumer Actor
FiniteDuration duration = Duration.create(10, SECONDS);
Future<ActorRef> activationFuture = camel.activationFutureFor(producer, duration, system.dispatcher());
//#CamelActivation
//#CamelDeactivation
// ..
system.stop(producer);
// get a future reference to the deactivation of the endpoint of the Consumer Actor
Future<ActorRef> deactivationFuture = camel.deactivationFutureFor(producer, duration, system.dispatcher());
//#CamelDeactivation
system.shutdown();
}
public static class MyConsumer extends UntypedConsumerActor {
public String getEndpointUri() {
return "direct:test";
}
public void onReceive(Object message) {
}
}
}

View file

@ -0,0 +1,5 @@
package docs.camel
import org.scalatest.junit.JUnitSuite
class CamelExtensionDocTest extends CamelExtensionTestBase with JUnitSuite

View file

@ -0,0 +1,31 @@
package docs.camel;
import akka.actor.ActorSystem;
import akka.camel.Camel;
import akka.camel.CamelExtension;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.junit.Test;
public class CamelExtensionTestBase {
@Test
public void getCamelExtension() {
//#CamelExtension
ActorSystem system = ActorSystem.create("some-system");
Camel camel = CamelExtension.get(system);
CamelContext camelContext = camel.context();
ProducerTemplate producerTemplate = camel.template();
//#CamelExtension
system.shutdown();
}
public void addActiveMQComponent() {
//#CamelExtensionAddComponent
ActorSystem system = ActorSystem.create("some-system");
Camel camel = CamelExtension.get(system);
CamelContext camelContext = camel.context();
// camelContext.addComponent("activemq", ActiveMQComponent.activeMQComponent("vm://localhost?broker.persistent=false"))
//#CamelExtensionAddComponent
system.shutdown();
}
}

View file

@ -0,0 +1,24 @@
package docs.camel;
//#Consumer1
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
public class Consumer1 extends UntypedConsumerActor {
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
public String getEndpointUri() {
return "file:data/input/actor";
}
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
String body = camelMessage.getBodyAs(String.class, getCamelContext());
log.info("Received message: {}", body);
} else
unhandled(message);
}
}
//#Consumer1

View file

@ -0,0 +1,20 @@
package docs.camel;
//#Consumer2
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
public class Consumer2 extends UntypedConsumerActor {
public String getEndpointUri() {
return "jetty:http://localhost:8877/camel/default";
}
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
String body = camelMessage.getBodyAs(String.class, getCamelContext());
getSender().tell(String.format("Received message: %s",body), getSelf());
} else
unhandled(message);
}
}
//#Consumer2

View file

@ -0,0 +1,31 @@
package docs.camel;
//#Consumer3
import akka.actor.Status;
import akka.camel.Ack;
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
public class Consumer3 extends UntypedConsumerActor{
@Override
public boolean autoAck() {
return false;
}
public String getEndpointUri() {
return "jms:queue:test";
}
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
getSender().tell(Ack.getInstance(), getSelf());
// on success
// ..
Exception someException = new Exception("e1");
// on failure
getSender().tell(new Status.Failure(someException), getSelf());
} else
unhandled(message);
}
}
//#Consumer3

View file

@ -0,0 +1,31 @@
package docs.camel;
//#Consumer4
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
import scala.concurrent.util.Duration;
import scala.concurrent.util.FiniteDuration;
import java.util.concurrent.TimeUnit;
public class Consumer4 extends UntypedConsumerActor {
private final static FiniteDuration timeout = Duration.create(500, TimeUnit.MILLISECONDS);
@Override
public Duration replyTimeout() {
return timeout;
}
public String getEndpointUri() {
return "jetty:http://localhost:8877/camel/default";
}
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
String body = camelMessage.getBodyAs(String.class, getCamelContext());
getSender().tell(String.format("Hello %s",body), getSelf());
} else
unhandled(message);
}
}
//#Consumer4

View file

@ -0,0 +1,18 @@
package docs.camel;
//#CustomRoute
import akka.actor.ActorRef;
import akka.camel.internal.component.CamelPath;
import org.apache.camel.builder.RouteBuilder;
public class CustomRouteBuilder extends RouteBuilder{
private String uri;
public CustomRouteBuilder(ActorRef responder) {
uri = CamelPath.toUri(responder);
}
public void configure() throws Exception {
from("jetty:http://localhost:8877/camel/custom").to(uri);
}
}
//#CustomRoute

View file

@ -0,0 +1,20 @@
package docs.camel;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.camel.Camel;
import akka.camel.CamelExtension;
public class CustomRouteTestBase {
public void customRoute() throws Exception{
//#CustomRoute
ActorSystem system = ActorSystem.create("some-system");
Camel camel = CamelExtension.get(system);
ActorRef responder = system.actorOf(new Props(Responder.class), "TestResponder");
camel.context().addRoutes(new CustomRouteBuilder(responder));
//#CustomRoute
system.stop(responder);
system.shutdown();
}
}

View file

@ -0,0 +1,42 @@
package docs.camel;
//#ErrorThrowingConsumer
import akka.actor.Status;
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
import org.apache.camel.builder.Builder;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.model.RouteDefinition;
import scala.Option;
public class ErrorThrowingConsumer extends UntypedConsumerActor{
private String uri;
public ErrorThrowingConsumer(String uri){
this.uri = uri;
}
public String getEndpointUri() {
return uri;
}
public void onReceive(Object message) throws Exception{
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
String body = camelMessage.getBodyAs(String.class, getCamelContext());
throw new Exception(String.format("error: %s",body));
} else
unhandled(message);
}
@Override
public ProcessorDefinition<?> onRouteDefinition(RouteDefinition rd) {
// Catch any exception and handle it by returning the exception message as response
return rd.onException(Exception.class).handled(true).transform(Builder.exceptionMessage()).end();
}
@Override
public void preRestart(Throwable reason, Option<Object> message) {
getSender().tell(new Status.Failure(reason), getSelf());
}
}
//#ErrorThrowingConsumer

View file

@ -0,0 +1,10 @@
package docs.camel;
//#Producer1
import akka.camel.javaapi.UntypedProducerActor;
public class FirstProducer extends UntypedProducerActor {
public String getEndpointUri() {
return "http://localhost:8080/news";
}
}
//#Producer1

View file

@ -0,0 +1,24 @@
package docs.camel;
//#RouteResponse
import akka.actor.ActorRef;
import akka.camel.javaapi.UntypedProducerActor;
public class Forwarder extends UntypedProducerActor {
private String uri;
private ActorRef target;
public Forwarder(String uri, ActorRef target) {
this.uri = uri;
this.target = target;
}
public String getEndpointUri() {
return uri;
}
@Override
public void onRouteResponse(Object message) {
target.forward(message, getContext());
}
}
//#RouteResponse

View file

@ -0,0 +1,15 @@
package docs.camel;
//#ProducerTemplate
import akka.actor.UntypedActor;
import akka.camel.Camel;
import akka.camel.CamelExtension;
import org.apache.camel.ProducerTemplate;
public class MyActor extends UntypedActor {
public void onReceive(Object message) {
Camel camel = CamelExtension.get(getContext().system());
ProducerTemplate template = camel.template();
template.sendBody("direct:news", message);
}
}
//#ProducerTemplate

View file

@ -0,0 +1,31 @@
package docs.camel;
//#Consumer-mina
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
public class MyEndpoint extends UntypedConsumerActor{
private String uri;
public String getEndpointUri() {
return uri;
}
public void onReceive(Object message) throws Exception {
if (message instanceof CamelMessage) {
/* ... */
} else
unhandled(message);
}
// Extra constructor to change the default uri,
// for instance to "jetty:http://localhost:8877/example"
public MyEndpoint(String uri) {
this.uri = uri;
}
public MyEndpoint() {
this.uri = "mina:tcp://localhost:6200?textline=true";
}
}
//#Consumer-mina

View file

@ -0,0 +1,26 @@
package docs.camel;
import akka.actor.*;
public class OnRouteResponseTestBase {
public void onRouteResponse(){
//#RouteResponse
ActorSystem system = ActorSystem.create("some-system");
Props receiverProps = new Props(ResponseReceiver.class);
final ActorRef receiver = system.actorOf(receiverProps,"responseReceiver");
UntypedActorFactory factory = new UntypedActorFactory() {
public Actor create() {
return new Forwarder("http://localhost:8080/news/akka", receiver);
}
};
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", null);
//#RouteResponse
system.stop(receiver);
system.stop(forwardResponse);
system.shutdown();
}
}

View file

@ -0,0 +1,20 @@
package docs.camel;
//#Oneway
import akka.camel.javaapi.UntypedProducerActor;
public class OnewaySender extends UntypedProducerActor{
private String uri;
public OnewaySender(String uri) {
this.uri = uri;
}
public String getEndpointUri() {
return uri;
}
@Override
public boolean isOneway() {
return true;
}
}
//#Oneway

View file

@ -0,0 +1,10 @@
package docs.camel;
//#Producer
import akka.camel.javaapi.UntypedProducerActor;
public class Orders extends UntypedProducerActor {
public String getEndpointUri() {
return "jms:queue:Orders";
}
}
//#Producer

View file

@ -0,0 +1,10 @@
package docs.camel;
//#Producer1
import akka.camel.javaapi.UntypedProducerActor;
public class Producer1 extends UntypedProducerActor {
public String getEndpointUri() {
return "http://localhost:8080/news";
}
}
//#Producer1

View file

@ -0,0 +1,47 @@
package docs.camel;
import java.util.HashMap;
import java.util.Map;
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() {
//#TellProducer
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\"/>", null);
//#TellProducer
system.shutdown();
}
public void askProducer() {
//#AskProducer
ActorSystem system = ActorSystem.create("some-system");
Props props = new Props(FirstProducer.class);
ActorRef producer = system.actorOf(props,"myproducer");
Future<Object> future = Patterns.ask(producer, "some request", 1000);
//#AskProducer
system.stop(producer);
system.shutdown();
}
public void correlate(){
//#Correlate
ActorSystem system = ActorSystem.create("some-system");
Props props = new Props(Orders.class);
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), null);
//#Correlate
system.stop(producer);
system.shutdown();
}
}

View file

@ -0,0 +1,15 @@
package docs.camel;
//#RequestProducerTemplate
import akka.actor.UntypedActor;
import akka.camel.Camel;
import akka.camel.CamelExtension;
import org.apache.camel.ProducerTemplate;
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), getSelf());
}
}
//#RequestProducerTemplate

View file

@ -0,0 +1,25 @@
package docs.camel;
//#CustomRoute
import akka.actor.UntypedActor;
import akka.camel.CamelMessage;
import akka.japi.Function;
public class Responder extends UntypedActor{
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
getSender().tell(createResponse(camelMessage), getSelf());
} else
unhandled(message);
}
private CamelMessage createResponse(CamelMessage msg) {
return msg.mapBody(new Function<String,String>() {
public String apply(String body) {
return String.format("received %s", body);
}
});
}
}
//#CustomRoute

View file

@ -0,0 +1,13 @@
package docs.camel;
//#RouteResponse
import akka.actor.UntypedActor;
import akka.camel.CamelMessage;
public class ResponseReceiver extends UntypedActor{
public void onReceive(Object message) {
if(message instanceof CamelMessage) {
// do something with the forwarded response
}
}
}
//#RouteResponse

View file

@ -0,0 +1,36 @@
package docs.camel;
//#TransformOutgoingMessage
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedProducerActor;
import akka.japi.Function;
public class Transformer extends UntypedProducerActor{
private String uri;
public Transformer(String uri) {
this.uri = uri;
}
public String getEndpointUri() {
return uri;
}
private CamelMessage upperCase(CamelMessage msg) {
return msg.mapBody(new Function<String,String>() {
public String apply(String body) {
return body.toUpperCase();
}
});
}
@Override
public Object onTransformOutgoingMessage(Object message) {
if(message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
return upperCase(camelMessage);
} else {
return message;
}
}
}
//#TransformOutgoingMessage

View file

@ -0,0 +1,23 @@
package docs.camel.sample.http;
import akka.actor.ActorRef;
import akka.camel.javaapi.UntypedConsumerActor;
//#HttpExample
public class HttpConsumer extends UntypedConsumerActor{
private ActorRef producer;
public HttpConsumer(ActorRef producer){
this.producer = producer;
}
public String getEndpointUri() {
return "jetty:http://0.0.0.0:8875/";
}
public void onReceive(Object message) {
producer.forward(message, getContext());
}
}
//#HttpExample

View file

@ -0,0 +1,38 @@
package docs.camel.sample.http;
import akka.actor.ActorRef;
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedProducerActor;
import org.apache.camel.Exchange;
import java.util.HashSet;
import java.util.Set;
//#HttpExample
public class HttpProducer extends UntypedProducerActor{
private ActorRef transformer;
public HttpProducer(ActorRef transformer) {
this.transformer = transformer;
}
public String getEndpointUri() {
return "jetty://http://akka.io/?bridgeEndpoint=true";
}
@Override
public Object onTransformOutgoingMessage(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
Set<String> httpPath = new HashSet<String>();
httpPath.add(Exchange.HTTP_PATH);
return camelMessage.withHeaders(camelMessage.getHeaders(httpPath));
} else return super.onTransformOutgoingMessage(message);
}
@Override
public void onRouteResponse(Object message) {
transformer.forward(message, getContext());
}
}
//#HttpExample

View file

@ -0,0 +1,26 @@
package docs.camel.sample.http;
import akka.actor.*;
public class HttpSample {
public static void main(String[] args) {
//#HttpExample
// Create the actors. this can be done in a Boot class so you can
// run the example in the MicroKernel. just add the below three lines to your boot class.
ActorSystem system = ActorSystem.create("some-system");
final ActorRef httpTransformer = system.actorOf(new Props(HttpTransformer.class));
final ActorRef httpProducer = system.actorOf(new Props(new UntypedActorFactory(){
public Actor create() {
return new HttpProducer(httpTransformer);
}
}));
ActorRef httpConsumer = system.actorOf(new Props(new UntypedActorFactory(){
public Actor create() {
return new HttpConsumer(httpProducer);
}
}));
//#HttpExample
}
}

View file

@ -0,0 +1,26 @@
package docs.camel.sample.http;
import akka.actor.Status;
import akka.actor.UntypedActor;
import akka.camel.CamelMessage;
import akka.japi.Function;
//#HttpExample
public class HttpTransformer extends UntypedActor{
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
CamelMessage replacedMessage = camelMessage.mapBody(new Function<Object, String>(){
public String apply(Object body) {
String text = new String((byte[])body);
return text.replaceAll("Akka ", "AKKA ");
}
});
getSender().tell(replacedMessage, getSelf());
} else if (message instanceof Status.Failure) {
getSender().tell(message, getSelf());
} else
unhandled(message);
}
}
//#HttpExample

View file

@ -0,0 +1,20 @@
package docs.camel.sample.quartz;
//#QuartzExample
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
public class MyQuartzActor extends UntypedConsumerActor{
public String getEndpointUri() {
return "quartz://example?cron=0/2+*+*+*+*+?";
}
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
String body = camelMessage.getBodyAs(String.class, getCamelContext());
System.out.println(String.format("==============> received %s ", body));
} else
unhandled(message);
}
}
//#QuartzExample

View file

@ -0,0 +1,12 @@
package docs.camel.sample.quartz;
//#QuartzExample
import akka.actor.ActorSystem;
import akka.actor.Props;
public class QuartzSample {
public static void main(String[] args) {
ActorSystem system = ActorSystem.create("my-quartz-system");
system.actorOf(new Props(MyQuartzActor.class));
}
}
//#QuartzExample

View file

@ -0,0 +1,27 @@
package docs.camel.sample.route;
//#CustomRouteExample
import akka.actor.ActorRef;
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
public class Consumer3 extends UntypedConsumerActor{
private ActorRef transformer;
public Consumer3(ActorRef transformer){
this.transformer = transformer;
}
public String getEndpointUri() {
return "jetty:http://0.0.0.0:8877/camel/welcome";
}
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
CamelMessage camelMessage = (CamelMessage) message;
transformer.forward(camelMessage.getBodyAs(String.class, getCamelContext()),getContext());
} else
unhandled(message);
}
}
//#CustomRouteExample

View file

@ -0,0 +1,17 @@
package docs.camel.sample.route;
//#CustomRouteExample
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
public class CustomRouteBuilder extends RouteBuilder{
public void configure() throws Exception {
from("direct:welcome").process(new Processor(){
public void process(Exchange exchange) throws Exception {
exchange.getOut().setBody(String.format("Welcome %s",exchange.getIn().getBody()));
}
});
}
}
//#CustomRouteExample

View file

@ -0,0 +1,31 @@
package docs.camel.sample.route;
import akka.actor.*;
import akka.camel.CamelExtension;
public class CustomRouteSample {
public static void main(String[] args) {
try {
//#CustomRouteExample
// the below lines can be added to a Boot class, so that you can run the example from a MicroKernel
ActorSystem system = ActorSystem.create("some-system");
final ActorRef producer = system.actorOf(new Props(Producer1.class));
final ActorRef mediator = system.actorOf(new Props(new UntypedActorFactory() {
public Actor create() {
return new Transformer(producer);
}
}));
ActorRef consumer = system.actorOf(new Props(new UntypedActorFactory() {
public Actor create() {
return new Consumer3(mediator);
}
}));
CamelExtension.get(system).context().addRoutes(new CustomRouteBuilder());
//#CustomRouteExample
} catch (Exception e) {
e.printStackTrace();
}
}
}

View file

@ -0,0 +1,10 @@
package docs.camel.sample.route;
//#CustomRouteExample
import akka.camel.javaapi.UntypedProducerActor;
public class Producer1 extends UntypedProducerActor{
public String getEndpointUri() {
return "direct:welcome";
}
}
//#CustomRouteExample

View file

@ -0,0 +1,29 @@
package docs.camel.sample.route;
//#CustomRouteExample
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.camel.CamelMessage;
import akka.japi.Function;
public class Transformer extends UntypedActor {
private ActorRef producer;
public Transformer(ActorRef producer) {
this.producer = producer;
}
public void onReceive(Object message) {
if (message instanceof CamelMessage) {
// example: transform message body "foo" to "- foo -" and forward result to producer
CamelMessage camelMessage = (CamelMessage) message;
CamelMessage transformedMessage = camelMessage.mapBody(new Function<String, String>(){
public String apply(String body) {
return String.format("- %s -",body);
}
});
producer.forward(transformedMessage, getContext());
} else
unhandled(message);
}
}
//#CustomRouteExample

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.dispatcher
import org.scalatest.junit.JUnitSuite
class DispatcherDocTest extends DispatcherDocTestBase with JUnitSuite

View file

@ -0,0 +1,174 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.dispatcher;
//#imports
import akka.actor.*;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
//#imports
//#imports-prio
import akka.event.Logging;
import akka.event.LoggingAdapter;
//#imports-prio
//#imports-prio-mailbox
import akka.dispatch.PriorityGenerator;
import akka.dispatch.UnboundedPriorityMailbox;
import com.typesafe.config.Config;
//#imports-prio-mailbox
//#imports-custom
import akka.dispatch.Envelope;
import akka.dispatch.MessageQueue;
import akka.dispatch.MailboxType;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
//#imports-custom
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import scala.Option;
import com.typesafe.config.ConfigFactory;
import docs.actor.MyUntypedActor;
import akka.testkit.AkkaSpec;
public class DispatcherDocTestBase {
ActorSystem system;
@Before
public void setUp() {
system = ActorSystem.create("MySystem",
ConfigFactory.parseString(DispatcherDocSpec.config()).withFallback(AkkaSpec.testConf()));
}
@After
public void tearDown() {
system.shutdown();
}
@Test
public void defineDispatcher() {
//#defining-dispatcher
ActorRef myActor =
system.actorOf(new Props(MyUntypedActor.class).withDispatcher("my-dispatcher"),
"myactor3");
//#defining-dispatcher
}
@Test
public void definePinnedDispatcher() {
//#defining-pinned-dispatcher
ActorRef myActor = system.actorOf(new Props(MyUntypedActor.class)
.withDispatcher("my-pinned-dispatcher"));
//#defining-pinned-dispatcher
}
@Test
public void priorityDispatcher() throws Exception {
//#prio-dispatcher
// We create a new Actor that just prints out what it processes
ActorRef myActor = system.actorOf(
new Props().withCreator(new UntypedActorFactory() {
public UntypedActor create() {
return new UntypedActor() {
LoggingAdapter log =
Logging.getLogger(getContext().system(), this);
{
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) {
log.info(message.toString());
}
};
}
}).withDispatcher("prio-dispatcher"));
/*
Logs:
'highpriority
'highpriority
'pigdog
'pigdog2
'pigdog3
'lowpriority
'lowpriority
*/
//#prio-dispatcher
for (int i = 0; i < 10; i++) {
if (myActor.isTerminated())
break;
Thread.sleep(100);
}
}
//#prio-mailbox
public static class MyPrioMailbox extends UnboundedPriorityMailbox {
public MyPrioMailbox(ActorSystem.Settings settings, Config config) { // needed for reflective instantiation
// Create a new PriorityGenerator, lower prio means more important
super(new PriorityGenerator() {
@Override
public int gen(Object message) {
if (message.equals("highpriority"))
return 0; // 'highpriority messages should be treated first if possible
else if (message.equals("lowpriority"))
return 2; // 'lowpriority messages should be treated last if possible
else if (message.equals(PoisonPill.getInstance()))
return 3; // PoisonPill when no other left
else
return 1; // By default they go between high and low prio
}
});
}
}
//#prio-mailbox
//#mailbox-implementation-example
class MyUnboundedMailbox implements MailboxType {
// This constructor signature must exist, it will be called by Akka
public MyUnboundedMailbox(ActorSystem.Settings settings, Config config) {
// put your initialization code here
}
// The create method is called to create the MessageQueue
public MessageQueue create(Option<ActorRef> owner, Option<ActorSystem> system) {
return new MessageQueue() {
private final Queue<Envelope> queue = new ConcurrentLinkedQueue<Envelope>();
// these must be implemented; queue used as example
public void enqueue(ActorRef receiver, Envelope handle) { queue.offer(handle); }
public Envelope dequeue() { return queue.poll(); }
public int numberOfMessages() { return queue.size(); }
public boolean hasMessages() { return !queue.isEmpty(); }
public void cleanUp(ActorRef owner, MessageQueue deadLetters) {
for (Envelope handle: queue) {
deadLetters.enqueue(owner, handle);
}
}
};
}
}
//#mailbox-implementation-example
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.event
import org.scalatest.junit.JUnitSuite
class LoggingDocTest extends LoggingDocTestBase with JUnitSuite

View file

@ -0,0 +1,122 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.event;
//#imports
import akka.event.Logging;
import akka.event.LoggingAdapter;
//#imports
//#imports-listener
import akka.event.Logging.InitializeLogger;
import akka.event.Logging.Error;
import akka.event.Logging.Warning;
import akka.event.Logging.Info;
import akka.event.Logging.Debug;
//#imports-listener
import org.junit.Test;
import scala.Option;
import akka.actor.UntypedActorFactory;
//#imports-deadletter
import akka.actor.Props;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.UntypedActor;
import akka.actor.DeadLetter;
//#imports-deadletter
public class LoggingDocTestBase {
@Test
public void useLoggingActor() {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef myActor = system.actorOf(new Props(new UntypedActorFactory() {
public UntypedActor create() {
return new MyActor();
}
}));
myActor.tell("test", null);
system.shutdown();
}
@Test
public void subscribeToDeadLetters() {
//#deadletters
final ActorSystem system = ActorSystem.create("DeadLetters");
final ActorRef actor = system.actorOf(new Props(DeadLetterActor.class));
system.eventStream().subscribe(actor, DeadLetter.class);
//#deadletters
system.shutdown();
}
@Test
public void demonstrateMultipleArgs() {
final ActorSystem system = ActorSystem.create("multiArg");
//#array
final Object[] args = new Object[] { "The", "brown", "fox", "jumps", 42 };
system.log().debug("five parameters: {}, {}, {}, {}, {}", args);
//#array
system.shutdown();
}
//#my-actor
class MyActor extends UntypedActor {
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
@Override
public void preStart() {
log.debug("Starting");
}
@Override
public void preRestart(Throwable reason, Option<Object> message) {
log.error(reason, "Restarting due to [{}] when processing [{}]", reason.getMessage(),
message.isDefined() ? message.get() : "");
}
public void onReceive(Object message) {
if (message.equals("test")) {
log.info("Received test");
} else {
log.warning("Received unknown message: {}", message);
}
}
}
//#my-actor
//#my-event-listener
class MyEventListener extends UntypedActor {
public void onReceive(Object message) {
if (message instanceof InitializeLogger) {
getSender().tell(Logging.loggerInitialized(), getSelf());
} else if (message instanceof Error) {
// ...
} else if (message instanceof Warning) {
// ...
} else if (message instanceof Info) {
// ...
} else if (message instanceof Debug) {
// ...
}
}
}
//#my-event-listener
//#deadletter-actor
public static class DeadLetterActor extends UntypedActor {
public void onReceive(Object message) {
if (message instanceof DeadLetter) {
System.out.println(message);
}
}
}
//#deadletter-actor
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.extension
import org.scalatest.junit.JUnitSuite
class ExtensionDocTest extends ExtensionDocTestBase with JUnitSuite

View file

@ -0,0 +1,75 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.extension;
//#imports
import akka.actor.*;
import java.util.concurrent.atomic.AtomicLong;
//#imports
import org.junit.Test;
public class ExtensionDocTestBase {
//#extension
public static class CountExtensionImpl implements Extension {
//Since this Extension is a shared instance
// per ActorSystem we need to be threadsafe
private final AtomicLong counter = new AtomicLong(0);
//This is the operation this Extension provides
public long increment() {
return counter.incrementAndGet();
}
}
//#extension
//#extensionid
public static class CountExtension extends AbstractExtensionId<CountExtensionImpl> implements ExtensionIdProvider {
//This will be the identifier of our CountExtension
public final static CountExtension CountExtensionProvider = new CountExtension();
//The lookup method is required by ExtensionIdProvider,
// so we return ourselves here, this allows us
// to configure our extension to be loaded when
// the ActorSystem starts up
public CountExtension lookup() {
return CountExtension.CountExtensionProvider; //The public static final
}
//This method will be called by Akka
// to instantiate our Extension
public CountExtensionImpl createExtension(ExtendedActorSystem system) {
return new CountExtensionImpl();
}
}
//#extensionid
//#extension-usage-actor
public static class MyActor extends UntypedActor {
public void onReceive(Object msg) {
// typically you would use static import of CountExtension.CountExtensionProvider field
CountExtension.CountExtensionProvider.get(getContext().system()).increment();
}
}
//#extension-usage-actor
@Test
public void demonstrateHowToCreateAndUseAnAkkaExtensionInJava() {
final ActorSystem system = null;
try {
//#extension-usage
// typically you would use static import of CountExtension.CountExtensionProvider field
CountExtension.CountExtensionProvider.get(system).increment();
//#extension-usage
} catch (Exception e) {
//do nothing
}
}
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.extension
import org.scalatest.junit.JUnitSuite
class SettingsExtensionDocTest extends SettingsExtensionDocTestBase with JUnitSuite

View file

@ -0,0 +1,87 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.extension;
//#imports
import akka.actor.Extension;
import akka.actor.AbstractExtensionId;
import akka.actor.ExtensionIdProvider;
import akka.actor.ActorSystem;
import akka.actor.ExtendedActorSystem;
import scala.concurrent.util.Duration;
import com.typesafe.config.Config;
import java.util.concurrent.TimeUnit;
//#imports
import akka.actor.UntypedActor;
import org.junit.Test;
public class SettingsExtensionDocTestBase {
//#extension
public static class SettingsImpl implements Extension {
public final String DB_URI;
public final Duration CIRCUIT_BREAKER_TIMEOUT;
public SettingsImpl(Config config) {
DB_URI = config.getString("myapp.db.uri");
CIRCUIT_BREAKER_TIMEOUT = Duration.create(config.getMilliseconds("myapp.circuit-breaker.timeout"),
TimeUnit.MILLISECONDS);
}
}
//#extension
//#extensionid
public static class Settings extends AbstractExtensionId<SettingsImpl> implements ExtensionIdProvider {
public final static Settings SettingsProvider = new Settings();
public Settings lookup() {
return Settings.SettingsProvider;
}
public SettingsImpl createExtension(ExtendedActorSystem system) {
return new SettingsImpl(system.settings().config());
}
}
//#extensionid
//#extension-usage-actor
public static class MyActor extends UntypedActor {
// typically you would use static import of CountExtension.CountExtensionProvider field
final SettingsImpl settings = Settings.SettingsProvider.get(getContext().system());
Connection connection = connect(settings.DB_URI, settings.CIRCUIT_BREAKER_TIMEOUT);
//#extension-usage-actor
public Connection connect(String dbUri, Duration circuitBreakerTimeout) {
return new Connection();
}
public void onReceive(Object msg) {
}
}
public static class Connection {
}
@Test
public void demonstrateHowToCreateAndUseAnAkkaExtensionInJava() {
final ActorSystem system = null;
try {
//#extension-usage
// typically you would use static import of CountExtension.CountExtensionProvider field
String dbUri = Settings.SettingsProvider.get(system).DB_URI;
//#extension-usage
} catch (Exception e) {
//do nothing
}
}
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.future
import org.scalatest.junit.JUnitSuite
class FutureDocTest extends FutureDocTestBase with JUnitSuite

View file

@ -0,0 +1,550 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.future;
//#imports1
import akka.dispatch.*;
import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
import scala.concurrent.Await;
import akka.util.Timeout;
//#imports1
//#imports2
import scala.concurrent.util.Duration;
import akka.japi.Function;
import java.util.concurrent.Callable;
import static akka.dispatch.Futures.future;
import static java.util.concurrent.TimeUnit.SECONDS;
//#imports2
//#imports3
import static akka.dispatch.Futures.sequence;
//#imports3
//#imports4
import static akka.dispatch.Futures.traverse;
//#imports4
//#imports5
import akka.japi.Function2;
import static akka.dispatch.Futures.fold;
//#imports5
//#imports6
import static akka.dispatch.Futures.reduce;
//#imports6
//#imports7
import scala.concurrent.ExecutionContext;
import scala.concurrent.ExecutionContext$;
//#imports7
//#imports8
import static akka.pattern.Patterns.after;
//#imports8
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import akka.testkit.AkkaSpec;
import akka.actor.Status.Failure;
import akka.actor.ActorSystem;
import akka.actor.UntypedActor;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.pattern.Patterns;
import static org.junit.Assert.*;
public class FutureDocTestBase {
ActorSystem system;
@Before
public void setUp() {
system = ActorSystem.create("MySystem", AkkaSpec.testConf());
}
@After
public void tearDown() {
system.shutdown();
}
@SuppressWarnings("unchecked") @Test public void useCustomExecutionContext() throws Exception {
ExecutorService yourExecutorServiceGoesHere = Executors.newSingleThreadExecutor();
//#diy-execution-context
ExecutionContext ec =
ExecutionContexts.fromExecutorService(yourExecutorServiceGoesHere);
//Use ec with your Futures
Future<String> f1 = Futures.successful("foo");
// Then you shut the ExecutorService down somewhere at the end of your program/application.
yourExecutorServiceGoesHere.shutdown();
//#diy-execution-context
}
@Test
public void useBlockingFromActor() throws Exception {
ActorRef actor = system.actorOf(new Props(MyActor.class));
String msg = "hello";
//#ask-blocking
Timeout timeout = new Timeout(Duration.create(5, "seconds"));
Future<Object> future = Patterns.ask(actor, msg, timeout);
String result = (String) Await.result(future, timeout.duration());
//#ask-blocking
assertEquals("HELLO", result);
}
@Test
public void useFutureEval() throws Exception {
//#future-eval
Future<String> f = future(new Callable<String>() {
public String call() {
return "Hello" + "World";
}
}, system.dispatcher());
String result = (String) Await.result(f, Duration.create(1, SECONDS));
//#future-eval
assertEquals("HelloWorld", result);
}
@Test
public void useMap() throws Exception {
//#map
final ExecutionContext ec = system.dispatcher();
Future<String> f1 = future(new Callable<String>() {
public String call() {
return "Hello" + "World";
}
}, ec);
Future<Integer> f2 = f1.map(new Mapper<String, Integer>() {
public Integer apply(String s) {
return s.length();
}
}, ec);
int result = Await.result(f2, Duration.create(1, SECONDS));
assertEquals(10, result);
//#map
}
@Test
public void useMap2() throws Exception {
//#map2
final ExecutionContext ec = system.dispatcher();
Future<String> f1 = future(new Callable<String>() {
public String call() throws Exception {
Thread.sleep(100);
return "Hello" + "World";
}
}, ec);
Future<Integer> f2 = f1.map(new Mapper<String, Integer>() {
public Integer apply(String s) {
return s.length();
}
}, ec);
//#map2
int result = Await.result(f2, Duration.create(1, SECONDS));
assertEquals(10, result);
}
@Test
public void useMap3() throws Exception {
//#map3
final ExecutionContext ec = system.dispatcher();
Future<String> f1 = future(new Callable<String>() {
public String call() {
return "Hello" + "World";
}
}, ec);
Thread.sleep(100);
Future<Integer> f2 = f1.map(new Mapper<String, Integer>() {
public Integer apply(String s) {
return s.length();
}
}, ec);
//#map3
int result = Await.result(f2, Duration.create(1, SECONDS));
assertEquals(10, result);
}
@Test
public void useFlatMap() throws Exception {
//#flat-map
final ExecutionContext ec = system.dispatcher();
Future<String> f1 = future(new Callable<String>() {
public String call() {
return "Hello" + "World";
}
}, ec);
Future<Integer> f2 = f1.flatMap(new Mapper<String, Future<Integer>>() {
public Future<Integer> apply(final String s) {
return future(new Callable<Integer>() {
public Integer call() {
return s.length();
}
}, ec);
}
}, ec);
//#flat-map
int result = Await.result(f2, Duration.create(1, SECONDS));
assertEquals(10, result);
}
@Test
public void useSequence() throws Exception {
List<Future<Integer>> source = new ArrayList<Future<Integer>>();
source.add(Futures.successful(1));
source.add(Futures.successful(2));
//#sequence
final ExecutionContext ec = system.dispatcher();
//Some source generating a sequence of Future<Integer>:s
Iterable<Future<Integer>> listOfFutureInts = source;
// now we have a Future[Iterable[Integer]]
Future<Iterable<Integer>> futureListOfInts = sequence(listOfFutureInts, ec);
// Find the sum of the odd numbers
Future<Long> futureSum = futureListOfInts.map(new Mapper<Iterable<Integer>, Long>() {
public Long apply(Iterable<Integer> ints) {
long sum = 0;
for (Integer i : ints)
sum += i;
return sum;
}
}, ec);
long result = Await.result(futureSum, Duration.create(1, SECONDS));
//#sequence
assertEquals(3L, result);
}
@Test
public void useTraverse() throws Exception {
//#traverse
final ExecutionContext ec = system.dispatcher();
//Just a sequence of Strings
Iterable<String> listStrings = Arrays.asList("a", "b", "c");
Future<Iterable<String>> futureResult = traverse(listStrings, new Function<String, Future<String>>() {
public Future<String> apply(final String r) {
return future(new Callable<String>() {
public String call() {
return r.toUpperCase();
}
}, ec);
}
}, ec);
//Returns the sequence of strings as upper case
Iterable<String> result = Await.result(futureResult, Duration.create(1, SECONDS));
assertEquals(Arrays.asList("A", "B", "C"), result);
//#traverse
}
@Test
public void useFold() throws Exception {
List<Future<String>> source = new ArrayList<Future<String>>();
source.add(Futures.successful("a"));
source.add(Futures.successful("b"));
//#fold
final ExecutionContext ec = system.dispatcher();
//A sequence of Futures, in this case Strings
Iterable<Future<String>> futures = source;
//Start value is the empty string
Future<String> resultFuture = fold("", futures, new Function2<String, String, String>() {
public String apply(String r, String t) {
return r + t; //Just concatenate
}
}, ec);
String result = Await.result(resultFuture, Duration.create(1, SECONDS));
//#fold
assertEquals("ab", result);
}
@Test
public void useReduce() throws Exception {
List<Future<String>> source = new ArrayList<Future<String>>();
source.add(Futures.successful("a"));
source.add(Futures.successful("b"));
//#reduce
final ExecutionContext ec = system.dispatcher();
//A sequence of Futures, in this case Strings
Iterable<Future<String>> futures = source;
Future<Object> resultFuture = reduce(futures, new Function2<Object, String, Object>() {
public Object apply(Object r, String t) {
return r + t; //Just concatenate
}
}, ec);
Object result = Await.result(resultFuture, Duration.create(1, SECONDS));
//#reduce
assertEquals("ab", result);
}
@Test
public void useSuccessfulAndFailed() throws Exception {
final ExecutionContext ec = system.dispatcher();
//#successful
Future<String> future = Futures.successful("Yay!");
//#successful
//#failed
Future<String> otherFuture = Futures.failed(new IllegalArgumentException("Bang!"));
//#failed
Object result = Await.result(future, Duration.create(1, SECONDS));
assertEquals("Yay!", result);
Throwable result2 = Await.result(otherFuture.failed(), Duration.create(1, SECONDS));
assertEquals("Bang!", result2.getMessage());
}
@Test
public void useFilter() throws Exception {
//#filter
final ExecutionContext ec = system.dispatcher();
Future<Integer> future1 = Futures.successful(4);
Future<Integer> successfulFilter = future1.filter(Filter.filterOf(new Function<Integer, Boolean>() {
public Boolean apply(Integer i) {
return i % 2 == 0;
}
}), ec);
Future<Integer> failedFilter = future1.filter(Filter.filterOf(new Function<Integer, Boolean>() {
public Boolean apply(Integer i) {
return i % 2 != 0;
}
}), ec);
//When filter fails, the returned Future will be failed with a scala.MatchError
//#filter
}
public void sendToTheInternetz(String s) {
}
public void sendToIssueTracker(Throwable t) {
}
@Test
public void useAndThen() {
//#and-then
final ExecutionContext ec = system.dispatcher();
Future<String> future1 = Futures.successful("value").andThen(new OnComplete<String>() {
public void onComplete(Throwable failure, String result) {
if (failure != null)
sendToIssueTracker(failure);
}
}, ec).andThen(new OnComplete<String>() {
public void onComplete(Throwable failure, String result) {
if (result != null)
sendToTheInternetz(result);
}
}, ec);
//#and-then
}
@Test
public void useRecover() throws Exception {
//#recover
final ExecutionContext ec = system.dispatcher();
Future<Integer> future = future(new Callable<Integer>() {
public Integer call() {
return 1 / 0;
}
}, ec).recover(new Recover<Integer>() {
public Integer recover(Throwable problem) throws Throwable {
if (problem instanceof ArithmeticException)
return 0;
else
throw problem;
}
}, ec);
int result = Await.result(future, Duration.create(1, SECONDS));
assertEquals(result, 0);
//#recover
}
@Test
public void useTryRecover() throws Exception {
//#try-recover
final ExecutionContext ec = system.dispatcher();
Future<Integer> future = future(new Callable<Integer>() {
public Integer call() {
return 1 / 0;
}
}, ec).recoverWith(new Recover<Future<Integer>>() {
public Future<Integer> recover(Throwable problem) throws Throwable {
if (problem instanceof ArithmeticException) {
return future(new Callable<Integer>() {
public Integer call() {
return 0;
}
}, ec);
} else
throw problem;
}
}, ec);
int result = Await.result(future, Duration.create(1, SECONDS));
assertEquals(result, 0);
//#try-recover
}
@Test
public void useOnSuccessOnFailureAndOnComplete() throws Exception {
{
Future<String> future = Futures.successful("foo");
//#onSuccess
final ExecutionContext ec = system.dispatcher();
future.onSuccess(new OnSuccess<String>() {
public void onSuccess(String result) {
if ("bar" == result) {
//Do something if it resulted in "bar"
} else {
//Do something if it was some other String
}
}
}, ec);
//#onSuccess
}
{
Future<String> future = Futures.failed(new IllegalStateException("OHNOES"));
//#onFailure
final ExecutionContext ec = system.dispatcher();
future.onFailure(new OnFailure() {
public void onFailure(Throwable failure) {
if (failure instanceof IllegalStateException) {
//Do something if it was this particular failure
} else {
//Do something if it was some other failure
}
}
}, ec);
//#onFailure
}
{
Future<String> future = Futures.successful("foo");
//#onComplete
final ExecutionContext ec = system.dispatcher();
future.onComplete(new OnComplete<String>() {
public void onComplete(Throwable failure, String result) {
if (failure != null) {
//We got a failure, handle it here
} else {
// We got a result, do something with it
}
}
}, ec);
//#onComplete
}
}
@Test
public void useOrAndZip() throws Exception {
{
//#zip
final ExecutionContext ec = system.dispatcher();
Future<String> future1 = Futures.successful("foo");
Future<String> future2 = Futures.successful("bar");
Future<String> future3 = future1.zip(future2).map(new Mapper<scala.Tuple2<String, String>, String>() {
public String apply(scala.Tuple2<String, String> zipped) {
return zipped._1() + " " + zipped._2();
}
}, ec);
String result = Await.result(future3, Duration.create(1, SECONDS));
assertEquals("foo bar", result);
//#zip
}
{
//#fallback-to
Future<String> future1 = Futures.failed(new IllegalStateException("OHNOES1"));
Future<String> future2 = Futures.failed(new IllegalStateException("OHNOES2"));
Future<String> future3 = Futures.successful("bar");
Future<String> future4 = future1.fallbackTo(future2).fallbackTo(future3); // Will have "bar" in this case
String result = Await.result(future4, Duration.create(1, SECONDS));
assertEquals("bar", result);
//#fallback-to
}
}
@Test(expected = IllegalStateException.class)
public void useAfter() throws Exception {
//#after
final ExecutionContext ec = system.dispatcher();
Future<String> failExc = Futures.failed(new IllegalStateException("OHNOES1"));
Future<String> delayed = Patterns.after(Duration.create(500, "millis"),
system.scheduler(), ec, failExc);
Future<String> future = future(new Callable<String>() {
public String call() throws InterruptedException {
Thread.sleep(1000);
return "foo";
}
}, ec);
Future<String> result = future.either(delayed);
//#after
Await.result(result, Duration.create(2, SECONDS));
}
public static class MyActor extends UntypedActor {
public void onReceive(Object message) {
if (message instanceof String) {
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")), getSelf());
} else {
getSender().tell(i, getSelf());
}
} else {
unhandled(message);
}
}
}
}

View file

@ -0,0 +1,5 @@
package docs.jrouting;
import org.scalatest.junit.JUnitSuite
class ConsistentHashingRouterDocTest extends ConsistentHashingRouterDocTestBase with JUnitSuite

View file

@ -0,0 +1,136 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.jrouting;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import akka.testkit.JavaTestKit;
import akka.actor.ActorSystem;
//#imports1
import akka.actor.UntypedActor;
import akka.routing.ConsistentHashingRouter.ConsistentHashable;
import java.util.Map;
import java.util.HashMap;
import java.io.Serializable;
//#imports1
//#imports2
import akka.actor.Props;
import akka.actor.ActorRef;
import akka.routing.ConsistentHashingRouter;
import akka.routing.ConsistentHashingRouter.ConsistentHashMapper;
import akka.routing.ConsistentHashingRouter.ConsistentHashableEnvelope;
//#imports2
public class ConsistentHashingRouterDocTestBase {
static ActorSystem system;
@BeforeClass
public static void setup() {
system = ActorSystem.create();
}
@AfterClass
public static void teardown() {
system.shutdown();
}
//#cache-actor
public static class Cache extends UntypedActor {
Map<String, String> cache = new HashMap<String, String>();
public void onReceive(Object msg) {
if (msg instanceof Entry) {
Entry entry = (Entry) msg;
cache.put(entry.key, entry.value);
} else if (msg instanceof Get) {
Get get = (Get) msg;
Object value = cache.get(get.key);
getSender().tell(value == null ? NOT_FOUND : value,
getContext().self());
} else if (msg instanceof Evict) {
Evict evict = (Evict) msg;
cache.remove(evict.key);
} else {
unhandled(msg);
}
}
}
public static final class Evict implements Serializable {
public final String key;
public Evict(String key) {
this.key = key;
}
}
public static final class Get implements Serializable, ConsistentHashable {
public final String key;
public Get(String key) {
this.key = key;
}
public Object consistentHashKey() {
return key;
}
}
public static final class Entry implements Serializable {
public final String key;
public final String value;
public Entry(String key, String value) {
this.key = key;
this.value = value;
}
}
public static final String NOT_FOUND = "NOT_FOUND";
//#cache-actor
@Test
public void demonstrateUsageOfConsistentHashableRouter() {
new JavaTestKit(system) {{
//#consistent-hashing-router
final ConsistentHashMapper hashMapper = new ConsistentHashMapper() {
@Override
public Object hashKey(Object message) {
if (message instanceof Evict) {
return ((Evict) message).key;
} else {
return null;
}
}
};
ActorRef cache = system.actorOf(new Props(Cache.class).withRouter(
new ConsistentHashingRouter(10).withHashMapper(hashMapper)),
"cache");
cache.tell(new ConsistentHashableEnvelope(
new Entry("hello", "HELLO"), "hello"), getRef());
cache.tell(new ConsistentHashableEnvelope(
new Entry("hi", "HI"), "hi"), getRef());
cache.tell(new Get("hello"), getRef());
expectMsgEquals("HELLO");
cache.tell(new Get("hi"), getRef());
expectMsgEquals("HI");
cache.tell(new Evict("hi"), getRef());
cache.tell(new Get("hi"), getRef());
expectMsgEquals(NOT_FOUND);
//#consistent-hashing-router
}};
}
}

View file

@ -0,0 +1,5 @@
package docs.jrouting;
import org.scalatest.junit.JUnitSuite
class CustomRouterDocTest extends CustomRouterDocTestBase with JUnitSuite

View file

@ -0,0 +1,186 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.jrouting;
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 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 akka.util.Timeout;
public class CustomRouterDocTestBase {
ActorSystem system;
@Before
public void setUp() {
system = ActorSystem.create("MySystem", AkkaSpec.testConf());
}
@After
public void tearDown() {
system.shutdown();
}
public static class MyActor extends UntypedActor {
@Override public void onReceive(Object o) {}
}
@Test
public void demonstrateDispatchers() {
//#dispatchers
final ActorRef router = system.actorOf(new Props(MyActor.class)
.withRouter(new RoundRobinRouter(5).withDispatcher("head")) // head router runs on "head" dispatcher
.withDispatcher("workers")); // MyActor workers run on "workers" dispatcher
//#dispatchers
}
@Test
public void demonstrateSupervisor() {
//#supervision
final SupervisorStrategy strategy = new OneForOneStrategy(5, Duration.parse("1 minute"),
new Class<?>[] { Exception.class });
final ActorRef router = system.actorOf(new Props(MyActor.class)
.withRouter(new RoundRobinRouter(5).withSupervisorStrategy(strategy)));
//#supervision
}
//#crTest
@Test
public void countVotesAsIntendedNotAsInFlorida() throws Exception {
ActorRef routedActor = system.actorOf(new Props().withRouter(new VoteCountRouter()));
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);
assertEquals(3, Await.result(democratsResult, timeout.duration()));
assertEquals(2, Await.result(republicansResult, timeout.duration()));
}
//#crTest
//#CustomRouter
//#crMessages
enum Message {
DemocratVote, DemocratCountResult, RepublicanVote, RepublicanCountResult
}
//#crMessages
//#crActors
public static class DemocratActor extends UntypedActor {
int counter = 0;
public void onReceive(Object msg) {
switch ((Message) msg) {
case DemocratVote:
counter++;
break;
case DemocratCountResult:
getSender().tell(counter, getSelf());
break;
default:
unhandled(msg);
}
}
}
public static class RepublicanActor extends UntypedActor {
int counter = 0;
public void onReceive(Object msg) {
switch ((Message) msg) {
case RepublicanVote:
counter++;
break;
case RepublicanCountResult:
getSender().tell(counter, getSelf());
break;
default:
unhandled(msg);
}
}
}
//#crActors
//#crRouter
public static class VoteCountRouter extends CustomRouterConfig {
@Override public String routerDispatcher() {
return Dispatchers.DefaultDispatcherId();
}
@Override public SupervisorStrategy supervisorStrategy() {
return SupervisorStrategy.defaultStrategy();
}
//#crRoute
@Override
public CustomRoute createCustomRoute(RouteeProvider routeeProvider) {
final ActorRef democratActor = routeeProvider.context().actorOf(new Props(DemocratActor.class), "d");
final ActorRef republicanActor = routeeProvider.context().actorOf(new Props(RepublicanActor.class), "r");
List<ActorRef> routees = Arrays.asList(new ActorRef[] { democratActor, republicanActor });
//#crRegisterRoutees
routeeProvider.registerRoutees(routees);
//#crRegisterRoutees
//#crRoutingLogic
return new CustomRoute() {
@Override
public Iterable<Destination> destinationsFor(ActorRef sender, Object msg) {
switch ((Message) msg) {
case DemocratVote:
case DemocratCountResult:
return Arrays.asList(new Destination[] { new Destination(sender, democratActor) });
case RepublicanVote:
case RepublicanCountResult:
return Arrays.asList(new Destination[] { new Destination(sender, republicanActor) });
default:
throw new IllegalArgumentException("Unknown message: " + msg);
}
}
};
//#crRoutingLogic
}
//#crRoute
}
//#crRouter
//#CustomRouter
}

View file

@ -0,0 +1,48 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.jrouting;
import java.io.Serializable;
import akka.actor.UntypedActor;
//#fibonacciActor
public class FibonacciActor extends UntypedActor {
public void onReceive(Object msg) {
if (msg instanceof FibonacciNumber) {
FibonacciNumber fibonacciNumber = (FibonacciNumber) msg;
getSender().tell(fibonacci(fibonacciNumber.getNbr()), getSelf());
} else {
unhandled(msg);
}
}
private int fibonacci(int n) {
return fib(n, 1, 0);
}
private int fib(int n, int b, int a) {
if (n == 0)
return a;
// recursion
return fib(n - 1, a + b, b);
}
public static class FibonacciNumber implements Serializable {
private static final long serialVersionUID = 1L;
private final int nbr;
public FibonacciNumber(int nbr) {
this.nbr = nbr;
}
public int getNbr() {
return nbr;
}
}
}
//#fibonacciActor

View file

@ -0,0 +1,69 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.jrouting;
import akka.routing.ScatterGatherFirstCompletedRouter;
import akka.routing.BroadcastRouter;
import akka.routing.RandomRouter;
import akka.routing.RoundRobinRouter;
import akka.routing.SmallestMailboxRouter;
import akka.actor.UntypedActor;
import akka.actor.ActorRef;
import akka.actor.Props;
import scala.concurrent.util.Duration;
import akka.util.Timeout;
import scala.concurrent.Future;
import scala.concurrent.Await;
//#parentActor
public class ParentActor extends UntypedActor {
public void onReceive(Object msg) throws Exception {
if (msg.equals("rrr")) {
//#roundRobinRouter
ActorRef roundRobinRouter = getContext().actorOf(
new Props(PrintlnActor.class).withRouter(new RoundRobinRouter(5)), "router");
for (int i = 1; i <= 10; i++) {
roundRobinRouter.tell(i, getSelf());
}
//#roundRobinRouter
} else if (msg.equals("rr")) {
//#randomRouter
ActorRef randomRouter = getContext().actorOf(new Props(PrintlnActor.class).withRouter(new RandomRouter(5)),
"router");
for (int i = 1; i <= 10; i++) {
randomRouter.tell(i, getSelf());
}
//#randomRouter
} else if (msg.equals("smr")) {
//#smallestMailboxRouter
ActorRef smallestMailboxRouter = getContext().actorOf(
new Props(PrintlnActor.class).withRouter(new SmallestMailboxRouter(5)), "router");
for (int i = 1; i <= 10; i++) {
smallestMailboxRouter.tell(i, getSelf());
}
//#smallestMailboxRouter
} else if (msg.equals("br")) {
//#broadcastRouter
ActorRef broadcastRouter = getContext().actorOf(new Props(PrintlnActor.class).withRouter(new BroadcastRouter(5)),
"router");
broadcastRouter.tell("this is a broadcast message", getSelf());
//#broadcastRouter
} else if (msg.equals("sgfcr")) {
//#scatterGatherFirstCompletedRouter
ActorRef scatterGatherFirstCompletedRouter = getContext().actorOf(
new Props(FibonacciActor.class).withRouter(new ScatterGatherFirstCompletedRouter(5, Duration
.create(2, "seconds"))), "router");
Timeout timeout = new Timeout(Duration.create(5, "seconds"));
Future<Object> futureResult = akka.pattern.Patterns.ask(scatterGatherFirstCompletedRouter,
new FibonacciActor.FibonacciNumber(10), timeout);
int result = (Integer) Await.result(futureResult, timeout.duration());
//#scatterGatherFirstCompletedRouter
System.out.println(String.format("The result of calculating Fibonacci for 10 is %d", result));
} else {
unhandled(msg);
}
}
}
//#parentActor

View file

@ -0,0 +1,15 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.jrouting;
import akka.actor.UntypedActor;
//#printlnActor
public class PrintlnActor extends UntypedActor {
public void onReceive(Object msg) {
System.out.println(String.format("Received message '%s' in actor %s", msg, getSelf().path().name()));
}
}
//#printlnActor

View file

@ -0,0 +1,58 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.jrouting;
import akka.routing.FromConfig;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.actor.ActorSystem;
import com.typesafe.config.ConfigFactory;
import com.typesafe.config.Config;
public class RouterViaConfigExample {
public static class ExampleActor extends UntypedActor {
public void onReceive(Object msg) {
if (msg instanceof Message) {
Message message = (Message) msg;
System.out.println(String.format("Received %s in router %s", message.getNbr(), getSelf().path().name()));
} else {
unhandled(msg);
}
}
public static class Message {
private final int nbr;
public Message(int nbr) {
this.nbr = nbr;
}
public int getNbr() {
return nbr;
}
}
}
public static void main(String... args) {
Config config = ConfigFactory.parseString("akka.actor.deployment {\n" + " /router {\n"
+ " router = round-robin\n" + " nr-of-instances = 5\n" + " }\n" + "}\n");
ActorSystem system = ActorSystem.create("Example", config);
//#configurableRouting
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), 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), null);
}
}
}

View file

@ -0,0 +1,96 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.jrouting;
import akka.routing.RoundRobinRouter;
import akka.routing.DefaultResizer;
import akka.remote.routing.RemoteRouterConfig;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.actor.AddressFromURIString;
import java.util.Arrays;
public class RouterViaProgramExample {
public static class ExampleActor extends UntypedActor {
public void onReceive(Object msg) {
if (msg instanceof Message) {
Message message = (Message) msg;
System.out.println(String.format("Received %s in router %s", message.getNbr(), getSelf().path().name()));
} else {
unhandled(msg);
}
}
public static class Message {
private final int nbr;
public Message(int nbr) {
this.nbr = nbr;
}
public int getNbr() {
return nbr;
}
}
}
public static void main(String... args) {
ActorSystem system = ActorSystem.create("RPE");
//#programmaticRoutingNrOfInstances
int nrOfInstances = 5;
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), null);
}
//#programmaticRoutingRoutees
ActorRef actor1 = system.actorOf(new Props(ExampleActor.class));
ActorRef actor2 = system.actorOf(new Props(ExampleActor.class));
ActorRef actor3 = system.actorOf(new Props(ExampleActor.class));
Iterable<ActorRef> routees = Arrays.asList(new ActorRef[] { actor1, actor2, actor3 });
ActorRef router2 = system.actorOf(new Props().withRouter(RoundRobinRouter.create(routees)));
//#programmaticRoutingRoutees
for (int i = 1; i <= 6; i++) {
router2.tell(new ExampleActor.Message(i), null);
}
//#programmaticRoutingWithResizer
int lowerBound = 2;
int upperBound = 15;
DefaultResizer resizer = new DefaultResizer(lowerBound, upperBound);
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), null);
}
//#remoteRoutees
Address addr1 = new Address("akka", "remotesys", "otherhost", 1234);
Address addr2 = AddressFromURIString.parse("akka://othersys@anotherhost:1234");
Address[] addresses = new Address[] { addr1, addr2 };
ActorRef routerRemote = system.actorOf(new Props(ExampleActor.class)
.withRouter(new RemoteRouterConfig(new RoundRobinRouter(5), addresses)));
//#remoteRoutees
}
private class CompileCheckJavaDocsForRouting extends UntypedActor {
@Override
public void onReceive(Object o) {
//#reply-with-parent
getSender().tell("reply", getContext().parent()); // replies go to router
//#reply-with-parent
//#reply-with-self
getSender().tell("reply", getSelf()); // replies go to this actor
//#reply-with-self
}
}
}

View file

@ -0,0 +1,18 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.pattern;
// this part will not appear in the docs
//#all-of-it
class JavaTemplate {
public JavaTemplate() {
System.out.println("Hello, Template!");
}
//#uninteresting-stuff
// dont show this plumbimg
//#uninteresting-stuff
}
//#all-of-it

View file

@ -0,0 +1,21 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.remoting;
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
public class RemoteActorExample extends UntypedActor {
//#localNodeActor
ActorRef a1 = getContext().actorFor("/serviceA/retrieval");
//#localNodeActor
//#remoteNodeActor
ActorRef a2 = getContext().actorFor("akka://app@10.0.0.1:2552/user/serviceA/retrieval");
//#remoteNodeActor
public void onReceive(Object message) throws Exception {
// Do something
}
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.remoting
import org.scalatest.junit.JUnitSuite
class RemoteDeploymentDocTest extends RemoteDeploymentDocTestBase with JUnitSuite

View file

@ -0,0 +1,46 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.remoting;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
//#import
import akka.actor.ActorRef;
import akka.actor.Address;
import akka.actor.AddressFromURIString;
import akka.actor.Deploy;
import akka.actor.Props;
import akka.actor.ActorSystem;
import akka.remote.RemoteScope;
//#import
public class RemoteDeploymentDocTestBase {
static ActorSystem system;
@BeforeClass
public static void init() {
system = ActorSystem.create();
}
@AfterClass
public static void cleanup() {
system.shutdown();
}
@Test
public void demonstrateDeployment() {
//#make-address
Address addr = new Address("akka", "sys", "host", 1234);
addr = AddressFromURIString.parse("akka://sys@host:1234"); // the same
//#make-address
//#deploy
ActorRef ref = system.actorOf(new Props(RemoteDeploymentDocSpec.Echo.class).withDeploy(new Deploy(new RemoteScope(addr))));
//#deploy
assert ref.path().address().equals(addr);
}
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.serialization
import org.scalatest.junit.JUnitSuite
class SerializationDocTest extends SerializationDocTestBase with JUnitSuite

View file

@ -0,0 +1,195 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.serialization;
import org.junit.Test;
import static org.junit.Assert.*;
//#imports
import akka.actor.*;
import akka.remote.RemoteActorRefProvider;
import akka.serialization.*;
//#imports
public class SerializationDocTestBase {
//#my-own-serializer
public static class MyOwnSerializer extends JSerializer {
// This is whether "fromBinary" requires a "clazz" or not
@Override public boolean includeManifest() {
return false;
}
// Pick a unique identifier for your Serializer,
// you've got a couple of billions to choose from,
// 0 - 16 is reserved by Akka itself
@Override public int identifier() {
return 1234567;
}
// "toBinary" serializes the given object to an Array of Bytes
@Override public byte[] toBinary(Object obj) {
// Put the code that serializes the object here
//#...
return new byte[0];
//#...
}
// "fromBinary" deserializes the given array,
// using the type hint (if any, see "includeManifest" above)
@Override public Object fromBinaryJava(byte[] bytes,
Class<?> clazz) {
// Put your code that deserializes here
//#...
return null;
//#...
}
}
//#my-own-serializer
@Test public void serializeActorRefs() {
final ActorSystem theActorSystem =
ActorSystem.create("whatever");
final ActorRef theActorRef =
theActorSystem.deadLetters(); // Of course this should be you
//#actorref-serializer
// Serialize
// (beneath toBinary)
final Address transportAddress =
Serialization.currentTransportAddress().value();
String identifier;
// If there is no transportAddress,
// it means that either this Serializer isn't called
// within a piece of code that sets it,
// so either you need to supply your own,
// or simply use the local path.
if (transportAddress == null) identifier = theActorRef.path().toString();
else identifier = theActorRef.path().toStringWithAddress(transportAddress);
// Then just serialize the identifier however you like
// Deserialize
// (beneath fromBinary)
final ActorRef deserializedActorRef = theActorSystem.actorFor(identifier);
// Then just use the ActorRef
//#actorref-serializer
theActorSystem.shutdown();
}
//#external-address
public static class ExternalAddressExt implements Extension {
private final ExtendedActorSystem system;
public ExternalAddressExt(ExtendedActorSystem system) {
this.system = system;
}
public Address getAddressFor(Address remoteAddress) {
final scala.Option<Address> optAddr = system.provider()
.getExternalAddressFor(remoteAddress);
if (optAddr.isDefined()) {
return optAddr.get();
} else {
throw new UnsupportedOperationException(
"cannot send to remote address " + remoteAddress);
}
}
}
public static class ExternalAddress extends
AbstractExtensionId<ExternalAddressExt> implements ExtensionIdProvider {
public static final ExternalAddress ID = new ExternalAddress();
public ExternalAddress lookup() {
return ID;
}
public ExternalAddressExt createExtension(ExtendedActorSystem system) {
return new ExternalAddressExt(system);
}
}
//#external-address
public void demonstrateExternalAddress() {
// this is not meant to be run, only to be compiled
final ActorSystem system = ActorSystem.create();
final Address remoteAddr = new Address("", "");
// #external-address
final Address addr = ExternalAddress.ID.get(system).getAddressFor(remoteAddr);
// #external-address
}
//#external-address-default
public static class DefaultAddressExt implements Extension {
private final ExtendedActorSystem system;
public DefaultAddressExt(ExtendedActorSystem system) {
this.system = system;
}
public Address getAddress() {
final ActorRefProvider provider = system.provider();
if (provider instanceof RemoteActorRefProvider) {
return ((RemoteActorRefProvider) provider).transport().address();
} else {
throw new UnsupportedOperationException("need RemoteActorRefProvider");
}
}
}
public static class DefaultAddress extends
AbstractExtensionId<DefaultAddressExt> implements ExtensionIdProvider {
public static final DefaultAddress ID = new DefaultAddress();
public DefaultAddress lookup() {
return ID;
}
public DefaultAddressExt createExtension(ExtendedActorSystem system) {
return new DefaultAddressExt(system);
}
}
//#external-address-default
public void demonstrateDefaultAddress() {
// this is not meant to be run, only to be compiled
final ActorSystem system = ActorSystem.create();
final Address remoteAddr = new Address("", "");
// #external-address-default
final Address addr = DefaultAddress.ID.get(system).getAddress();
// #external-address-default
}
@Test
public void demonstrateTheProgrammaticAPI() {
//#programmatic
ActorSystem system = ActorSystem.create("example");
// Get the Serialization Extension
Serialization serialization = SerializationExtension.get(system);
// Have something to serialize
String original = "woohoo";
// Find the Serializer for it
Serializer serializer = serialization.findSerializerFor(original);
// Turn it into bytes
byte[] bytes = serializer.toBinary(original);
// Turn it back into an object,
// the nulls are for the class manifest and for the classloader
String back = (String) serializer.fromBinary(bytes);
// Voilá!
assertEquals(original, back);
//#programmatic
system.shutdown();
}
}

View file

@ -0,0 +1,399 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.testkit;
import static org.junit.Assert.*;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import com.typesafe.config.ConfigFactory;
import com.typesafe.config.Config;
import akka.actor.ActorKilledException;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Kill;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
import scala.concurrent.Await;
import scala.concurrent.Future;
import akka.testkit.CallingThreadDispatcher;
import akka.testkit.TestActor;
import akka.testkit.TestActor.AutoPilot;
import akka.testkit.TestActorRef;
import akka.testkit.JavaTestKit;
import scala.concurrent.util.Duration;
public class TestKitDocTest {
//#test-actor-ref
static class MyActor extends UntypedActor {
public void onReceive(Object o) throws Exception {
if (o.equals("say42")) {
getSender().tell(42, getSelf());
} else if (o instanceof Exception) {
throw (Exception) o;
}
}
public boolean testMe() { return true; }
}
//#test-actor-ref
private static ActorSystem system;
@BeforeClass
public static void setup() {
final Config config = ConfigFactory.parseString(
"akka.event-handlers = [akka.testkit.TestEventListener]");
system = ActorSystem.create("demoSystem", config);
}
@AfterClass
public static void cleanup() {
system.shutdown();
}
//#test-actor-ref
@Test
public void demonstrateTestActorRef() {
final Props props = new Props(MyActor.class);
final TestActorRef<MyActor> ref = TestActorRef.create(system, props, "testA");
final MyActor actor = ref.underlyingActor();
assertTrue(actor.testMe());
}
//#test-actor-ref
@Test
public void demonstrateAsk() throws Exception {
//#test-behavior
final Props props = new Props(MyActor.class);
final TestActorRef<MyActor> ref = TestActorRef.create(system, props, "testB");
final Future<Object> future = akka.pattern.Patterns.ask(ref, "say42", 3000);
assertTrue(future.isCompleted());
assertEquals(42, Await.result(future, Duration.Zero()));
//#test-behavior
}
@Test
public void demonstrateExceptions() {
//#test-expecting-exceptions
final Props props = new Props(MyActor.class);
final TestActorRef<MyActor> ref = TestActorRef.create(system, props, "myActor");
try {
ref.receive(new Exception("expected"));
fail("expected an exception to be thrown");
} catch (Exception e) {
assertEquals("expected", e.getMessage());
}
//#test-expecting-exceptions
}
@Test
public void demonstrateWithin() {
//#test-within
new JavaTestKit(system) {{
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() {
assertEquals((Integer) 42, expectMsgClass(Integer.class));
}
};
}};
//#test-within
}
@Test
public void demonstrateExpectMsg() {
//#test-expectmsg
new JavaTestKit(system) {{
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) {
if (in instanceof Integer) {
return "match";
} else {
throw noMatch();
}
}
}.get(); // this extracts the received message
assertEquals("match", out);
}};
//#test-expectmsg
}
@Test
public void demonstrateReceiveWhile() {
//#test-receivewhile
new JavaTestKit(system) {{
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
protected String match(Object in) {
if (in instanceof Integer) {
return in.toString();
} else {
throw noMatch();
}
}
}.get(); // this extracts the received messages
assertArrayEquals(new String[] {"42", "43"}, out);
expectMsgEquals("hello");
}};
//#test-receivewhile
new JavaTestKit(system) {{
//#test-receivewhile-full
new ReceiveWhile<String>( // type of array to be created must match ...
String.class, // ... this class which is needed to that end
duration("100 millis"), // maximum collect time
duration("50 millis"), // maximum time between messages
12 // maximum number of messages to collect
) {
//#match-elided
protected String match(Object in) {
throw noMatch();
}
//#match-elided
};
//#test-receivewhile-full
}};
}
@Test
public void demonstrateAwaitCond() {
//#test-awaitCond
new JavaTestKit(system) {{
getRef().tell(42, null);
new AwaitCond(
duration("1 second"), // maximum wait time
duration("100 millis") // interval at which to check the condition
) {
// do not put code outside this method, will run afterwards
protected boolean cond() {
// typically used to wait for something to start up
return msgAvailable();
}
};
}};
//#test-awaitCond
}
@Test
@SuppressWarnings("unchecked") // due to generic varargs
public void demonstrateExpect() {
new JavaTestKit(system) {{
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");
final Object[] all = expectMsgAllOf("hello", "world");
final int i = expectMsgClass(Integer.class);
final Number j = expectMsgAnyClassOf(Integer.class, Long.class);
expectNoMsg();
//#test-expect
assertEquals("hello", hello);
assertEquals("hello", any);
assertEquals(42, i);
assertEquals(42, j);
assertArrayEquals(new String[] {"hello", "world"}, all);
}};
}
@Test
public void demonstrateIgnoreMsg() {
//#test-ignoreMsg
new JavaTestKit(system) {{
// ignore all Strings
new IgnoreMsg() {
protected boolean ignore(Object msg) {
return msg instanceof String;
}
};
getRef().tell("hello", null);
getRef().tell(42, null);
expectMsgEquals(42);
// remove message filter
ignoreNoMsg();
getRef().tell("hello", null);
expectMsgEquals("hello");
}};
//#test-ignoreMsg
}
@Test
public void demonstrateDilated() {
//#duration-dilation
new JavaTestKit(system) {{
final Duration original = duration("1 second");
final Duration stretched = dilated(original);
assertTrue("dilated", stretched.gteq(original));
}};
//#duration-dilation
}
@Test
public void demonstrateProbe() {
//#test-probe
// simple actor which just forwards messages
class Forwarder extends UntypedActor {
final ActorRef target;
public Forwarder(ActorRef target) {
this.target = target;
}
public void onReceive(Object msg) {
target.forward(msg, getContext());
}
}
new JavaTestKit(system) {{
// create a test probe
final JavaTestKit probe = new JavaTestKit(system);
// create a forwarder, injecting the probes testActor
final Props props = new Props(new UntypedActorFactory() {
private static final long serialVersionUID = 8927158735963950216L;
public UntypedActor create() {
return new Forwarder(probe.getRef());
}
});
final ActorRef forwarder = system.actorOf(props, "forwarder");
// verify correct forwarding
forwarder.tell(42, getRef());
probe.expectMsgEquals(42);
assertEquals(getRef(), probe.getLastSender());
}};
//#test-probe
}
@Test
public void demonstrateSpecialProbe() {
//#test-special-probe
new JavaTestKit(system) {{
class MyProbe extends JavaTestKit {
public MyProbe() {
super(system);
}
public void assertHello() {
expectMsgEquals("hello");
}
}
final MyProbe probe = new MyProbe();
probe.getRef().tell("hello", null);
probe.assertHello();
}};
//#test-special-probe
}
@Test
public void demonstrateReply() {
//#test-probe-reply
new JavaTestKit(system) {{
final JavaTestKit probe = new JavaTestKit(system);
probe.getRef().tell("hello", getRef());
probe.expectMsgEquals("hello");
probe.reply("world");
expectMsgEquals("world");
assertEquals(probe.getRef(), getLastSender());
}};
//#test-probe-reply
}
@Test
public void demonstrateForward() {
//#test-probe-forward
new JavaTestKit(system) {{
final JavaTestKit probe = new JavaTestKit(system);
probe.getRef().tell("hello", getRef());
probe.expectMsgEquals("hello");
probe.forward(getRef());
expectMsgEquals("hello");
assertEquals(getRef(), getLastSender());
}};
//#test-probe-forward
}
@Test
public void demonstrateWithinProbe() {
try {
//#test-within-probe
new JavaTestKit(system) {{
final JavaTestKit probe = new JavaTestKit(system);
new Within(duration("1 second")) {
public void run() {
probe.expectMsgEquals("hello");
}
};
}};
//#test-within-probe
} catch (AssertionError e) {
// expected to fail
}
}
@Test
public void demonstrateAutoPilot() {
//#test-auto-pilot
new JavaTestKit(system) {{
final JavaTestKit probe = new JavaTestKit(system);
// install auto-pilot
probe.setAutoPilot(new TestActor.AutoPilot() {
public AutoPilot run(ActorRef sender, Object msg) {
sender.tell(msg, null);
return noAutoPilot();
}
});
// first one is replied to directly ...
probe.getRef().tell("hello", getRef());
expectMsgEquals("hello");
// ... but then the auto-pilot switched itself off
probe.getRef().tell("world", getRef());
expectNoMsg();
}};
//#test-auto-pilot
}
// only compilation
public void demonstrateCTD() {
//#calling-thread-dispatcher
system.actorOf(
new Props(MyActor.class)
.withDispatcher(CallingThreadDispatcher.Id()));
//#calling-thread-dispatcher
}
@Test
public void demonstrateEventFilter() {
//#test-event-filter
new JavaTestKit(system) {{
assertEquals("demoSystem", system.name());
final ActorRef victim = system.actorOf(Props.empty(), "victim");
final int result = new EventFilter<Integer>(ActorKilledException.class) {
protected Integer run() {
victim.tell(Kill.getInstance(), null);
return 42;
}
}.from("akka://demoSystem/user/victim").occurrences(1).exec();
assertEquals(42, result);
}};
//#test-event-filter
}
}

View file

@ -0,0 +1,95 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.testkit;
//#fullsample
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.testkit.JavaTestKit;
import scala.concurrent.util.Duration;
public class TestKitSampleTest {
public static class SomeActor extends UntypedActor {
ActorRef target = null;
public void onReceive(Object msg) {
if (msg.equals("hello")) {
getSender().tell("world", getSelf());
if (target != null) target.forward(msg, getContext());
} else if (msg instanceof ActorRef) {
target = (ActorRef) msg;
getSender().tell("done", getSelf());
}
}
}
static ActorSystem system;
@BeforeClass
public static void setup() {
system = ActorSystem.create();
}
@AfterClass
public static void teardown() {
system.shutdown();
}
@Test
public void testIt() {
/*
* Wrap the whole test procedure within a testkit constructor
* if you want to receive actor replies or use Within(), etc.
*/
new JavaTestKit(system) {{
final Props props = new Props(SomeActor.class);
final ActorRef subject = system.actorOf(props);
// can also use JavaTestKit from the outside
final JavaTestKit probe = new JavaTestKit(system);
// inject the probe by passing it to the test subject
// like a real resource would be passed in production
subject.tell(probe.getRef(), getRef());
// await the correct response
expectMsgEquals(duration("1 second"), "done");
// the run() method needs to finish within 3 seconds
new Within(duration("3 seconds")) {
protected void run() {
subject.tell("hello", getRef());
// This is a demo: would normally use expectMsgEquals().
// Wait time is bounded by 3-second deadline above.
new AwaitCond() {
protected boolean cond() {
return probe.msgAvailable();
}
};
// response must have been enqueued to us before probe
expectMsgEquals(Duration.Zero(), "world");
// check that the probe we injected earlier got the msg
probe.expectMsgEquals(Duration.Zero(), "hello");
Assert.assertEquals(getRef(), probe.getLastSender());
// Will wait for the rest of the 3 seconds
expectNoMsg();
}
};
}};
}
}
//#fullsample

View file

@ -0,0 +1,38 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor;
//#class
import akka.actor.*;
import akka.transactor.*;
import scala.concurrent.stm.Ref;
import scala.concurrent.stm.japi.STM;
public class CoordinatedCounter extends UntypedActor {
private Ref.View<Integer> count = STM.newRef(0);
public void onReceive(Object incoming) throws Exception {
if (incoming instanceof Coordinated) {
Coordinated coordinated = (Coordinated) incoming;
Object message = coordinated.getMessage();
if (message instanceof Increment) {
Increment increment = (Increment) message;
if (increment.hasFriend()) {
increment.getFriend().tell(coordinated.coordinate(new Increment()), getSelf());
}
coordinated.atomic(new Runnable() {
public void run() {
STM.increment(count, 1);
}
});
}
} else if ("GetCount".equals(incoming)) {
getSender().tell(count.get(), getSelf());
} else {
unhandled(incoming);
}
}
}
//#class

View file

@ -0,0 +1,28 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor;
import akka.actor.*;
import akka.transactor.*;
public class Coordinator extends UntypedActor {
public void onReceive(Object incoming) throws Exception {
if (incoming instanceof Coordinated) {
Coordinated coordinated = (Coordinated) incoming;
Object message = coordinated.getMessage();
if (message instanceof Message) {
//#coordinated-atomic
coordinated.atomic(new Runnable() {
public void run() {
// do something in the coordinated transaction ...
}
});
//#coordinated-atomic
}
} else {
unhandled(incoming);
}
}
}

View file

@ -0,0 +1,28 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor;
//#class
import akka.transactor.*;
import scala.concurrent.stm.Ref;
import scala.concurrent.stm.japi.STM;
public class Counter extends UntypedTransactor {
Ref.View<Integer> count = STM.newRef(0);
public void atomically(Object message) {
if (message instanceof Increment) {
STM.increment(count, 1);
}
}
@Override public boolean normally(Object message) {
if ("GetCount".equals(message)) {
getSender().tell(count.get(), getSelf());
return true;
} else return false;
}
}
//#class

View file

@ -0,0 +1,38 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor;
//#class
import akka.transactor.*;
import java.util.Set;
import scala.concurrent.stm.Ref;
import scala.concurrent.stm.japi.STM;
public class FriendlyCounter extends UntypedTransactor {
Ref.View<Integer> count = STM.newRef(0);
@Override public Set<SendTo> coordinate(Object message) {
if (message instanceof Increment) {
Increment increment = (Increment) message;
if (increment.hasFriend())
return include(increment.getFriend(), new Increment());
}
return nobody();
}
public void atomically(Object message) {
if (message instanceof Increment) {
STM.increment(count, 1);
}
}
@Override public boolean normally(Object message) {
if ("GetCount".equals(message)) {
getSender().tell(count.get(), getSelf());
return true;
} else return false;
}
}
//#class

View file

@ -0,0 +1,27 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor;
//#class
import akka.actor.ActorRef;
public class Increment {
private ActorRef friend = null;
public Increment() {}
public Increment(ActorRef friend) {
this.friend = friend;
}
public boolean hasFriend() {
return friend != null;
}
public ActorRef getFriend() {
return friend;
}
}
//#class

View file

@ -0,0 +1,7 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor;
public class Message {}

View file

@ -0,0 +1,11 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor
import org.scalatest.junit.JUnitWrapperSuite
class TransactorDocJavaSpec extends JUnitWrapperSuite(
"docs.transactor.TransactorDocTest",
Thread.currentThread.getContextClassLoader)

View file

@ -0,0 +1,99 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor;
import static org.junit.Assert.*;
import org.junit.Test;
//#imports
import akka.actor.*;
import scala.concurrent.Await;
import static akka.pattern.Patterns.ask;
import akka.transactor.Coordinated;
import akka.util.Timeout;
import static java.util.concurrent.TimeUnit.SECONDS;
//#imports
public class TransactorDocTest {
@Test
public void coordinatedExample() throws Exception {
//#coordinated-example
ActorSystem system = ActorSystem.create("CoordinatedExample");
ActorRef counter1 = system.actorOf(new Props(CoordinatedCounter.class));
ActorRef counter2 = system.actorOf(new Props(CoordinatedCounter.class));
Timeout timeout = new Timeout(5, SECONDS);
counter1.tell(new Coordinated(new Increment(counter2), timeout), null);
Integer count = (Integer) Await.result(ask(counter1, "GetCount", timeout), timeout.duration());
//#coordinated-example
assertEquals(count, new Integer(1));
system.shutdown();
}
@Test
public void coordinatedApi() {
//#create-coordinated
Timeout timeout = new Timeout(5, SECONDS);
Coordinated coordinated = new Coordinated(timeout);
//#create-coordinated
ActorSystem system = ActorSystem.create("CoordinatedApi");
ActorRef actor = system.actorOf(new Props(Coordinator.class));
//#send-coordinated
actor.tell(new Coordinated(new Message(), timeout), null);
//#send-coordinated
//#include-coordinated
actor.tell(coordinated.coordinate(new Message()), null);
//#include-coordinated
coordinated.await();
system.shutdown();
}
@Test
public void counterTransactor() throws Exception {
ActorSystem system = ActorSystem.create("CounterTransactor");
ActorRef counter = system.actorOf(new Props(Counter.class));
Timeout timeout = new Timeout(5, SECONDS);
Coordinated coordinated = new Coordinated(timeout);
counter.tell(coordinated.coordinate(new Increment()), null);
coordinated.await();
Integer count = (Integer) Await.result(ask(counter, "GetCount", timeout), timeout.duration());
assertEquals(count, new Integer(1));
system.shutdown();
}
@Test
public void friendlyCounterTransactor() throws Exception {
ActorSystem system = ActorSystem.create("FriendlyCounterTransactor");
ActorRef friend = system.actorOf(new Props(Counter.class));
ActorRef friendlyCounter = system.actorOf(new Props(FriendlyCounter.class));
Timeout timeout = new Timeout(5, SECONDS);
Coordinated coordinated = new Coordinated(timeout);
friendlyCounter.tell(coordinated.coordinate(new Increment(friend)), null);
coordinated.await();
Integer count1 = (Integer) Await.result(ask(friendlyCounter, "GetCount", timeout), timeout.duration());
assertEquals(count1, new Integer(1));
Integer count2 = (Integer) Await.result(ask(friend, "GetCount", timeout), timeout.duration());
assertEquals(count2, new Integer(1));
system.shutdown();
}
}

View file

@ -0,0 +1,8 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.zeromq
import org.scalatest.junit.JUnitSuite
class ZeromqDocTest extends ZeromqDocTestBase with JUnitSuite

View file

@ -0,0 +1,283 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.zeromq;
//#pub-socket
import akka.zeromq.Bind;
import akka.zeromq.ZeroMQExtension;
//#pub-socket
//#sub-socket
import akka.zeromq.Connect;
import akka.zeromq.Listener;
import akka.zeromq.Subscribe;
//#sub-socket
//#unsub-topic-socket
import akka.zeromq.Unsubscribe;
//#unsub-topic-socket
//#pub-topic
import akka.zeromq.Frame;
import akka.zeromq.ZMQMessage;
//#pub-topic
import akka.zeromq.HighWatermark;
import akka.zeromq.SocketOption;
import akka.zeromq.ZeroMQVersion;
//#health
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.actor.Props;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import scala.concurrent.util.Duration;
import akka.serialization.SerializationExtension;
import akka.serialization.Serialization;
import java.io.Serializable;
import java.lang.management.ManagementFactory;
//#health
import com.typesafe.config.ConfigFactory;
import java.lang.management.MemoryMXBean;
import java.lang.management.MemoryUsage;
import java.lang.management.OperatingSystemMXBean;
import java.util.Date;
import java.text.SimpleDateFormat;
import akka.actor.ActorSystem;
import akka.testkit.AkkaSpec;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.Assume;
public class ZeromqDocTestBase {
ActorSystem system;
@Before
public void setUp() {
system = ActorSystem.create("ZeromqDocTest",
ConfigFactory.parseString("akka.loglevel=INFO").withFallback(AkkaSpec.testConf()));
}
@After
public void tearDown() {
system.shutdown();
}
@Test
public void demonstrateCreateSocket() {
Assume.assumeTrue(checkZeroMQInstallation());
//#pub-socket
ActorRef pubSocket = ZeroMQExtension.get(system).newPubSocket(new Bind("tcp://127.0.0.1:1233"));
//#pub-socket
//#sub-socket
ActorRef listener = system.actorOf(new Props(ListenerActor.class));
ActorRef subSocket = ZeroMQExtension.get(system).newSubSocket(new Connect("tcp://127.0.0.1:1233"),
new Listener(listener), Subscribe.all());
//#sub-socket
//#sub-topic-socket
ActorRef subTopicSocket = ZeroMQExtension.get(system).newSubSocket(new Connect("tcp://127.0.0.1:1233"),
new Listener(listener), new Subscribe("foo.bar"));
//#sub-topic-socket
//#unsub-topic-socket
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)), null);
//#pub-topic
//#high-watermark
ActorRef highWatermarkSocket = ZeroMQExtension.get(system).newRouterSocket(
new SocketOption[] { new Listener(listener), new Bind("tcp://127.0.0.1:1233"), new HighWatermark(50000) });
//#high-watermark
}
@Test
public void demonstratePubSub() throws Exception {
Assume.assumeTrue(checkZeroMQInstallation());
//#health2
system.actorOf(new Props(HealthProbe.class), "health");
//#health2
//#logger2
system.actorOf(new Props(Logger.class), "logger");
//#logger2
//#alerter2
system.actorOf(new Props(HeapAlerter.class), "alerter");
//#alerter2
// Let it run for a while to see some output.
// Don't do like this in real tests, this is only doc demonstration.
Thread.sleep(3000L);
}
private boolean checkZeroMQInstallation() {
try {
ZeroMQVersion v = ZeroMQExtension.get(system).version();
return (v.major() == 2 && v.minor() == 1);
} catch (LinkageError e) {
return false;
}
}
//#listener-actor
public static class ListenerActor extends UntypedActor {
public void onReceive(Object message) throws Exception {
//...
}
}
//#listener-actor
//#health
public static final Object TICK = "TICK";
public static class Heap implements Serializable {
public final long timestamp;
public final long used;
public final long max;
public Heap(long timestamp, long used, long max) {
this.timestamp = timestamp;
this.used = used;
this.max = max;
}
}
public static class Load implements Serializable {
public final long timestamp;
public final double loadAverage;
public Load(long timestamp, double loadAverage) {
this.timestamp = timestamp;
this.loadAverage = loadAverage;
}
}
public static class HealthProbe extends UntypedActor {
ActorRef pubSocket = ZeroMQExtension.get(getContext().system()).newPubSocket(new Bind("tcp://127.0.0.1:1237"));
MemoryMXBean memory = ManagementFactory.getMemoryMXBean();
OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean();
Serialization ser = SerializationExtension.get(getContext().system());
@Override
public void preStart() {
getContext().system().scheduler()
.schedule(Duration.create(1, "second"), Duration.create(1, "second"), getSelf(), TICK, getContext().dispatcher());
}
@Override
public void postRestart(Throwable reason) {
// don't call preStart, only schedule once
}
@Override
public void onReceive(Object message) {
if (message.equals(TICK)) {
MemoryUsage currentHeap = memory.getHeapMemoryUsage();
long timestamp = System.currentTimeMillis();
// use akka SerializationExtension to convert to bytes
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)), 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)), getSelf());
} else {
unhandled(message);
}
}
}
//#health
//#logger
public static class Logger extends UntypedActor {
ActorRef subSocket = ZeroMQExtension.get(getContext().system()).newSubSocket(new Connect("tcp://127.0.0.1:1237"),
new Listener(getSelf()), new Subscribe("health"));
Serialization ser = SerializationExtension.get(getContext().system());
SimpleDateFormat timestampFormat = new SimpleDateFormat("HH:mm:ss.SSS");
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
@Override
public void onReceive(Object message) {
if (message instanceof ZMQMessage) {
ZMQMessage m = (ZMQMessage) message;
// the first frame is the topic, second is the message
if (m.firstFrameAsString().equals("health.heap")) {
Heap heap = (Heap) ser.serializerFor(Heap.class).fromBinary(m.payload(1));
log.info("Used heap {} bytes, at {}", heap.used, timestampFormat.format(new Date(heap.timestamp)));
} else if (m.firstFrameAsString().equals("health.load")) {
Load load = (Load) ser.serializerFor(Load.class).fromBinary(m.payload(1));
log.info("Load average {}, at {}", load.loadAverage, timestampFormat.format(new Date(load.timestamp)));
}
} else {
unhandled(message);
}
}
}
//#logger
//#alerter
public static class HeapAlerter extends UntypedActor {
ActorRef subSocket = ZeroMQExtension.get(getContext().system()).newSubSocket(new Connect("tcp://127.0.0.1:1237"),
new Listener(getSelf()), new Subscribe("health.heap"));
Serialization ser = SerializationExtension.get(getContext().system());
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
int count = 0;
@Override
public void onReceive(Object message) {
if (message instanceof ZMQMessage) {
ZMQMessage m = (ZMQMessage) message;
// the first frame is the topic, second is the message
if (m.firstFrameAsString().equals("health.heap")) {
Heap heap = (Heap) ser.serializerFor(Heap.class).fromBinary(m.payload(1));
if (((double) heap.used / heap.max) > 0.9) {
count += 1;
} else {
count = 0;
}
if (count > 10) {
log.warning("Need more memory, using {} %", (100.0 * heap.used / heap.max));
}
}
} else {
unhandled(message);
}
}
}
//#alerter
}