Javascript Client for Frontend

[Koliham]

I was able to extract the payloads sent between client and server to create a Javascript client with websocket (code below).

But there are 2 issues, which I haven't solved yet:

1. While working fine on Windows and Android Browsers, there are issues on Safari: The connection to the Websocket is established, but the browser doesn't receive the messages, when triggering pub.
2. The connection pool doesn't seem to be cleaned up, when a client disconnects. I get these error messages, because a subscriber was not found:

Click to see stack trace!

Failed to notify subscriber sub_id=3d0477159cdc4eb0bd8afefa9df82423 with topic=topicA
Traceback (most recent call last):
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/fastapi_websocket_pubsub/event_notifier.py", line 220, in callback_subscribers
  await self.trigger_callback(data, topic, subscriber_id, event)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/fastapi_websocket_pubsub/event_notifier.py", line 178, in trigger_callback
  await subscription.callback(subscription, data)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/fastapi_websocket_pubsub/event_broadcaster.py", line 159, in __broadcast_notifications__
  async with self._sharing_broadcast_channel:
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/broadcaster/_base.py", line 48, in __aenter__
  await self.connect()
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/broadcaster/_base.py", line 55, in connect
  await self._backend.connect()
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/broadcaster/_backends/postgres.py", line 13, in connect
  self._conn = await asyncpg.connect(self._url)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connection.py", line 2093, in connect
  return await connect_utils._connect(
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 903, in _connect
  raise last_error
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 889, in _connect
  return await _connect_addr(
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 781, in _connect_addr
  return await __connect_addr(params, timeout, True, *args)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 833, in __connect_addr
  tr, pr = await compat.wait_for(connector, timeout=timeout)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/compat.py", line 66, in wait_for
  return await asyncio.wait_for(fut, timeout)
File "/home/linuxbrew/.linuxbrew/opt/python@3.9/lib/python3.9/asyncio/tasks.py", line 481, in wait_for
  return fut.result()
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 700, in _create_ssl_connection
  do_ssl_upgrade = await pr.on_data
ConnectionError: unexpected connection_lost() call
Failed to notify subscriber sub_id=3d0477159cdc4eb0bd8afefa9df82423 with topic=topicC
Traceback (most recent call last):
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/fastapi_websocket_pubsub/event_notifier.py", line 220, in callback_subscribers
  await self.trigger_callback(data, topic, subscriber_id, event)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/fastapi_websocket_pubsub/event_notifier.py", line 178, in trigger_callback
  await subscription.callback(subscription, data)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/fastapi_websocket_pubsub/event_broadcaster.py", line 159, in __broadcast_notifications__
  async with self._sharing_broadcast_channel:
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/broadcaster/_base.py", line 48, in __aenter__
  await self.connect()
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/broadcaster/_base.py", line 55, in connect
  await self._backend.connect()
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/broadcaster/_backends/postgres.py", line 13, in connect
  self._conn = await asyncpg.connect(self._url)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connection.py", line 2093, in connect
  return await connect_utils._connect(
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 903, in _connect
  raise last_error
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 889, in _connect
  return await _connect_addr(
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 781, in _connect_addr
  return await __connect_addr(params, timeout, True, *args)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 833, in __connect_addr
  tr, pr = await compat.wait_for(connector, timeout=timeout)
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/compat.py", line 66, in wait_for
  return await asyncio.wait_for(fut, timeout)
File "/home/linuxbrew/.linuxbrew/opt/python@3.9/lib/python3.9/asyncio/tasks.py", line 481, in wait_for
  return fut.result()
File "/home/koliham/.cache/pypoetry/virtualenvs/websocketsample-dFaV7d37-py3.9/lib/python3.9/site-packages/asyncpg/connect_utils.py", line 700, in _create_ssl_connection
  do_ssl_upgrade = await pr.on_data
ConnectionError: unexpected connection_lost() call

Frontend Code

<template>
<div class="text-center self-center">
     <div>Status is: {{status}} <q-btn color="primary" @click="checkStatus()">Check Status</q-btn></div>
    <div v-for="response in responses" :key="response">{{ response }}</div>
  </div>
</template>

<script>
export default {
name: 'PageIndex',

data() {
  return {
    inhalt: "",
    responses: [],
    status: -1
  }
},
created() {
  var payload = {request: {method: "subscribe", "arguments": {"topics":["topic1", "topic2", "topic3"]}}}
  var payload_str = JSON.stringify(payload)
  this.connection = new WebSocket("ws://localhost:8000/pubsub")

  this.connection.onmessage = (event) => {
    console.log(event)
    this.messageReceived(event.data)
  }

  this.connection.onopen = (event) => {
    this.connection.send(payload_str)
  }

},

methods: {
  checkStatus() {
    this.status = this.connection.readyState
  },
  messageReceived (content) {
  //called, when message is received
    this.responses.push(content)
  },
}
}
</script>

Backend Code

import asyncio

import uvicorn
from fastapi import FastAPI
from fastapi.routing import APIRouter
from fastapi_websocket_pubsub import PubSubEndpoint
from starlette.websockets import WebSocket

PORT = 8000


app = FastAPI()
router = APIRouter()
endpoint = PubSubEndpoint(broadcaster="postgres://postgres:pw@localhost:5432/")

@router.websocket("/pubsub")
async def websocket_rpc_endpoint(websocket: WebSocket):
  async with endpoint.broadcaster:
      await endpoint.main_loop(websocket)

app.include_router(router)


async def events(topic: str = "NoTopic"):
  await endpoint.publish([topic], {"somekey": "SomeValue"})


@app.get("/trigger")
async def trigger_events():
  asyncio.create_task(events("topic1"))
  asyncio.create_task(events("topic3"))


if __name__ == "__main__":
  uvicorn.run(app, host="0.0.0.0", port=PORT)

[orweis]

Hi @Koliham :)
Kudos, on working on a lightweight JS version :)

That being said, while you got the basic schema correct; you have left out the rest of the RPC protocol, which would be my prime suspect for the issues you're experiencing. Primarily not receiving the responses sent back for your RPC (subscribe) call