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

+act,sam,doc #3940 Added receive setter for Java Lambda actors

Browse source 
* Added a setter for Java lambda actors to "hide" the not so nice looking type signature of the "receive" method.
* Updated docs to reflect the changes.
* Converted samples to use the new setter.
This commit is contained in:
Björn Antonsson 2014-03-20 12:05:32 +01:00
parent 30ae25a90c
commit 723c931d16
23 changed files with 683 additions and 570 deletions
Download patch file Download diff file Expand all files Collapse all files

227
akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java
View file

@ -35,24 +35,25 @@ public class LambdaPersistenceDocTest {
static Object o1 = new Object() {
//#definition
class MyProcessor extends AbstractProcessor {
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> {
// message successfully written to journal
Object payload = p.payload();
Long sequenceNr = p.sequenceNr();
// ...
}).
match(PersistenceFailure.class, failure -> {
// message failed to be written to journal
Object payload = failure.payload();
Long sequenceNr = failure.sequenceNr();
Throwable cause = failure.cause();
// ...
}).
matchAny(otherwise -> {
// message not written to journal
}).build();
public MyProcessor() {
receive(ReceiveBuilder.
match(Persistent.class, p -> {
// message successfully written to journal
Object payload = p.payload();
Long sequenceNr = p.sequenceNr();
// ...
}).
match(PersistenceFailure.class, failure -> {
// message failed to be written to journal
Object payload = failure.payload();
Long sequenceNr = failure.sequenceNr();
Throwable cause = failure.cause();
// ...
}).
matchAny(otherwise -> {
// message not written to journal
}).build()
);
}
}
//#definition
@ -67,11 +68,10 @@ public class LambdaPersistenceDocTest {
processor.tell(Persistent.create("foo"), null);
processor.tell("bar", null);
//#usage
}
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, received -> {/* ... */}).build();
receive(ReceiveBuilder.
match(Persistent.class, received -> {/* ... */}).build()
);
}
private void recover() {
@ -124,9 +124,10 @@ public class LambdaPersistenceDocTest {
}
//#processor-id-override
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, received -> {/* ... */}).build();
public MyProcessor4() {
receive(ReceiveBuilder.
match(Persistent.class, received -> {/* ... */}).build()
);
}
}
};
@ -140,27 +141,27 @@ public class LambdaPersistenceDocTest {
public MyProcessor() {
this.destination = context().actorOf(Props.create(MyDestination.class));
this.channel = context().actorOf(Channel.props(), "myChannel");
}
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> {
Persistent out = p.withPayload("done " + p.payload());
channel.tell(Deliver.create(out, destination.path()), self());
}).build();
receive(ReceiveBuilder.
match(Persistent.class, p -> {
Persistent out = p.withPayload("done " + p.payload());
channel.tell(Deliver.create(out, destination.path()), self());
}).build()
);
}
}
class MyDestination extends AbstractActor {
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(ConfirmablePersistent.class, p -> {
Object payload = p.payload();
Long sequenceNr = p.sequenceNr();
int redeliveries = p.redeliveries();
// ...
p.confirm();
}).build();
public MyDestination() {
receive(ReceiveBuilder.
match(ConfirmablePersistent.class, p -> {
Object payload = p.payload();
Long sequenceNr = p.sequenceNr();
int redeliveries = p.redeliveries();
// ...
p.confirm();
}).build()
);
}
}
//#channel-example
@ -177,39 +178,38 @@ public class LambdaPersistenceDocTest {
//#channel-custom-settings
context().actorOf(
Channel.props(ChannelSettings.create()
.withRedeliverInterval(Duration.create(30, TimeUnit.SECONDS))
.withRedeliverMax(15)));
Channel.props(ChannelSettings.create()
.withRedeliverInterval(Duration.create(30, TimeUnit.SECONDS))
.withRedeliverMax(15)));
//#channel-custom-settings
//#channel-custom-listener
class MyListener extends AbstractActor {
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(RedeliverFailure.class, r -> {
Iterable<ConfirmablePersistent> messages = r.getMessages();
// ...
}).build();
public MyListener() {
receive(ReceiveBuilder.
match(RedeliverFailure.class, r -> {
Iterable<ConfirmablePersistent> messages = r.getMessages();
// ...
}).build()
);
}
}
final ActorRef myListener = context().actorOf(Props.create(MyListener.class));
context().actorOf(Channel.props(
ChannelSettings.create().withRedeliverFailureListener(null)));
ChannelSettings.create().withRedeliverFailureListener(null)));
//#channel-custom-listener
}
receive(ReceiveBuilder.
match(Persistent.class, p -> {
Persistent out = p.withPayload("done " + p.payload());
channel.tell(Deliver.create(out, destination.path()), self());
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> {
Persistent out = p.withPayload("done " + p.payload());
channel.tell(Deliver.create(out, destination.path()), self());
//#channel-example-reply
channel.tell(Deliver.create(out, sender().path()), self());
//#channel-example-reply
}).build();
//#channel-example-reply
channel.tell(Deliver.create(out, sender().path()), self());
//#channel-example-reply
}).build()
);
}
}
};
@ -219,18 +219,19 @@ public class LambdaPersistenceDocTest {
class MyProcessor extends AbstractProcessor {
private Object state;
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(String.class, s -> s.equals("snap"),
s -> saveSnapshot(state)).
match(SaveSnapshotSuccess.class, ss -> {
SnapshotMetadata metadata = ss.metadata();
// ...
}).
match(SaveSnapshotFailure.class, sf -> {
SnapshotMetadata metadata = sf.metadata();
// ...
}).build();
public MyProcessor() {
receive(ReceiveBuilder.
match(String.class, s -> s.equals("snap"),
s -> saveSnapshot(state)).
match(SaveSnapshotSuccess.class, ss -> {
SnapshotMetadata metadata = ss.metadata();
// ...
}).
match(SaveSnapshotFailure.class, sf -> {
SnapshotMetadata metadata = sf.metadata();
// ...
}).build()
);
}
}
//#save-snapshot
@ -241,13 +242,14 @@ public class LambdaPersistenceDocTest {
class MyProcessor extends AbstractProcessor {
private Object state;
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(SnapshotOffer.class, s -> {
state = s.snapshot();
// ...
}).
match(Persistent.class, p -> {/* ...*/}).build();
public MyProcessor() {
receive(ReceiveBuilder.
match(SnapshotOffer.class, s -> {
state = s.snapshot();
// ...
}).
match(Persistent.class, p -> {/* ...*/}).build()
);
}
}
//#snapshot-offer
@ -257,17 +259,16 @@ public class LambdaPersistenceDocTest {
public MyActor() {
processor = context().actorOf(Props.create(MyProcessor.class));
}
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.match(Object.class, o -> {/* ... */}).build();
receive(ReceiveBuilder.
match(Object.class, o -> {/* ... */}).build()
);
}
private void recover() {
//#snapshot-criteria
processor.tell(Recover.create(
SnapshotSelectionCriteria
.create(457L, System.currentTimeMillis())), null);
SnapshotSelectionCriteria
.create(457L, System.currentTimeMillis())), null);
//#snapshot-criteria
}
}
@ -276,12 +277,13 @@ public class LambdaPersistenceDocTest {
static Object o6 = new Object() {
//#batch-write
class MyProcessor extends AbstractProcessor {
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> p.payload().equals("a"),
p -> {/* ... */}).
match(Persistent.class, p -> p.payload().equals("b"),
p -> {/* ... */}).build();
public MyProcessor() {
receive(ReceiveBuilder.
match(Persistent.class, p -> p.payload().equals("a"),
p -> {/* ... */}).
match(Persistent.class, p -> p.payload().equals("b"),
p -> {/* ... */}).build()
);
}
}
@ -291,8 +293,8 @@ public class LambdaPersistenceDocTest {
public void batchWrite() {
processor.tell(PersistentBatch
.create(asList(Persistent.create("a"),
Persistent.create("b"))), null);
.create(asList(Persistent.create("a"),
Persistent.create("b"))), null);
}
// ...
@ -307,18 +309,18 @@ public class LambdaPersistenceDocTest {
public void foo() {
//#persistent-channel-example
final ActorRef channel = context().actorOf(
PersistentChannel.props(
PersistentChannelSettings.create()
.withRedeliverInterval(Duration.create(30, TimeUnit.SECONDS))
.withRedeliverMax(15)),
"myPersistentChannel");
PersistentChannel.props(
PersistentChannelSettings.create()
.withRedeliverInterval(Duration.create(30, TimeUnit.SECONDS))
.withRedeliverMax(15)),
"myPersistentChannel");
channel.tell(Deliver.create(Persistent.create("example"), destination.path()), self());
//#persistent-channel-example
//#persistent-channel-watermarks
PersistentChannelSettings.create()
.withPendingConfirmationsMax(10000)
.withPendingConfirmationsMin(2000);
.withPendingConfirmationsMax(10000)
.withPendingConfirmationsMin(2000);
//#persistent-channel-watermarks
//#persistent-channel-reply
PersistentChannelSettings.create().withReplyPersistent(true);
@ -343,19 +345,19 @@ public class LambdaPersistenceDocTest {
// ...
// reliably deliver events
channel.tell(Deliver.create(
Persistent.create(event, getCurrentPersistentMessage()),
destination.path()), self());
Persistent.create(event, getCurrentPersistentMessage()),
destination.path()), self());
}
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(String.class, this::handleEvent).build();
match(String.class, this::handleEvent).build();
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, s -> s.equals("cmd"),
s -> persist("evt", this::handleEvent)).build();
match(String.class, s -> s.equals("cmd"),
s -> persist("evt", this::handleEvent)).build();
}
}
//#reliable-event-delivery
@ -369,11 +371,12 @@ public class LambdaPersistenceDocTest {
return "some-processor-id";
}
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, peristent -> {
// ...
}).build();
public MyView() {
receive(ReceiveBuilder.
match(Persistent.class, persistent -> {
// ...
}).build()
);
}
}
//#view

