Scala Slick: Stream Never Ends

Question

With Slick, you can do the following to create a stream of results from a table:

val q = for (e <- events) yield e.name
val p: DatabasePublisher[String] = db.stream(q.result)

p.foreach { s => println(s"Event: $s") }

This will print all events in the table eventsand end after the last row.

Assuming you can be notified in some way about when new rows are entered into the table events, is it possible to write a stream that will continuously output events as they are inserted? Type tail -ffor the database table.

I think Slick will not support this natively, but I think you need to use Akka's streaming usage to help. Therefore, if you could have something that was taken from Slick Source while it was not empty, you would expect the event to indicate more data in the table, and then flush new data. Perhaps using ActorPublisherto link this logic?

Just wondering if anyone has any experience in this area or any tips?

Answer 1

You were right about it ActorPublisher:) Here is a simple example of using PostgreSQL, async DB driver and LISTEN / NOTIFY mechanism :

Actor:

class PostgresListener extends ActorPublisher[String] {

  override def receive = {
    case _ ⇒
      val configuration = URLParser.parse(s"jdbc://postgresql://$host:$port/$db?user=$user&password=$password")
      val connection = new PostgreSQLConnection(configuration)
      Await.result(connection.connect, 5.seconds)

      connection.sendQuery(s"LISTEN $channel")
      connection.registerNotifyListener { message ⇒ onNext(message.payload) }
  }
}

Services:

def stream: Source[ServerSentEvent, Unit] = {
  val dataPublisherRef = Props[PostgresListener]
  val dataPublisher = ActorPublisher[String](dataPublisherRef)

  dataPublisherRef ! "go"

  Source(dataPublisher)
    .map(ServerSentEvent(_))
    .via(WithHeartbeats(10.second))
}

build.sbt libraryDependencies:

"com.github.mauricio"  %% "postgresql-async"         % "0.2.18"

Postgres select pg_notify('foo', 'payload')

, Slick LISTEN.