Qwebsocket收发测试

​ 书接前文，这里写个Qwebsocket收发测试的程序，直接用python的websocket的库也可以，但websocket是个新东西，python3.6.8对这个支持并不是太好（python3.9往后的会好些）。

​ 开发环境：

​ python3.6.8

​ PyQt5==5.15.4

client：

iimport sys
from PyQt5.QtCore import QUrl, QTimer
from PyQt5.QtWidgets import QApplication
from PyQt5.QtWebSockets import QWebSocket

class WebSocketClient(QWebSocket):
    def __init__(self, url):
        super().__init__()
        self.connected.connect(self.on_connected)
        self.disconnected.connect(self.close)
        self.error.connect(self.on_error)
        self.textMessageReceived.connect(self.on_message_received)
        self.open(QUrl(url))
        self.timer = QTimer()
        self.timer.timeout.connect(self.send_msg)
        self.timer.start(1000)  # 每隔1秒触发一次定时器

    def on_connected(self):
        print("Connected to WebSocket server")

    def on_error(self, error_code):
        print(f"WebSocket error: {error_code}")

    def on_message_received(self, message):
        print(f"Received message from server: {message}")

    def send_msg(self):
        if self.isValid():
            self.sendTextMessage("Hello, server!")
        else:
            print("send fail!")

if __name__ == "__main__":
    app = QApplication(sys.argv)
    client = WebSocketClient("ws://127.0.0.1:8000")
    sys.exit(app.exec_())

server：

from PyQt5.QtCore import QCoreApplication
from PyQt5.QtWebSockets import QWebSocketServer, QWebSocket
from PyQt5.QtNetwork import QHostAddress

class WebSocketServer(QWebSocketServer):
    def __init__(self, address, port):
        super().__init__("WebSocket Server", QWebSocketServer.NonSecureMode)
        self.listen(QHostAddress(address), port)
        self.newConnection.connect(self.on_new_connection)

    def on_new_connection(self):
        self.client_socket = self.nextPendingConnection()
        self.client_socket.textMessageReceived.connect(self.on_message_received)
        # client_socket.disconnected.connect(self.on_client_disconnected)

    def on_message_received(self, message):
        client_socket = self.sender()
        print(f"Received message from client: {message}")
        # 在这里可以对接收到的消息进行处理
        client_socket.sendTextMessage("Echo: " + message)

    def on_client_disconnected(self):
        client_socket = self.sender()
        print("Client disconnected")

if __name__ == "__main__":
    app = QCoreApplication([])
    server = WebSocketServer("127.0.0.1", 8000)
    print(f"WebSocket server listening on {server.serverAddress().toString()}:{server.serverPort()}")
    app.exec_()