Setting websocket port

[bobbilichandu]

First check

• I added a very descriptive title to this issue.
• [*] I used the GitHub search to find a similar issue and didn't find it.
• [*] I searched the FastAPI documentation, with the integrated search.
• [*] I already searched in Google "How to X in FastAPI" and didn't find any information.
• [*] I already read and followed all the tutorial in the docs and didn't find an answer.
• [*] I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.
• [*] After submitting this, I commit to one of:
  • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
  • I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
  • Implement a Pull Request for a confirmed bug

Port

How to add port in a websocket. I want specifically mention a port in the websocket server. I can specity the port from JS while connecting to a websocket, but can't I do it from the server side. Is there any default value? Can I change that default value?

@app.websocket("/ws")
async def websocketEndpoint(websocket: WebSocket):
    print("Awaiting clients connection")
    await websocket.accept()
    while True:
        try:
            await websocket.receive_text()
            resp = {'value': random.uniform(0,1)}
            await websocket.send_json(resp)
        except Exception as e:
            print('error:', e)
            break
    print("connection ended")

Environment

• OS: [e.g. Linux / Windows / macOS]: linux 18.04
• FastAPI Version [e.g. 0.3.0]: 0.65.1

To know the FastAPI version use:

• Python version: 3.8

To know the Python version use:

[roveo]

I think websockets' endpoints run on the same port as your http server, so in JS you specify the same port that you specify when calling uvicorn.run (or whatever else you're using).

[bobbilichandu]

Got it. Is there a way for Server to know that websocket connection is broken or disconnected from client side?
Suppose, a web page is reloaded or closed, how would server know that client cut the connection?

[roveo]

There are WebSocketDisconnect and WebSocketClose exceptions in fastapi.websockets module. I think you need experiment to figure which one is raised when, but as far as I remember, they are raised when you're trying to do something, like send or await no receiving data. So, usually there's some kind of loop inside the ws endpoint, and you wrap the code in this loop in try: .. except: .., so that the endpoint function exits when the client is no longer available.

[bobbilichandu]

So I am facing this issue. I am not able to add multiple clients to same server with different client Ids

from time import sleep, time
from typing import Optional

from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from connectionManager import ConnectionManager

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
logfolder = "logfiles"
manager = ConnectionManager()


@app.get("/logdata/{filename}")
async def getLogUpdates(request: Request, filename: str):
    return templates.TemplateResponse("main.html", context={"request": request, "filename": filename})

@app.websocket("/ws/{client_id}/{filename}")
async def websocket_endpoint(websocket: WebSocket, filename: str, client_id: str):
    print("new client added to the file log " + str(filename) + ": " + str(client_id))
    await manager.connect(websocket)
    filepath = os.path.join(logfolder, str(filename) + ".log")
    f = open(filepath, "r")
    seek = 0
    while True:
        f.seek(seek)
        line = f.readline()
        seek = f.tell()
        if not line:
            break
    try:
        while True:
            f.seek(seek)
            line = f.readline()
            if line:
                await manager.broadcast(line.strip())
                seek = f.tell()
            else:
                sleep(0.5)
    except WebSocketDisconnect:
        print("client " + client_id + " disconnected from the file " + str(filename))
        manager.disconnect(websocket)

My connection manager:

from fastapi import WebSocket


class ConnectionManager:
    
    def __init__(self):
        self.activeConnections: List[WebSocket] = []
        
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.activeConnections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.activeConnections.remove(websocket)
        
    async def send_text(self, text: str, websocket: WebSocket):
        await websocket.send_text(text)
    
    async def broadcast(self, text: str):
        for connection in self.activeConnections:
            await connection.send_text(text)```

My frontend: [I am kinda new to frontend, I did what I can]
```<!DOCTYPE html>
<html>
<head>
	<meta charset="utf-8">
	<meta http-equiv="X-UA-Compatible" content="IE=edge">
	<title>Tail </title>
	<style>
		body {font-family: monospace;}
		p{ margin: 0;}
		ul{padding: 0;}
		li{list-style:none; padding: 0; margin: 0;}
	</style>
