Persistence Query

public class RichEvent {
  public final Set<String >tags;
  public final Object payload;
  
  public RichEvent(Set<String> tags, Object payload) {
    this.tags = tags;
    this.payload = payload;
  }
}
// a plugin can provide:
public final class QueryMetadata{
  public final boolean deterministicOrder;
  public final boolean infinite;

  public QueryMetadata(boolean deterministicOrder, boolean infinite) {
    this.deterministicOrder = deterministicOrder;
    this.infinite = infinite;
  }
}

public Source<RichEvent, QueryMetadata> byTagsWithMeta(Set<String> tags) {

Set<String> tags = new HashSet<String>();
tags.add("red");
tags.add("blue");
final Source<RichEvent, QueryMetadata> events = readJournal.byTagsWithMeta(tags)
  .mapMaterializedValue(meta -> {
    System.out.println("The query is: " +
      "ordered deterministically: " + meta.deterministicOrder + " " +
      "infinite: " + meta.infinite);
    return meta;
  });

events.map(event -> {
  System.out.println("Event payload: " + event.payload); 
  return event.payload;
}).runWith(Sink.ignore(), mat);

final ReactiveStreamsCompatibleDBDriver driver = new ReactiveStreamsCompatibleDBDriver();
final Subscriber<List<Object>> dbBatchWriter = driver.batchWriter();

// Using an example (Reactive Streams) Database driver
readJournal
  .eventsByPersistenceId("user-1337", 0L, Long.MAX_VALUE)
  .map(envelope -> envelope.event())
  .grouped(20) // batch inserts into groups of 20
  .runWith(Sink.fromSubscriber(dbBatchWriter), mat); // write batches to read-side database

class ExampleStore {
  CompletionStage<Void> save(Object any) {
    // ...
  }
}

final ExampleStore store = new ExampleStore();

readJournal
  .eventsByTag("bid", new Sequence(0L))
  .mapAsync(1, store::save)
  .runWith(Sink.ignore(), mat);

  final Timeout timeout = Timeout.apply(3, TimeUnit.SECONDS);

  final MyResumableProjection bidProjection = new MyResumableProjection("bid");

  final Props writerProps = Props.create(TheOneWhoWritesToQueryJournal.class, "bid");
  final ActorRef writer = system.actorOf(writerProps, "bid-projection-writer");

  long startFromOffset = bidProjection.latestOffset().toCompletableFuture().get(3, TimeUnit.SECONDS);

  readJournal
    .eventsByTag("bid", new Sequence(startFromOffset))
    .mapAsync(8, envelope -> {
      final CompletionStage<Object> f = ask(writer, envelope.event(), timeout);
      return f.thenApplyAsync(in -> envelope.offset(), system.dispatcher());
    })
    .mapAsync(1, offset -> bidProjection.saveProgress(offset))
    .runWith(Sink.ignore(), mat);
}

final class TheOneWhoWritesToQueryJournal extends AbstractActor {
  private final ExampleStore store;

  private ComplexState state = new ComplexState();

  public TheOneWhoWritesToQueryJournal() {
    store = new ExampleStore();

    receive(ReceiveBuilder.matchAny(message -> {
      state = updateState(state, message);

      // example saving logic that requires state to become ready:
      if (state.readyToSave())
        store.save(Record.of(state));
      
    }).build());
  }


  ComplexState updateState(ComplexState state, Object msg) {
    // some complicated aggregation logic here ...
    return state;
  }
}

public class MyReadJournalProvider implements ReadJournalProvider {
  private final MyJavadslReadJournal javadslReadJournal;

  public MyReadJournalProvider(ExtendedActorSystem system, Config config) {
    this.javadslReadJournal = new MyJavadslReadJournal(system, config);
  }
  
  @Override
  public MyScaladslReadJournal scaladslReadJournal() {
    return new MyScaladslReadJournal(javadslReadJournal);
  }

