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

=dur #3664 Deprecate durable mailboxes

Browse source
This commit is contained in:
Patrik Nordwall 2013-12-11 13:38:13 +01:00
parent a5b52e9bea
commit 4bd1586b1e
33 changed files with 41 additions and 561 deletions
Download patch file Download diff file Expand all files Collapse all files

44
akka-docs/rst/java/code/docs/actor/mailbox/DurableMailboxDocTest.java
View file

@ -1,44 +0,0 @@
/**
* Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor.mailbox;
//#imports
import akka.actor.Props;
import akka.actor.ActorRef;
//#imports
import akka.testkit.AkkaJUnitActorSystemResource;
import org.junit.ClassRule;
import org.junit.Test;
import akka.testkit.AkkaSpec;
import com.typesafe.config.ConfigFactory;
import akka.actor.ActorSystem;
import akka.actor.UntypedActor;
public class DurableMailboxDocTest {
@ClassRule
public static AkkaJUnitActorSystemResource actorSystemResource =
new AkkaJUnitActorSystemResource("DurableMailboxDocTest",
ConfigFactory.parseString(DurableMailboxDocSpec.config()).withFallback(AkkaSpec.testConf()));
private final ActorSystem system = actorSystemResource.getSystem();
@Test
public void configDefinedDispatcher() {
//#dispatcher-config-use
ActorRef myActor = system.actorOf(Props.create(MyUntypedActor.class).
withDispatcher("my-dispatcher"), "myactor");
//#dispatcher-config-use
myActor.tell("test", ActorRef.noSender());
}
public static class MyUntypedActor extends UntypedActor {
public void onReceive(Object message) {
}
}
}

31
akka-docs/rst/java/code/docs/actor/mailbox/MyDurableMailboxDocSpec.scala
View file

@ -1,31 +0,0 @@
/**
* Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor.mailbox
import akka.actor.mailbox.DurableMailboxSpec
object MyDurableMailboxDocSpec {
val config = """
MyStorage-dispatcher {
mailbox-type = docs.actor.mailbox.MyDurableMailboxType
}
"""
}
class MyDurableMailboxDocSpec extends DurableMailboxSpec("MyStorage", MyDurableMailboxDocSpec.config) {
override def atStartup() {
}
override def afterTermination() {
}
"MyDurableMailbox (Java)" must {
"deliver a message" in {
val actor = createMailboxTestActor()
implicit val sender = testActor
actor ! "hello"
expectMsg("hello")
}
}
}

30
akka-docs/rst/java/code/docs/actor/mailbox/MyDurableMailboxType.java
View file

@ -1,30 +0,0 @@
/**
* Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor.mailbox;
//#custom-mailbox-type
import scala.Option;
import com.typesafe.config.Config;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.ExtendedActorSystem;
import akka.dispatch.MailboxType;
import akka.dispatch.MessageQueue;
public class MyDurableMailboxType implements MailboxType {
public MyDurableMailboxType(ActorSystem.Settings settings, Config config) {
}
@Override
public MessageQueue create(Option<ActorRef> owner,
Option<ActorSystem> system) {
if (owner.isEmpty())
throw new IllegalArgumentException("requires an owner " +
"(i.e. does not work with BalancingDispatcher)");
return new MyDurableMessageQueue(owner.get(),
(ExtendedActorSystem) system.get());
}
}
//#custom-mailbox-type

96
akka-docs/rst/java/code/docs/actor/mailbox/MyDurableMessageQueue.java
View file

@ -1,96 +0,0 @@
/**
* Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor.mailbox;
//#durable-message-queue
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Callable;
import scala.concurrent.duration.Duration;
import akka.actor.ActorRef;
import akka.actor.ExtendedActorSystem;
import akka.actor.mailbox.DurableMessageQueueWithSerialization;
import akka.dispatch.Envelope;
import akka.dispatch.MessageQueue;
import akka.pattern.CircuitBreaker;
public class MyDurableMessageQueue extends DurableMessageQueueWithSerialization {
public MyDurableMessageQueue(ActorRef owner, ExtendedActorSystem system) {
super(owner, system);
}
private final QueueStorage storage = new QueueStorage();
// A real-world implementation would use configuration to set the last
// three parameters below
private final CircuitBreaker breaker = CircuitBreaker.create(system().scheduler(),
5, Duration.create(30, "seconds"), Duration.create(1, "minute"));
@Override
public void enqueue(ActorRef receiver, final Envelope envelope) {
breaker.callWithSyncCircuitBreaker(new Callable<Object>() {
@Override
public Object call() {
byte[] data = serialize(envelope);
storage.push(data);
return null;
}
});
}
@Override
public Envelope dequeue() {
return breaker.callWithSyncCircuitBreaker(new Callable<Envelope>() {
@Override
public Envelope call() {
byte[] data = storage.pull();
if (data == null)
return null;
else
return deserialize(data);
}
});
}
@Override
public boolean hasMessages() {
return breaker.callWithSyncCircuitBreaker(new Callable<Boolean>() {
@Override
public Boolean call() {
return !storage.isEmpty();
}
});
}
@Override
public int numberOfMessages() {
return breaker.callWithSyncCircuitBreaker(new Callable<Integer>() {
@Override
public Integer call() {
return storage.size();
}
});
}
/**
* Called when the mailbox is disposed.
* An ordinary mailbox would send remaining messages to deadLetters,
* but the purpose of a durable mailbox is to continue
* with the same message queue when the actor is started again.
*/
@Override
public void cleanUp(ActorRef owner, MessageQueue deadLetters) {}
//#dummy-queue-storage
// dummy
private static class QueueStorage {
private final ConcurrentLinkedQueue<byte[]> queue =
new ConcurrentLinkedQueue<byte[]>();
public void push(byte[] data) { queue.offer(data); }
public byte[] pull() { return queue.poll(); }
public boolean isEmpty() { return queue.isEmpty(); }
public int size() { return queue.size(); }
}
//#dummy-queue-storage
}
//#durable-message-queue

1
akka-docs/rst/java/code/docs/testkit/TestKitDocTest.java
View file

@ -6,7 +6,6 @@ package docs.testkit;
import static org.junit.Assert.*;
import akka.testkit.*;
import docs.actor.mailbox.DurableMailboxDocSpec;
import org.junit.ClassRule;
import org.junit.Test;