使用 Slick 监听 PostgreSQL NOTIFY 事件答案

【问题标题】：Listen to PostgreSQL NOTIFY events with Slick使用 Slick 监听 PostgreSQL NOTIFY 事件
【发布时间】：2018-04-18 08:44:05
【问题描述】：

我可以使用 Slick / Play Framework (Scala) 来收听 PostgreSQL NOTIFY 语句吗？

我想做类似的事情：

http://bjorngylling.com/2011-04-13/postgres-listen-notify-with-node-js.html

【问题讨论】：

这有帮助吗？ smartjava.org/content/listen-notifications-postgresql-scala

标签： postgresql scala playframework slick slick-3.0

【解决方案1】：

我认为 Slick 不支持 PostgreSQL's NOTIFY，但 postgresql-async 库支持。可以使用后者创建一个 Akka Streams Source 并将其合并到一个 Play 端点中，该端点将数据库通知流式传输到客户端：

package controllers

import javax.inject.{Inject, Singleton}

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._

import com.github.mauricio.async.db.postgresql.PostgreSQLConnection
import com.github.mauricio.async.db.postgresql.util.URLParser
import com.github.mauricio.async.db.util.ExecutorServiceUtils.CachedExecutionContext

import play.api.Logger
import play.api.http.ContentTypes
import play.api.libs.EventSource
import play.api.mvc._

import scala.concurrent.duration._
import scala.concurrent.Await

@Singleton
class DbNotificationController @Inject()(cc: ControllerComponents,
                                         materializer: Materializer)
  extends AbstractController(cc) {

  implicit val mat = materializer

  val configuration = URLParser.parse("jdbc:postgresql://localhost:5233/my_db?user=dbuser&password=pwd")
  val connection = new PostgreSQLConnection(configuration)
  Await.result(connection.connect, 5 seconds)

  val (actor, dbSource) =
    Source.actorRef[String](Int.MaxValue, OverflowStrategy.dropNew)
          .toMat(BroadcastHub.sink[String])(Keep.both)
          .run()

  connection.sendQuery("LISTEN my_channel")
  connection.registerNotifyListener { message =>
    val msg = message.payload
    Logger.debug(s"Sending the payload: $msg")
    actor ! msg
  }

  def index() = Action {
    Ok(views.html.scaladbnotification())
  }

  def streamDb() = Action {
    Ok.chunked(dbSource via EventSource.flow).as(ContentTypes.EVENT_STREAM)
  }
}

在上面的控制器中，当监听器收到来自数据库的通知时，通知中的有效负载（String）会被记录并发送给参与者。发送到此参与者的消息提供Source 用于streamDb 端点。在负载消息发送到客户端之前，它们会被转换为 Play 的 EventSource 类。

我从Play streaming example application 改编了DbNotificationController，你可以用它来做实验。如果您想这样做，显然您需要将DbNotificationController 集成到该项目中：

将"com.github.mauricio" %% "postgresql-async" % "0.2.21" 添加到build.sbt。
根据需要设置 PostgreSQL，包括NOTIFY，并根据您的配置调整控制器中的数据库 URL。
将DbNotificationController 复制并粘贴到/app/controllers/。
将以下文件（称为scaladbnotification.scala.html）复制到app/views/：

@main {

    <h1>Server Sent Event from DB</h1>

    <h1 id="db"></h1>

    <p>
        DB events are pushed from the Server using a Server Sent Event connection.
    </p>

    <script type="text/javascript" charset="utf-8">
        if (!!window.EventSource) {
            var stringSource = new EventSource("@routes.DbNotificationController.streamDb()");
            stringSource.addEventListener('message', function(e) {
                $('#db').html(e.data.replace(/(\d)/g, '<span>$1</span>'))
            });
        } else {
            $("#db").html("Sorry. This browser doesn't seem to support Server sent event. Check <a href='http://html5test.com/compare/feature/communication-eventSource.html'>html5test</a> for browser compatibility."); 
        }
    </script>    
}

在/conf/routes 文件中，添加以下行：

GET /scala/dbNotification controllers.DbNotificationController.index() 获取 /scala/dbNotification/liveDb 控制器.DbNotificationController.streamDb()

使用sbt run 启动应用程序并在浏览器中导航到以下 URL：

http://localhost:9000/scala/dbNotification

【讨论】：