CameraMachine/python/sockets/socket_server.py

import json
import asyncio
from models import *
from .connect_manager import ConnectionManager

def jsonMessage(code=0, msg="", data: object = None):
    """json字符串数据"""
    jsonData = {"code": code, "msg": msg, "data": data}
    return json.dumps(jsonData)
manager = ConnectionManager()




@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            '''接收json数据'''
            data = await websocket.receive_json()
            # 处理发送消息逻辑
            await handlerSend(data,websocket)
            # await manager.broadcast(f"Client # says: {data}")
    except WebSocketDisconnect:
        data = jsonMessage(code=1,data=None,msg="socket 内部错误")
        await manager.broadcast(data,websocket)
        manager.disconnect(websocket)

async def handlerSend(receiveData,websocket):
    # 处理消息发送逻辑
    jsonType = receiveData.get("type")
    match jsonType:
        case 'ping':
            '''发送心跳'''
            data = jsonMessage("pong")
            await manager.send_personal_message(data,websocket)
        case 'pong':
            '''发送心跳'''
            pass