Python Websockets舊版改寫為15.0.1版目前寫法範例

這次筆記是修正Websockets改為15版的寫法
因為把Python昇級，加上Websockets自己也升級
所以之前這篇使用Python的websockets套件控制樹莓派的寫法不能用了
要改新的寫法
這部分其實很簡單
但我因為太久沒碰這隻程式忘光了基礎
還是搞了不短的一段時間(花了半天 Orz)
為了之後能快速使用，留下這個記錄吧

廢話不多說，直接用原始碼說明
原本是長這樣子

import asyncio
import websockets
import pathlib

# 設定全域變數
mission_lists = []
    
#無限循環用的伺服程式
async def custom_server():
    while True:
        #避免一直陷在迴圈中無法異步化，會中止5秒，讓其它websocket任務可以動作
        await asyncio.sleep(5)
        #清單不為零，開始執行任務清單
        if len(mission_lists) > 0:
            app_log.info('start mission')
            mission = mission_lists.pop()
            await do_mission(mission)

#主動發送訊息用的生產者程式
async def message_handler(websocket, path):
    while True:
        message = "目前處理狀況"
        await websocket.send(message)
        await asyncio.sleep(3) #3秒的間隔發送        

#接收並處理使用者訊息的生產者程式
async def cmd_handler(websocket, path):
    async for message in websocket:
        #取得送進來的訊息
        data = json.loads(message)
        #處理訊息轉為任務
        mission = process(data)
        #加入任務清單
        mission_lists.append(mission)

#Websocket伺服器主程式, 只有消費者與生產者
async def main_handler(websocket, path):
    consumer_task = asyncio.ensure_future(
        cmd_handler(websocket, path))
    producer_task = asyncio.ensure_future(
        message_handler(websocket, path))
    done, pending = await asyncio.wait(
        [consumer_task, producer_task],
        return_when=asyncio.FIRST_COMPLETED,
    )
    #當沒有websocket連線時，停止任務執行
    for task in pending:
        task.cancel()
        
#加入SSL通信功能
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
ssl_context.load_cert_chain(localhost_pem)

#設定websocket伺服器
start_server = websockets.serve(main_handler,
                     "0.0.0.0", 
                     8765, 
                     ssl=ssl_context)

#將websockets伺服任務丟入異步的迴圈中
asyncio.get_event_loop().run_until_complete(start_server)
#補上自己額外的任務，丟入異步的迴圈中
asyncio.get_event_loop().run_until_complete(custom_server())
#讓異步的迴圈一直執行
asyncio.get_event_loop().run_forever()

改版成15版後變成這個樣子

import asyncio
import websockets
import pathlib

#設定全域變數
mission_lists = []

#無限循環用的伺服程式
#這裡都不用變
async def custom_server():
    while True:
        #避免一直陷在迴圈中無法異步化，會中止5秒，讓其它websocket任務可以動作
        await asyncio.sleep(5)
        #清單不為零，開始下載
        if len(mission_lists) > 0:
            app_log.info('start mission')
            mission = mission_lists.pop()
            await do_mission(mission)

#主動發送訊息用的生產者程式
#這裡就輸入變數取消path這項
async def message_handler(websocket): #取消path
    while True:
        message = "目前處理狀況"
        await websocket.send(message)
        await asyncio.sleep(3) #3秒的間隔發送        

#接收並處理使用者訊息的生產者程式
#一樣，輸入變數取消path這項
async def cmd_handler(websocket): #取消path
    async for message in websocket:
        data = json.loads(message)
        mission = process(data)
        mission_lists.append(mission)

#Websocket伺服器主程式, 只有消費者與生產者
#一樣只有取消path的輸入，其它不變
async def main_handler(websocket):
    consumer_task = asyncio.ensure_future(
        cmd_handler(websocket))
    producer_task = asyncio.ensure_future(
        message_handler(websocket))
    done, pending = await asyncio.wait(
        [consumer_task, producer_task],
        return_when=asyncio.FIRST_COMPLETED,
    )
    #當沒有websocket連線時，停止任務執行
    for task in pending:
        task.cancel()

#新的伺服啟動寫法
#建立異步執行專用函數，如果沒有其它要跑的項目，直接用main也行
async def main_server():

    #加入SSL通信功能
    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
    ssl_context.load_cert_chain(localhost_pem)

    # 異步建立自己的伺服任務
    asyncio.create_task(custom_server(mission_lists))

    #設定websocket伺服器
    async with websockets.serve(
        main_handler,
        "0.0.0.0",
        8765,
        ssl=ssl_context
    ) as server:
        # 設定這個伺服器持續運行
        await server.serve_forever()

#以異步運轉，正式執行設定好的websockets伺服器
asyncio.run(main_server())

主要的差異其實是Websockets伺服的啟動
改用常用的異步執行asyncio.run去跑
然後還有函數的引數變不同了，那個path被拿掉了
原本消費者與生產者的主架構是沒什麼差異的
新的說明可以看下面官方的頁面
https://websockets.readthedocs.io/en/stable/howto/patterns.html
所以說改新寫法真的很簡單