19
akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistencePluginDocTest.java
View file

@ -37,16 +37,17 @@ public class LambdaPersistencePluginDocTest {
selection.tell(new Identify(1), self());
}
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(ActorIdentity.class, ai -> {
if (ai.correlationId().equals(1)) {
ActorRef store = ai.getRef();
if (store != null) {
SharedLeveldbJournal.setStore(store, context().system());
}
public SharedStorageUsage() {
receive(ReceiveBuilder.
match(ActorIdentity.class, ai -> {
if (ai.correlationId().equals(1)) {
ActorRef store = ai.getRef();
if (store != null) {
SharedLeveldbJournal.setStore(store, context().system());
}
}).build();
}
}).build()
);
}
}
//#shared-store-usage

100
akka-samples/akka-sample-persistence-java-lambda/src/main/java/sample/persistence/ConversationRecoveryExample.java
View file

@ -13,62 +13,64 @@ import scala.PartialFunction;
import scala.runtime.BoxedUnit;
public class ConversationRecoveryExample {
public static String PING = "PING";
public static String PONG = "PONG";
public static String PING = "PING";
public static String PONG = "PONG";
public static class Ping extends AbstractProcessor {
final ActorRef pongChannel = context().actorOf(Channel.props(), "pongChannel");
int counter = 0;
public static class Ping extends AbstractProcessor {
final ActorRef pongChannel = context().actorOf(Channel.props(), "pongChannel");
int counter = 0;
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> cp.payload().equals(PING), cp -> {
counter += 1;
System.out.println(String.format("received ping %d times", counter));
cp.confirm();
if (!recoveryRunning()) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
pongChannel.tell(Deliver.create(cp.withPayload(PONG), sender().path()), self());
}).
match(String.class,
s -> s.equals("init"),
s -> pongChannel.tell(Deliver.create(Persistent.create(PONG), sender().path()), self())).build();
}
public Ping() {
receive(ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> cp.payload().equals(PING), cp -> {
counter += 1;
System.out.println(String.format("received ping %d times", counter));
cp.confirm();
if (!recoveryRunning()) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
pongChannel.tell(Deliver.create(cp.withPayload(PONG), sender().path()), self());
}).
match(String.class,
s -> s.equals("init"),
s -> pongChannel.tell(Deliver.create(Persistent.create(PONG), sender().path()), self())).build()
);
}
}
public static class Pong extends AbstractProcessor {
private final ActorRef pingChannel = context().actorOf(Channel.props(), "pingChannel");
private int counter = 0;
public static class Pong extends AbstractProcessor {
private final ActorRef pingChannel = context().actorOf(Channel.props(), "pingChannel");
private int counter = 0;
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> cp.payload().equals(PONG), cp -> {
counter += 1;
System.out.println(String.format("received pong %d times", counter));
cp.confirm();
if (!recoveryRunning()) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
pingChannel.tell(Deliver.create(cp.withPayload(PING), sender().path()), self());
}).build();
}
public Pong() {
receive(ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> cp.payload().equals(PONG), cp -> {
counter += 1;
System.out.println(String.format("received pong %d times", counter));
cp.confirm();
if (!recoveryRunning()) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
pingChannel.tell(Deliver.create(cp.withPayload(PING), sender().path()), self());
}).build()
);
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef ping = system.actorOf(Props.create(Ping.class), "ping");
final ActorRef pong = system.actorOf(Props.create(Pong.class), "pong");
final ActorRef ping = system.actorOf(Props.create(Ping.class), "ping");
final ActorRef pong = system.actorOf(Props.create(Pong.class), "pong");
ping.tell("init", pong);
}
ping.tell("init", pong);
}
}

