diff --git a/modules/core/shared/src/main/scala/Channel.scala b/modules/core/shared/src/main/scala/Channel.scala index 2c10c07b..a6ba53c9 100644 --- a/modules/core/shared/src/main/scala/Channel.scala +++ b/modules/core/shared/src/main/scala/Channel.scala @@ -10,9 +10,10 @@ import cats.effect.{ Resource, MonadCancel } import cats.effect.kernel.MonadCancelThrow import cats.syntax.all._ import fs2.{ Pipe, Stream } +import skunk.codec.all._ import skunk.data.{ Identifier, Notification } import skunk.net.Protocol -import skunk.util.Origin +import skunk.util.{ Origin, Typer } /** * A '''channel''' that can be used for inter-process communication, implemented in terms of @@ -140,9 +141,14 @@ object Channel { } yield stream.filter(_.channel === name) + val notifyQuery = + Query("SELECT pg_notify($1, $2) IS NULL", Origin.unknown, text ~ text, bool) + def notify(message: String): F[Unit] = - // TODO: escape the message - proto.execute(Command(s"NOTIFY ${name.sql}, '$message'", Origin.unknown, Void.codec)).void + for { + pq <- proto.prepare(notifyQuery, Typer.Static) + _ <- pq.executeSized((name.value, message), Origin.unknown, 1).void + } yield () } diff --git a/modules/tests/shared/src/test/scala/ChannelTest.scala b/modules/tests/shared/src/test/scala/ChannelTest.scala index a3a183a7..c9edde0e 100644 --- a/modules/tests/shared/src/test/scala/ChannelTest.scala +++ b/modules/tests/shared/src/test/scala/ChannelTest.scala @@ -56,6 +56,19 @@ class ChannelTest extends SkunkTest { } } + sessionTest("channel notify payload is passed as a bound parameter") { s => + val payload = "it's a 'test\"message" + val ch = s.channel(ident"channel_test") + + ch.listenR(42).use { r => + for { + _ <- ch.notify(payload) + d <- r.map(_.value).take(1).compile.toList + _ <- assert(s"channel payload $d", d == List(payload)) + } yield "ok" + } + } + sessionTest("channel with quoted identifier round-trips through LISTEN/NOTIFY/UNLISTEN") { s => val data = List("foo", "bar", "baz") val ch = s.channel(ident"q_my_queue.INSERT")