Asyncpg: Don't work remove_listener(channel, callback)

[Belanchuk]

• asyncpg version: 0.13.0
• PostgreSQL version: PostgreSQL 9.5.10 on x86_64-pc-linux-gnu
• Python version: Python 3.6.3 :: Anaconda, Inc.
• Platform: Ubuntu 5.4.0-6ubuntu1~16.04.4
• Do you use pgbouncer?: no
• Did you install asyncpg with pip?: yes (Anaconda)
• Can the issue be reproduced under both asyncio and
  uvloop?: yes

Доброго дня!
Хочу управлять оповещениями Postgresql NOTIFY через websocket.
В качестве сервера использую aiohttp.
Успешно подключаюсь с вебстранички к серверу и отправляю по websocket'у {'subscribe':'channel_name'}
Подписка работает и уведомления соответствующего канала отправляются на веб-страничку.
Когда я хочу отписаться, я отправляю с веб-странички по websocket'у {'unsubscribe':'channel_name'}, но срабатывает только print('unsub') и подписка продолжает работать.

Буду очень признателен за помощь.

Код сервера:

import asyncio
# import uvloop
# asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
import asyncpg
import ujson
import aiohttp
from aiohttp import web
import routes


async def subscribe(data, ws, conn):

    def callback_listener(connection, pid, channel, payload):
        print(payload)
        ws.send_str(payload)

    if 'subscribe' in data.keys():
        print('sub')
        await conn.add_listener(data['subscribe'], callback_listener)                    
    if 'unsubscribe' in data.keys():
        print('unsub')
        await conn.remove_listener(data['unsubscribe'], callback_listener) 


async def websocket_handler(request):

    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.type == aiohttp.WSMsgType.TEXT:
            if msg.data == 'close':
                await ws.close()
            else:
                data = ujson.loads(msg.data)
                await subscribe(data, ws, request.app['db'])

        elif msg.type == aiohttp.WSMsgType.ERROR:
            print('ws connection closed with exception %s' %
                  ws.exception())

    print('websocket connection closed')
    return ws


async def init_pg(app):
    conn = await asyncpg.connect(host='localhost', database='db', user='user', password='pass')    
    app['db'] = conn
    print('init_pg')


async def close_pg(app):
    await app['db'].close()
    print('close_pg')


app = web.Application()
app.router.add_get('/ws', handler=websocket_handler, name='ws')
app.on_startup.append(init_pg)
app.on_cleanup.append(close_pg)
web.run_app(app, host='127.0.0.1', port=8080)

[elprans]

Make def callback_listener global. You are creating a new callback_listener object on every call to subscribe, so the object you pass to add_listener is not the same as the one you are passing to remove_listener.

@1st1, is there a reason why we silently ignore invalid channel/callback in remove_listener?

[Belanchuk]

Спасибо!
p.s. Насколько для Вас удобно общение на русском языке?

[vitaly-burovoy]

@elprans
I guess it is because the final point of the remove_listener call should be absence of the callback.
If it does not present there, OK, nothing to do.

The same way as UNLISTEN does not complain about the channel argument which was not LISTENed before.

[vitaly-burovoy]

@Belanchuk
I'm afraid communications in non-English language are inconvenient for people in the community and for people who find this page via search engines.

One of rules of good taste is discussing in language that can be understood by as many people as it possible even if you make mistakes in each sentence.

P.S.: if you get a response from owners of the repo in a language different from one in your initial question then it is a hint for you and the answer to your second question.

[elprans]

@vitaly-burovoy

The same way as UNLISTEN does not complain about the channel argument which was not LISTENed before.

I get that, but I don't think Postgres is doing the right thing either, especially given how it customarily raises Notices for much smaller offences. On the other hand it's very easy to make a mistake like @Belanchuk did. Perhaps, raising a UserWarning wouldn't be a bad idea.

[Belanchuk]

@vitaly-burovoy
Yes, sure.

[1st1]

@Belanchuk Мы свободно изъясняемся по-русски, но как отметил Виталий, хотим строго придерживаться общения на английском на гитхабе.

[1st1]

Perhaps, raising a UserWarning wouldn't be a bad idea.

Or we can use loop.call_exception_handler() API. This won't stop the execution of a program, but will be more noticeable than a warning.

[Belanchuk]

@1st1
I was sure of that. )
But I'm a bit annoyed to write in English.
All your arguments are accepted.