6
akka-samples/akka-sample-persistence-java-lambda/src/main/java/sample/persistence/EventsourcedExample.java
View file

@ -80,13 +80,15 @@ class ExampleProcessor extends AbstractEventsourcedProcessor {
return state.size();
}
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
@Override
public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(Evt.class, state::update).
match(SnapshotOffer.class, ss -> state = (ExampleState) ss.snapshot()).build();
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
@Override
public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.match(Cmd.class, c -> {
final String data = c.getData();
final Evt evt1 = new Evt(data + "-" + getNumEvents());

66
akka-samples/akka-sample-persistence-java-lambda/src/main/java/sample/persistence/ProcessorChannelExample.java
View file

@ -14,45 +14,45 @@ import scala.PartialFunction;
import scala.runtime.BoxedUnit;
public class ProcessorChannelExample {
public static class ExampleProcessor extends AbstractProcessor {
private ActorRef destination;
private ActorRef channel;
public static class ExampleProcessor extends AbstractProcessor {
private ActorRef destination;
private ActorRef channel;
public ExampleProcessor(ActorRef destination) {
this.destination = destination;
this.channel = context().actorOf(Channel.props(), "channel");
}
public ExampleProcessor(ActorRef destination) {
this.destination = destination;
this.channel = context().actorOf(Channel.props(), "channel");
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> {
System.out.println("processed " + p.payload());
channel.tell(Deliver.create(p.withPayload("processed " + p.payload()), destination.path()), self());
}).
match(String.class, s -> System.out.println("reply = " + s)).build();
}
receive(ReceiveBuilder.
match(Persistent.class, p -> {
System.out.println("processed " + p.payload());
channel.tell(Deliver.create(p.withPayload("processed " + p.payload()), destination.path()), self());
}).
match(String.class, s -> System.out.println("reply = " + s)).build()
);
}
}
public static class ExampleDestination extends AbstractActor {
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> {
System.out.println("received " + cp.payload());
sender().tell(String.format("re: %s (%d)", cp.payload(), cp.sequenceNr()), null);
cp.confirm();
}).build();
}
public static class ExampleDestination extends AbstractActor {
public ExampleDestination() {
receive(ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> {
System.out.println("received " + cp.payload());
sender().tell(String.format("re: %s (%d)", cp.payload(), cp.sequenceNr()), null);
cp.confirm();
}).build()
);
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef destination = system.actorOf(Props.create(ExampleDestination.class));
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class, destination), "processor-1");
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef destination = system.actorOf(Props.create(ExampleDestination.class));
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class, destination), "processor-1");
processor.tell(Persistent.create("a"), null);
processor.tell(Persistent.create("b"), null);
processor.tell(Persistent.create("a"), null);
processor.tell(Persistent.create("b"), null);
Thread.sleep(1000);
system.shutdown();
}
Thread.sleep(1000);
system.shutdown();
}
}

