websocket_server.py 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
  1. import asyncio
  2. import json
  3. import threading
  4. import websockets
  5. ##
  6. from message_base import MessageBase
  7. class WebServer:
  8. def __init__(self, host, port, message_base: MessageBase):
  9. self.host = host
  10. self.port = port
  11. self.clients = []
  12. self.message_base = message_base
  13. async def echo(self, websocket, path):
  14. self.clients.append(websocket)
  15. client_ip, client_port = websocket.remote_address
  16. print(f"客户连接:{client_ip}:{client_port}")
  17. while True:
  18. try:
  19. recv_text = await websocket.recv()
  20. if recv_text:
  21. print("收到", recv_text)
  22. self.message_base.add("ws_recv", recv_text)
  23. send_text = self.message_base.get("ws_send")
  24. if send_text:
  25. print("发送", send_text)
  26. await websocket.send(send_text)
  27. # data = json.loads(recv_text)
  28. # device = data.get("device")
  29. # if device:
  30. # self.message_base.add(device, data)
  31. # else:
  32. # continue
  33. except websockets.ConnectionClosed:
  34. print("ConnectionClosed...") # 链接断开
  35. self.clients.remove(websocket)
  36. break
  37. except websockets.InvalidState:
  38. print("InvalidState...") # 无效状态
  39. self.clients.remove(websocket)
  40. break
  41. except Exception as e:
  42. print(e)
  43. def connect(self):
  44. print("连接成功!")
  45. asyncio.set_event_loop(asyncio.new_event_loop())
  46. start_server = websockets.serve(self.echo, self.host, self.port)
  47. asyncio.get_event_loop().run_until_complete(start_server)
  48. asyncio.get_event_loop().run_forever()
  49. def run(self):
  50. t = threading.Thread(target=self.connect)
  51. t.start()
  52. print("已启动!")
  53. if __name__ == '__main__':
  54. mb = MessageBase()
  55. ws = WebServer("192.168.6.28", 8001, mb)
  56. ws.run()