scala - Implementing a "live" stream to drive an Akka 2.4 Persistence Query -

i have been investigating experimental akka persistence query module , interested in implementing custom read journal application. documentation describes 2 main flavors of queries, ones return current state of journal (e.g currentpersistenceidsquery) , ones return subscribe-able stream emit events events committed journal via write side of application (e.g. allpersistenceidsquery)

for contrived application, using postgres , slick 3.1.1 drive guts of these queries. can stream database query results doing like:

override def allpersistenceids = {   val db = database.forconfig("postgres")   val metadata = tablequery[metadata]    val query = (m <- metadata) yield m.persistenceid   source.frompublisher(db.stream(query.result)) } 

however, stream signaled complete underlying slick db action completed. doesn't seem fulfill requirement of perpetually open stream capable of emitting new events.

my questions are:

• is there way purely using akka streams dsl? is, can sent flow cannot closed?
• i have done exploring on how leveldb read journal works , seems handle new events having read journal subscribe write journal. seems reasonable must ask - in general, there recommended approach dealing requirement?
• the other approach have thought polling (e.g. periodically have read journal query db , check new events / ids). more experience able offer advice?

thanks!

it's not trivial 1 line of code you're 1 right track already.

in order implement "infinite" stream you'll need query multiple times - i.e. implement polling, unless underlying db allows infinite query (which here not afaics).

the polling needs keep track of "offset", if you're querying tag, , issue poll, need start (2nd now) query "last emitted element", , not beginning of table again. need somewhere, actor, keeps offset.

the query side leveldb plugin not best role model other implementations assumes underlying journal , how work. also, leveldb not meant production akka persistence – it's journal ship in order have persistent journal can play around out of box (without starting cassandra etc).

if you're looking inspiration mongodb plugins should pretty source that, have similar limitations sql stores. i'm not sure if of sql journals did implement query side.