97
akka-samples/akka-sample-persistence-java-lambda/src/main/java/sample/persistence/ProcessorFailureExample.java
View file

@ -17,56 +17,57 @@ import scala.runtime.BoxedUnit;
import java.util.ArrayList;
public class ProcessorFailureExample {
public static class ExampleProcessor extends AbstractProcessor {
private ArrayList<Object> received = new ArrayList<Object>();
public static class ExampleProcessor extends AbstractProcessor {
private ArrayList<Object> received = new ArrayList<Object>();
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> p.payload().equals("boom"), p -> {throw new RuntimeException("boom");}).
match(Persistent.class, p -> !p.payload().equals("boom"), p -> received.add(p.payload())).
match(String.class, s -> s.equals("boom"), s -> {throw new RuntimeException("boom");}).
match(String.class, s -> s.equals("print"), s -> System.out.println("received " + received)).build();
}
@Override
public void preRestart(Throwable reason, Option<Object> message) {
if (message.isDefined() && message.get() instanceof Persistent) {
deleteMessage(((Persistent) message.get()).sequenceNr(), false);
}
super.preRestart(reason, message);
}
public ExampleProcessor() {
receive(ReceiveBuilder.
match(Persistent.class, p -> p.payload().equals("boom"), p -> {throw new RuntimeException("boom");}).
match(Persistent.class, p -> !p.payload().equals("boom"), p -> received.add(p.payload())).
match(String.class, s -> s.equals("boom"), s -> {throw new RuntimeException("boom");}).
match(String.class, s -> s.equals("print"), s -> System.out.println("received " + received)).build()
);
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class), "processor-2");
processor.tell(Persistent.create("a"), null);
processor.tell("print", null);
processor.tell("boom", null);
processor.tell("print", null);
processor.tell(Persistent.create("b"), null);
processor.tell("print", null);
processor.tell(Persistent.create("boom"), null);
processor.tell("print", null);
processor.tell(Persistent.create("c"), null);
processor.tell("print", null);
// Will print in a first run (i.e. with empty journal):
// received [a]
// received [a, b]
// received [a, b, c]
// Will print in a second run:
// received [a, b, c, a]
// received [a, b, c, a, b]
// received [a, b, c, a, b, c]
// etc ...
Thread.sleep(1000);
system.shutdown();
@Override
public void preRestart(Throwable reason, Option<Object> message) {
if (message.isDefined() && message.get() instanceof Persistent) {
deleteMessage(((Persistent) message.get()).sequenceNr(), false);
}
super.preRestart(reason, message);
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class), "processor-2");
processor.tell(Persistent.create("a"), null);
processor.tell("print", null);
processor.tell("boom", null);
processor.tell("print", null);
processor.tell(Persistent.create("b"), null);
processor.tell("print", null);
processor.tell(Persistent.create("boom"), null);
processor.tell("print", null);
processor.tell(Persistent.create("c"), null);
processor.tell("print", null);
// Will print in a first run (i.e. with empty journal):
// received [a]
// received [a, b]
// received [a, b, c]
// Will print in a second run:
// received [a, b, c, a]
// received [a, b, c, a, b]
// received [a, b, c, a, b, c]
// etc ...
Thread.sleep(1000);
system.shutdown();
}
}

