How to implement PersistenceQuery of ReadJournalFor in Akka.Net Hosting Model

user_374189 0
Viewed 65

I've worked through the documentation on Akka.Net PersistenceQuery here, but I'm struggling to figure out how I would hook up any of those queries inside an ASP.Net6 Blazor Server startup pipeline using the new Akka.Net Hosting model.

What I have in mind is to Sink such a query out to a SignalR hub that will cause views to refresh their data based on the output of a ReadForJournal stream.

Has anyone done this, and if so, please can you provide me with some guidance in this regard?

akka.net
1 Answers

I have not done this before, much less an expert are this, but I can try to point you in the right direction! :)

If you want to run a local actor, you can spawn the ProjectionBehavior as any other Behavior. This can be useful for testing or when running a local ActorSystem without Akka Cluster.

SourceProvider<Offset, EventEnvelope<ShoppingCart.Event>> sourceProvider(String tag) {
  return EventSourcedProvider.eventsByTag(system, CassandraReadJournal.Identifier(), tag);
}

Projection<EventEnvelope<ShoppingCart.Event>> projection(String tag) {
  return CassandraProjection.atLeastOnce(
      ProjectionId.of("shopping-carts", tag), sourceProvider(tag), ShoppingCartHandler::new);
}

Projection<EventEnvelope<ShoppingCart.Event>> projection1 = projection("carts-1");

ActorRef<ProjectionBehavior.Command> projection1Ref =
    context.spawn(ProjectionBehavior.create(projection1), projection1.projectionId().id());

You can combine this with your predefined query, e.g.:

var queries = PersistenceQuery.Get(actorSystem)
    .ReadJournalFor<SqlReadJournal>("akka.persistence.query.my-read-journal");

var mat = ActorMaterializer.Create(actorSystem);
Source<string, NotUsed> src = queries.AllPersistenceIds();

So I was thinking maybe your queries could be linked to your ProjectionBehavior for the akka hosting model to host it.

Related sources:

user_176985320
Related