Publish-Subscribe / PostgreSQL

Pub/Subについては下記で扱ったことがありますが、ここではPostgreSQLの標準機能にあるPub(Notify)/Sub(Listen)を試してみました。

MQTT

Robot Operating System

下記環境を使って、psqlと外部プログラム(Python)で動作確認をしました。
https://decode.red/ed/archives/1465

環境） PostgreSQL 16 (Postgres.app v2.6.7), PyCharm(Python3) / Mac(arm64)
インストール）

pip install psycopg2

Error: pg_config executable not found.
pgコマンド関連のパスが通っていないため上記エラーとなりした。

PATH=/Applications/Postgres.app/Contents/Versions/16/bin:$PATH

パス追加で解決。
(DBへの接続はすべて同じですが、いろいろな表記を試しました。)

pg_pub.py

import sys
import psycopg2
import psycopg2.extensions

#DSN = "dbname=postgres"
#conn = psycopg2.connect(DSN)
conn = psycopg2.connect('postgresql://postgres:postgres@localhost')
conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
curs = conn.cursor()

channel  = sys.argv[1]
nickname = sys.argv[2]
print("PUBLISH to channel : %s" % channel)

while True:
    try:
        message = input()
        chat_message = psycopg2.extensions.QuotedString(
                "[%s]:%s" % (nickname, message))
        print(chat_message)
        curs.execute("NOTIFY %s, %s;" % (channel, chat_message))
    except EOFError:
        break

import sys

import psycopg2

import psycopg2.extensions

#DSN = "dbname=postgres"

#conn = psycopg2.connect(DSN)

conn = psycopg2.connect('postgresql://postgres:postgres@localhost')

conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

curs = conn.cursor()

channel = sys.argv[1]

nickname = sys.argv[2]

print("PUBLISH to channel : %s" % channel)

while True:

try:

message = input()

chat_message = psycopg2.extensions.QuotedString(

"[%s]:%s" % (nickname, message))

print(chat_message)

curs.execute("NOTIFY %s, %s;" % (channel, chat_message))

except EOFError:

break

pg_sub.py

import select
import sys

import psycopg2
import psycopg2.extensions

DSN = "dbname=postgres user=postgres"
conn = psycopg2.connect(DSN)
conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

curs = conn.cursor()

for channel in sys.argv[1:]:
    print ("SUBSCRIBE TO channel : %s" % channel)
    curs.execute("LISTEN %s;" % channel)

while True:
    try:
        conn.poll()
        while conn.notifies:
            notify = conn.notifies.pop()
            print ("%s - %s" % (notify.channel, notify.payload))
    except BaseException as err:
        print(err)
        break

for channel in sys.argv[1:]:
    print ("UNSUBSCRIBE FROM channel : %s" % channel)
    curs.execute("UNLISTEN %s;" % channel)

import select

import sys

import psycopg2

import psycopg2.extensions

DSN = "dbname=postgres user=postgres"

conn = psycopg2.connect(DSN)

conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

curs = conn.cursor()

for channel in sys.argv[1:]:

print ("SUBSCRIBE TO channel : %s" % channel)

curs.execute("LISTEN %s;" % channel)

while True:

try:

conn.poll()

while conn.notifies:

notify = conn.notifies.pop()

print ("%s - %s" % (notify.channel, notify.payload))

except BaseException as err:

print(err)

break

for channel in sys.argv[1:]:

print ("UNSUBSCRIBE FROM channel : %s" % channel)

curs.execute("UNLISTEN %s;" % channel)

まず外部プログラムでPublishしたものを、psqlで受け取ります。LISTENコマンドを逐次実行して確かめます。
LISTENしているチャンネルしか受け取りません。外部プログラムは改行により、文字列を何度も送れるしくみになっています。
ユーザ名とメッセージはこのアプリの仕様で、結合して送られます。CTL-Cで終了。
次に、外部プログラムを受信まち状態にして、psqlからNOTIFYするごとに送信します。ここでもLISTEN(SUBSCRIBE)しているチャンネルしか受信できないことを確認できます。外部プログラムは起動でSUBSCRIBE、停止でUNSUBSCRIBEし、メッセージを受信するたびに表示します。

下記を参考にさせていただきました。コードはPython3による変更をしています。

参考）　https://siguniang.wordpress.com/2012/12/09/postgresql-pub-sub/