105
akka-samples/akka-sample-persistence-java-lambda/src/main/java/sample/persistence/SnapshotExample.java
View file

@ -18,62 +18,63 @@ import java.io.Serializable;
import java.util.ArrayList;
public class SnapshotExample {
public static class ExampleState implements Serializable {
private final ArrayList<String> received;
public static class ExampleState implements Serializable {
private final ArrayList<String> received;
public ExampleState() {
this(new ArrayList<String>());
}
public ExampleState(ArrayList<String> received) {
this.received = received;
}
public ExampleState copy() {
return new ExampleState(new ArrayList<String>(received));
}
public void update(String s) {
received.add(s);
}
@Override
public String toString() {
return received.toString();
}
public ExampleState() {
this(new ArrayList<String>());
}
public static class ExampleProcessor extends AbstractProcessor {
private ExampleState state = new ExampleState();
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> state.update(String.format("%s-%d", p.payload(), p.sequenceNr()))).
match(SnapshotOffer.class, s -> {
ExampleState exState = (ExampleState) s.snapshot();
System.out.println("offered state = " + exState);
state = exState;
}).
match(String.class, s -> s.equals("print"), s -> System.out.println("current state = " + state)).
match(String.class, s -> s.equals("snap"), s ->
// IMPORTANT: create a copy of snapshot
// because ExampleState is mutable !!!
saveSnapshot(state.copy())).build();
}
public ExampleState(ArrayList<String> received) {
this.received = received;
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class), "processor-3-java");
processor.tell(Persistent.create("a"), null);
processor.tell(Persistent.create("b"), null);
processor.tell("snap", null);
processor.tell(Persistent.create("c"), null);
processor.tell(Persistent.create("d"), null);
processor.tell("print", null);
Thread.sleep(1000);
system.shutdown();
public ExampleState copy() {
return new ExampleState(new ArrayList<String>(received));
}
public void update(String s) {
received.add(s);
}
@Override
public String toString() {
return received.toString();
}
}
public static class ExampleProcessor extends AbstractProcessor {
private ExampleState state = new ExampleState();
public ExampleProcessor() {
receive(ReceiveBuilder.
match(Persistent.class, p -> state.update(String.format("%s-%d", p.payload(), p.sequenceNr()))).
match(SnapshotOffer.class, s -> {
ExampleState exState = (ExampleState) s.snapshot();
System.out.println("offered state = " + exState);
state = exState;
}).
match(String.class, s -> s.equals("print"), s -> System.out.println("current state = " + state)).
match(String.class, s -> s.equals("snap"), s ->
// IMPORTANT: create a copy of snapshot
// because ExampleState is mutable !!!
saveSnapshot(state.copy())).build()
);
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class), "processor-3-java");
processor.tell(Persistent.create("a"), null);
processor.tell(Persistent.create("b"), null);
processor.tell("snap", null);
processor.tell(Persistent.create("c"), null);
processor.tell(Persistent.create("d"), null);
processor.tell("print", null);
Thread.sleep(1000);
system.shutdown();
}
}

