Python Falcon - WebSocket

WebSocket 是客戶端和伺服器之間的一種持久連線，用於在兩者之間提供雙向、全雙工通訊。通訊透過單個 TCP/IP 套接字連線在 HTTP 上進行。它可以看作是 HTTP 的升級，而不是協議本身。

HTTP 的侷限性之一在於它是一種嚴格的半雙工或單向協議。而 WebSocket 則允許我們傳送基於訊息的資料，類似於 UDP，但具有 TCP 的可靠性。WebSocket 使用 HTTP 作為初始傳輸機制，但在收到 HTTP 響應後保持 TCP 連線存活。同一個連線物件可以用於客戶端和伺服器之間的雙向通訊。因此，可以使用 WebSocket API 構建即時應用程式。

Falcon 的 WebSocket 支援僅適用於 ASGI 應用程式。為了提供 WebSocket 功能，資源類應該具有on_websocket() 響應器協程。

async def on_websocket(self, req, ws):
   . . .

WebSocket 請求也可以被 Hook 和中介軟體攔截。它傳遞的是falcon.asgi.WebSocket 物件，而不是 Response 物件。

WebSocket 在 Falcon 中是如何工作的？

下面的例子演示了 WebSocket 在 Falcon 應用程式中的工作方式。首先，我們有一個on_get() 響應器來渲染模板。

示例

客戶端瀏覽器顯示一個帶有文字欄位和按鈕的表單，單擊按鈕時，將建立一個websocket 物件，並觸發on_websocket() 響應器。它接收使用者輸入的訊息，並在前面加上“The message text was”字首後將其回顯給客戶端。

import falcon
import falcon.asgi
import jinja2
html = """
<!DOCTYPE html>
<html>
   <head>
      <title>Chat</title>
   </head>
   <body>
      <script>
         var ws = new WebSocket("ws://:8000/hello");
         ws.onmessage = function(event) {
            var messages =document.getElementById('messages')
            var message = document.createElement('li')
            var content = document.createTextNode(event.data)
            message.appendChild(content)
            messages.appendChild(message)
         };
         function sendMessage(event) {
            var input = document.getElementById("messageText")
            ws.send(input.value)
            input.value = ''
            event.preventDefault()
         }
      </script>
      <h1>WebSocket Chat</h1>
      <form action="" onsubmit="sendMessage(event)">
         <input type="text" id="messageText" autocomplete="off"/>
         <button>Send</button>
      </form>
      <ul id='messages'></ul>
   </body>
</html>
"""
class HelloResource:
   async def on_get(self, req, resp):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = 'text/html'
      template=jinja2.Template(html)
      resp.body=template.render()
   async def on_websocket(self, req, websocket):
      await websocket.accept()
      while True:
         data = await websocket.receive_text()
         await websocket.send_text(f"Message text was: {data}")
app = falcon.asgi.App()
hello = HelloResource()
app.add_route('/hello', hello)
import uvicorn
if __name__ == "__main__":
   uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

輸出

啟動 Uvicorn 伺服器並訪問https://:8000/ws URL 來顯示聊天表單。

輸入一些文字並按下傳送按鈕。

列印頁面