</head>
<body>
<ul id="logger">
	
</ul>
<script>
	function closeIt() {
		console.log("closing the socket");
		socket.close();
	}
	var client_id = Date.now()
	const socket = new WebSocket('ws://localhost:8000/ws/' + client_id + '/{{filename}}');
	var log = document.getElementById('logger')
	socket.addEventListener('message', function (event) {
		var text = document.createTextNode(event.data)
		var li = document.createElement("li")
		li.appendChild(text)
		log.prepend(li)
	});
	window.onbeforeunload = closeIt;
	window.onreload = closeIt;
</script>
</body>
</html>

I am not able to run a different client with same server. Not sure what I am doing wrong.

[roveo]

@chandu1263 the code you posted is not formatted properly. For multi-line code snippets you should use triple backticks (```) at the beginning and end.

[bobbilichandu]

@roveo I formatted it correctly. Can you help me with what I might be doing wrong?

[roveo]

Not sure. I would suggest you test parts of your code separately to make sure they work as intended (ConnectionManager class, ws endpoint and then frontend part).

That said, I see several issues with your solution:

• time.sleep function is not asynchronous and so it is blocking (the whole app), you should use await asyncio.sleep(0.5)

• file IO is also blocking. You should use aiofiles or run_in_executor to avoid this.

• I would factor out the file listening logic to another class, so you would do something like this in the endpoint:

listener = FileListener(file)
while True:
    new_line = await listener.getnext()
    await websocket.send(new_line)

or even

listener = FileListener(file)
async for line in listener:
    await websocket.send(line)

• I don't think that you need a separate ws endpoint for each client. If you eliminate the client_id part from the endpoint, you can also get rid of ConnectionManager class and broadcasting. Just open the log file for each connected client and send lines as they appear.

• there's this library: https://github.com/samuelcolvin/watchgod that allows you to asynchronously listen to changes in files. Might help in your case.

[bobbilichandu]

I have modified this accordingly on what you suggested. I added a new class fileListener

import time

class FileListener:
    
    def __init__(self, filepath):
        self.seek = 0
        self.f = open(filepath, "r")
        self.latest_line = ""
        self.get_latest_seek()
        self.last_updated_time = time.time()
        self.thread = threading.Thread(target=self.update, args=())
        self.thread.daemon = True
        self.thread.start()
        
    def get_latest_seek(self):
        while True:
            self.f.seek(self.seek)    
            self.latest_line = self.f.read()
            self.seek = self.f.tell()
            if not self.latest_line:
                break
            
    def update(self):
        while True:
            self.f.seek(self.seek)
            latest_line = self.f.read()
            if latest_line:
                if latest_line.strip():
                    self.latest_line = latest_line.strip()
                    self.last_updated_time = time.time()
                self.seek = self.f.tell() 
            
    def get_latest(self):
        return self.last_updated_time, self.latest_line

Then I removed the client_id and connection manager. I also removed the time.sleep()

import pickle
import os
from time import sleep, time
from typing import Optional

from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from fileListener import FileListener

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
logfolder = "logfiles"


@app.get("/logdata/{filename}")
async def getLogUpdates(request: Request, filename: str):
    return templates.TemplateResponse("main.html", context={"request": request, "filename": filename})

@app.websocket("/ws/{filename}")
async def websocket_endpoint(websocket: WebSocket, filename: str):
    print("new client added to the file log " + str(filename))
    await websocket.accept()
    fileListener = FileListener(os.path.join(logfolder, filename + ".log"))
    t = time()
    try:
        while True:
            latest_t, line = fileListener.get_latest()
            if latest_t != t:
                await websocket.send_text(line)
                t = latest_t
    except WebSocketDisconnect:
        print("client disconnected from the file " + str(filename))
        await websocket.close()

But still I am not able to add more than one client to the server. Second client is waiting for the first client to disconnect.

I modified the front end code [ removing the client id]

<!DOCTYPE html>
<html>
<head>
	<meta charset="utf-8">
	<meta http-equiv="X-UA-Compatible" content="IE=edge">
	<title>Tail </title>
	<style>
		body {font-family: monospace;}
		p{ margin: 0;}
		ul{padding: 0;}
		li{list-style:none; padding: 0; margin: 0;}
	</style>
</head>
<body>
<ul id="logger">
	
</ul>
<script>
	function closeIt() {
		console.log("closing the socket");
		socket.close();
	}
	const socket = new WebSocket('ws://localhost:8000/ws/{{filename}}');
	var log = document.getElementById('logger')
	socket.addEventListener('message', function (event) {
		var text = document.createTextNode(event.data)
		var li = document.createElement("li")
		li.appendChild(text)
		log.prepend(li)
	});
	window.onbeforeunload = closeIt;
	window.onreload = closeIt;
</script>
</body>
</html>

Am I missing some basic async and await functionality. I am still learning about them. So I am kinda confused.
Any help would be great!! Thanks!

[roveo]

Am I missing some basic async and await functionality

The way to think about async here is that the second client can connect only while the first endpoint is awaiting on something. Because your whole app works in the same thread, it can't accept a second connection because when client2 tries to connect, the thread is blocked at this line:

latest_t, line = fileListener.get_latest()

fileListener exists in a separate thread, BUT your main thread is blocked by waiting for the next line to appear. You can avoid this by making your file listener asynchronous.

[roveo]

Here's a working example with just one file per app and so one endpoint. Run this with python app.py yourlog.log, so your log file goes after app.py as an argument.

import asyncio
import sys

import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse

PATH = sys.argv[1]


page = """
<div id="logs">
    <ul>

    </ul>
</div>

<script>
var ws = new WebSocket("ws://localhost:8000/ws/");
const ul = document.querySelector("#logs > ul");

ws.onmessage = (event) => {
    const li = document.createElement("li");
    li.innerText = event.data;
    ul.appendChild(li);
};
</script>
"""


class Listener:
    def __init__(self, path: str, polling_interval: float = 0.5) -> None:
        self.path = path
        self.pos = 0
        self.polling_interval = polling_interval

    async def listen(self):
        loop = asyncio.get_running_loop()
        while True:
            lines = await loop.run_in_executor(None, self.poll)
            for line in lines:
                yield line
            await asyncio.sleep(self.polling_interval)

    def poll(self):
        with open(self.path, "r") as f:
            f.seek(self.pos)
            lines = f.readlines()
            self.pos = f.tell()
            return lines


app = FastAPI()


@app.get("/")
def index():
    return HTMLResponse(page)


@app.websocket("/ws/")
async def ws(websocket: WebSocket):
    await websocket.accept()
    port = websocket.client.port
    print(f"Client at :{port} connected")
    listener = Listener(PATH)
    try:
        async for line in listener.listen():
            await websocket.send_text(line)
    except WebSocketDisconnect:
        print(f"Client at {port} disconnected")


if __name__ == "__main__":
    uvicorn.run(app)

lines = await loop.run_in_executor(None, self.poll) — this line is not strictly necessary: polling should be fast in a local filesystem, so you can just do lines = self.poll(). But if, in theory, you query a database here, it may take longer, so it's better to run in executor (which is a way to make synchronous blocking code asynchronous and non-blocking by utilising threads).

Another thing I noticed is that you open your file only once and not before each polling. I thought it would work, but it doesn't. Apparently, when you call open(), python records file size at that moment, so even when something is appended to the file, this value doesn't update. So you need to open() the file each time if you want to get the new lines.

[bobbilichandu]

Thank you so much! I finally able to understand what I was doing wrong. That's a great code and very nice explanation.