How to query multiple tags

Akka
1

Hi all

I am using EventSourcedBehavior to persist incoming messages.
The definition of EventSourcedBehavior actor.

def create(maintainer: ActorRef[MessageStore], logger: ActorRef[Log]): EventSourcedBehavior[Command, Event, State] =
EventSourcedBehavior[Command, Event, State](
    persistenceId = PersistenceId("connector-store"),
    emptyState = State(Nil),
    commandHandler = commandHandler(maintainer),
    eventHandler = eventHandler)
.onPersistFailure(SupervisorStrategy.restartWithBackoff(minBackoff = 10.seconds, maxBackoff = 60.seconds, randomFactor = 0.1))
.withTagger(state => Set("messages", state.msg.origin))
.receiveSignal {
    case (_, Terminated) =>
    println("Message store actor was terminated")
    logger ! Log(Error, "Message store actor was terminated")
}

As you can see, I’ve tagged the messages with two tags.
With the https://doc.akka.io/docs/akka/current/persistence-query.html, I would like to read
the messages from store.

I could read it as follows:

val blueThings: Source[EventEnvelope, NotUsed] =
 readJournal.eventsByTag("blue", Offset.noOffset)

The question is, how to pass the second tag to the method eventsByTag.
That means, I would like to read messages from the store, that are tagged with for example messages and blue.

Thanks

2

Hey,

I am afraid that’s not possible. The Persistence Query tries to stay away from implementation details, as much as possible, and thus it cannot make any assumptions about capabilities of the underlying technology. There are cases when the engine won’t be capable of fetching events using multiple tags.

Having said that, if the database engine you’re using is capable, you can always extend the interface to add such functionality. You might want to take a look at one of the existing plugins for inspiration.

Hope this is helpful.

1 Like
3

Thanks so much for answer. I am using https://doc.akka.io/docs/akka-persistence-cassandra/current/overview.html. Does it support multi tags query?

4

Looking at the implementation of the CassandraReadJournal, it does not seem like it supports such query. The journal plugins I’ve used so far tend to support only the “official” interface. That said, the Materialized values of queries section in the Persistence Query documentation explicitly notes that

Journals are able to provide additional information related to a query by exposing Materialized values, which are a feature of Streams that allows to expose additional values at stream materialization time.

More advanced query journals may use this technique to expose information about the character of the materialized stream, for example if it’s finite or infinite, strictly ordered or not ordered at all.

Funnily enough, the pseudo example given is that of readJournal.byTagsWithMeta(Set("red", "blue")). I, personally, have never needed such functionality (or have managed to convince myself that I can implement the functionality with other means), so I haven’t looked more thoroughly for more powerful plugin implementations. If you do happen to discover one, I would be most appreciative if you update the thread!