143
akka-samples/akka-sample-persistence-java-lambda/src/main/java/sample/persistence/ViewExample.java
View file

@ -17,84 +17,87 @@ import scala.runtime.BoxedUnit;
import java.util.concurrent.TimeUnit;
public class ViewExample {
public static class ExampleProcessor extends AbstractProcessor {
@Override
public String processorId() {
return "processor-5";
}
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class,
p -> System.out.println(String.format("processor received %s (sequence nr = %d)",
p.payload(),
p.sequenceNr()))).build();
}
public static class ExampleProcessor extends AbstractProcessor {
@Override
public String processorId() {
return "processor-5";
}
public static class ExampleView extends AbstractView {
private final ActorRef destination = context().actorOf(Props.create(ExampleDestination.class));
private final ActorRef channel = context().actorOf(Channel.props("channel"));
public ExampleProcessor() {
receive(ReceiveBuilder.
match(Persistent.class,
p -> System.out.println(String.format("processor received %s (sequence nr = %d)",
p.payload(),
p.sequenceNr()))).build()
);
}
}
private int numReplicated = 0;
public static class ExampleView extends AbstractView {
private final ActorRef destination = context().actorOf(Props.create(ExampleDestination.class));
private final ActorRef channel = context().actorOf(Channel.props("channel"));
@Override
public String viewId() {
return "view-5";
}
private int numReplicated = 0;
@Override
public String processorId() {
return "processor-5";
}
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(Persistent.class, p -> {
numReplicated += 1;
System.out.println(String.format("view received %s (sequence nr = %d, num replicated = %d)",
p.payload(),
p.sequenceNr(),
numReplicated));
channel.tell(Deliver.create(p.withPayload("replicated-" + p.payload()), destination.path()),
self());
}).
match(SnapshotOffer.class, so -> {
numReplicated = (Integer) so.snapshot();
System.out.println(String.format("view received snapshot offer %s (metadata = %s)",
numReplicated,
so.metadata()));
}).
match(String.class, s -> s.equals("snap"), s -> saveSnapshot(numReplicated)).build();
}
@Override
public String viewId() {
return "view-5";
}
public static class ExampleDestination extends AbstractActor {
@Override public PartialFunction<Object, BoxedUnit> receive() {
return ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> {
System.out.println(String.format("destination received %s (sequence nr = %s)",
cp.payload(),
cp.sequenceNr()));
cp.confirm();
}).build();
}
@Override
public String processorId() {
return "processor-5";
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class));
final ActorRef view = system.actorOf(Props.create(ExampleView.class));
system.scheduler()
.schedule(Duration.Zero(),
Duration.create(2, TimeUnit.SECONDS),
processor,
Persistent.create("scheduled"),
system.dispatcher(),
null);
system.scheduler()
.schedule(Duration.Zero(), Duration.create(5, TimeUnit.SECONDS), view, "snap", system.dispatcher(), null);
public ExampleView() {
receive(ReceiveBuilder.
match(Persistent.class, p -> {
numReplicated += 1;
System.out.println(String.format("view received %s (sequence nr = %d, num replicated = %d)",
p.payload(),
p.sequenceNr(),
numReplicated));
channel.tell(Deliver.create(p.withPayload("replicated-" + p.payload()), destination.path()),
self());
}).
match(SnapshotOffer.class, so -> {
numReplicated = (Integer) so.snapshot();
System.out.println(String.format("view received snapshot offer %s (metadata = %s)",
numReplicated,
so.metadata()));
}).
match(String.class, s -> s.equals("snap"), s -> saveSnapshot(numReplicated)).build()
);
}
}
public static class ExampleDestination extends AbstractActor {
public ExampleDestination() {
receive(ReceiveBuilder.
match(ConfirmablePersistent.class, cp -> {
System.out.println(String.format("destination received %s (sequence nr = %s)",
cp.payload(),
cp.sequenceNr()));
cp.confirm();
}).build()
);
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class));
final ActorRef view = system.actorOf(Props.create(ExampleView.class));
system.scheduler()
.schedule(Duration.Zero(),
Duration.create(2, TimeUnit.SECONDS),
processor,
Persistent.create("scheduled"),
system.dispatcher(),
null);
system.scheduler()
.schedule(Duration.Zero(), Duration.create(5, TimeUnit.SECONDS), view, "snap", system.dispatcher(), null);
}
}