  @Override
  public MyJavadslReadJournal javadslReadJournal() {
    return this.javadslReadJournal;
  }  
}
public class MyJavadslReadJournal implements 
akka.persistence.query.javadsl.ReadJournal,
akka.persistence.query.javadsl.EventsByTagQuery2,
akka.persistence.query.javadsl.EventsByPersistenceIdQuery,
akka.persistence.query.javadsl.AllPersistenceIdsQuery,
akka.persistence.query.javadsl.CurrentPersistenceIdsQuery {

  private final FiniteDuration refreshInterval;

  public MyJavadslReadJournal(ExtendedActorSystem system, Config config) {
    refreshInterval = 
        FiniteDuration.create(config.getDuration("refresh-interval", 
            TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
  }

  @Override
  public Source<EventEnvelope2, NotUsed> eventsByTag(String tag, Offset offset) {
    if(offset instanceof Sequence){
      Sequence sequenceOffset = (Sequence) offset;
      final Props props = MyEventsByTagPublisher.props(tag, sequenceOffset.value(), refreshInterval);
      return Source.<EventEnvelope2>actorPublisher(props).
        mapMaterializedValue(m -> NotUsed.getInstance());
    }
    else
      throw new IllegalArgumentException("MyJavadslReadJournal does not support " + offset.getClass().getName() + " offsets");
  }

  @Override
  public Source<EventEnvelope, NotUsed> eventsByPersistenceId(String persistenceId,
      long fromSequenceNr, long toSequenceNr) {
    // implement in a similar way as eventsByTag
    throw new UnsupportedOperationException("Not implemented yet");
  }

  @Override
  public Source<String, NotUsed> allPersistenceIds() {
    // implement in a similar way as eventsByTag
    throw new UnsupportedOperationException("Not implemented yet");
  }
  
  @Override
  public Source<String, NotUsed> currentPersistenceIds() {
    // implement in a similar way as eventsByTag
    throw new UnsupportedOperationException("Not implemented yet");
  }
  
  // possibility to add more plugin specific queries

  public Source<RichEvent, QueryMetadata> byTagsWithMeta(Set<String> tags) {
    // implement in a similar way as eventsByTag
    throw new UnsupportedOperationException("Not implemented yet");
  }

}
public class MyScaladslReadJournal implements 
akka.persistence.query.scaladsl.ReadJournal,
akka.persistence.query.scaladsl.EventsByTagQuery2,
akka.persistence.query.scaladsl.EventsByPersistenceIdQuery,
akka.persistence.query.scaladsl.AllPersistenceIdsQuery,
akka.persistence.query.scaladsl.CurrentPersistenceIdsQuery {
  
  private final MyJavadslReadJournal javadslReadJournal;

  public MyScaladslReadJournal(MyJavadslReadJournal javadslReadJournal) {
    this.javadslReadJournal = javadslReadJournal;
  }

  @Override
  public akka.stream.scaladsl.Source<EventEnvelope2, NotUsed> eventsByTag(
      String tag, akka.persistence.query.Offset offset) {
    return javadslReadJournal.eventsByTag(tag, offset).asScala();
  }

  @Override
  public akka.stream.scaladsl.Source<EventEnvelope, NotUsed> eventsByPersistenceId(
      String persistenceId, long fromSequenceNr, long toSequenceNr) {
    return javadslReadJournal.eventsByPersistenceId(persistenceId, fromSequenceNr, 
        toSequenceNr).asScala();
  }

  @Override
  public akka.stream.scaladsl.Source<String, NotUsed> allPersistenceIds() {
    return javadslReadJournal.allPersistenceIds().asScala();
  }
  
  @Override
  public akka.stream.scaladsl.Source<String, NotUsed> currentPersistenceIds() {
    return javadslReadJournal.currentPersistenceIds().asScala();
  }
  
  // possibility to add more plugin specific queries

  public akka.stream.scaladsl.Source<RichEvent, QueryMetadata> byTagsWithMeta(
      scala.collection.Set<String> tags) {
    Set<String> jTags = scala.collection.JavaConversions.setAsJavaSet(tags);
    return javadslReadJournal.byTagsWithMeta(jTags).asScala();
  }

}

class MyEventsByTagJavaPublisher extends AbstractActorPublisher<EventEnvelope> {
  private final Serialization serialization =
    SerializationExtension.get(context().system());

  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 = context().system().scheduler();
    this.continueTask = scheduler
      .schedule(refreshInterval, refreshInterval, self(), CONTINUE,
                context().dispatcher(), self());

    receive(ReceiveBuilder
              .matchEquals(CONTINUE, (in) -> {
                query();
                deliverBuf();
              })
      .match(Cancel.class, (in) -> {
        context().stop(self());
      })
      .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(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));
  }
}