ActorContext.getParent for Java API #22413

akka-docs/rst/java/code/jdocs/persistence/query/LeveldbPersistenceQueryDocTest.java

@@ -0,0 +1,94 @@
+/**
+ * Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package jdocs.persistence.query;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import akka.NotUsed;
+import akka.actor.ActorSystem;
+import akka.persistence.journal.WriteEventAdapter;
+import akka.persistence.journal.Tagged;
+import akka.persistence.query.EventEnvelope;
+import akka.persistence.query.Sequence;
+import akka.persistence.query.PersistenceQuery;
+import akka.persistence.query.journal.leveldb.javadsl.LeveldbReadJournal;
+import akka.stream.ActorMaterializer;
+import akka.stream.javadsl.Source;
+
+public class LeveldbPersistenceQueryDocTest {
+
+  final ActorSystem system = ActorSystem.create();
+
+  public void demonstrateReadJournal() {
+    //#get-read-journal
+    final ActorMaterializer mat = ActorMaterializer.create(system);
+    
+    LeveldbReadJournal queries =
+      PersistenceQuery.get(system).getReadJournalFor(LeveldbReadJournal.class, 
+          LeveldbReadJournal.Identifier());
+    //#get-read-journal
+  }
+  
+  public void demonstrateEventsByPersistenceId() {
+    //#EventsByPersistenceId
+    LeveldbReadJournal queries =
+        PersistenceQuery.get(system).getReadJournalFor(LeveldbReadJournal.class, 
+            LeveldbReadJournal.Identifier());
+    
+    Source<EventEnvelope, NotUsed> source =
+        queries.eventsByPersistenceId("some-persistence-id", 0, Long.MAX_VALUE);
+    //#EventsByPersistenceId
+  }
+  
+  public void demonstrateAllPersistenceIds() {
+    //#AllPersistenceIds
+    LeveldbReadJournal queries =
+        PersistenceQuery.get(system).getReadJournalFor(LeveldbReadJournal.class, 
+            LeveldbReadJournal.Identifier());
+    
+    Source<String, NotUsed> source = queries.persistenceIds();
+    //#AllPersistenceIds
+  }
+  
+  public void demonstrateEventsByTag() {
+    //#EventsByTag
+    LeveldbReadJournal queries =
+        PersistenceQuery.get(system).getReadJournalFor(LeveldbReadJournal.class, 
+            LeveldbReadJournal.Identifier());
+    
+    Source<EventEnvelope, NotUsed> source =
+        queries.eventsByTag("green", new Sequence(0L));
+    //#EventsByTag
+  }
+  
+  static 
+  //#tagger
+  public class MyTaggingEventAdapter implements WriteEventAdapter {
+
+    @Override
+    public Object toJournal(Object event) {
+      if (event instanceof String) {
+        String s = (String) event;
+        Set<String> tags = new HashSet<String>();
+        if (s.contains("green")) tags.add("green");
+        if (s.contains("black")) tags.add("black");
+        if (s.contains("blue")) tags.add("blue");
+        if (tags.isEmpty())
+          return event;
+        else
+          return new Tagged(event, tags);
+      } else {
+        return event;
+      }
+    }
+    
+    @Override
+    public String manifest(Object event) {
+      return "";
+    }
+  }
+  //#tagger
+}

akka-docs/rst/java/code/jdocs/persistence/query/MyEventsByTagJavaPublisher.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package jdocs.persistence.query;
+
+import akka.actor.Cancellable;
+import akka.actor.Scheduler;
+import akka.japi.Pair;
+import akka.persistence.PersistentRepr;
+import akka.persistence.query.Offset;
+import akka.serialization.Serialization;
+import akka.serialization.SerializationExtension;
+import akka.stream.actor.AbstractActorPublisher;
+
+import akka.actor.Props;
+import akka.persistence.query.EventEnvelope;
+import akka.stream.actor.ActorPublisherMessage.Cancel;
+
+import scala.concurrent.duration.FiniteDuration;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+//#events-by-tag-publisher
+class MyEventsByTagJavaPublisher extends AbstractActorPublisher<EventEnvelope> {
+  private final Serialization serialization =
+    SerializationExtension.get(getContext().getSystem());
+
+  private final Connection connection;
+
+  private final String tag;
+
+  private final String CONTINUE = "CONTINUE";
+  private final int LIMIT = 1000;
+  private long currentOffset;
+  private List<EventEnvelope> buf = new LinkedList<>();
+
+  private Cancellable continueTask;
+
+  public MyEventsByTagJavaPublisher(Connection connection,
+                                    String tag,
+                                    Long offset,
+                                    FiniteDuration refreshInterval) {
+    this.connection = connection;
+    this.tag = tag;
+    this.currentOffset = offset;
+
+    final Scheduler scheduler = getContext().getSystem().scheduler();
+    this.continueTask = scheduler
+      .schedule(refreshInterval, refreshInterval, getSelf(), CONTINUE,
+                getContext().dispatcher(), getSelf());
+  }
+  
+  @Override
+  public Receive createReceive() {
+    return receiveBuilder()
+      .matchEquals(CONTINUE, (in) -> {
+        query();
+        deliverBuf();
+      })
+      .match(Cancel.class, (in) -> {
+        getContext().stop(getSelf());
+      })
+      .build();
+  }
+
+  public static Props props(Connection conn, String tag, Long offset, 
+      FiniteDuration refreshInterval) {
+    return Props.create(() -> 
+      new MyEventsByTagJavaPublisher(conn, tag, offset, refreshInterval));
+  }
+
+  @Override
+  public void postStop() {
+    continueTask.cancel();
+  }
+
+  private void query() {
+    if (buf.isEmpty()) {
+      final String query = "SELECT id, persistent_repr " +
+        "FROM journal WHERE tag = ? AND id > ? " +
+        "ORDER BY id LIMIT ?";
+
+      try (PreparedStatement s = connection.prepareStatement(query)) {
+        s.setString(1, tag);
+        s.setLong(2, currentOffset);
+        s.setLong(3, LIMIT);
+        try (ResultSet rs = s.executeQuery()) {
+
+          final List<Pair<Long, byte[]>> res = new ArrayList<>(LIMIT);
+          while (rs.next())
+            res.add(Pair.create(rs.getLong(1), rs.getBytes(2)));
+
+          if (!res.isEmpty()) {
+            currentOffset = res.get(res.size() - 1).first();
+          }
+
+          buf = res.stream().map(in -> {
+            final Long id = in.first();
+            final byte[] bytes = in.second();
+
+            final PersistentRepr p = 
+                serialization.deserialize(bytes, PersistentRepr.class).get();
+
+            return new EventEnvelope(Offset.sequence(id), p.persistenceId(), p.sequenceNr(), p.payload());
+          }).collect(toList());
+        }
+      } catch(Exception e) {
+          onErrorThenStop(e);
+      }
+    }
+  }
+
+  private void deliverBuf() {
+    while (totalDemand() > 0 && !buf.isEmpty())
+      onNext(buf.remove(0));
+  }
+}
+//#events-by-tag-publisher