63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import sys
|
|
from PySide6.QtCore import QCoreApplication, QByteArray, Slot
|
|
from PySide6.QtNetwork import QTcpServer, QTcpSocket, QHostAddress
|
|
from PySide6.QtCore import Signal
|
|
|
|
|
|
class TcpServer(QCoreApplication):
|
|
def __init__(self, host, port):
|
|
super().__init__(sys.argv)
|
|
|
|
# 创建一个 TCP 服务器对象
|
|
self.server = QTcpServer(self)
|
|
self.host = host
|
|
self.port = port
|
|
|
|
# 连接信号，新的客户端连接时触发
|
|
self.server.newConnection.connect(self.on_new_connection)
|
|
|
|
# 绑定并开始监听指定的地址和端口
|
|
if not self.server.listen(QHostAddress(self.host), self.port):
|
|
print(f"Server could not start on {self.host}:{self.port}")
|
|
sys.exit(1)
|
|
|
|
print(f"Server started on {self.host}:{self.port}")
|
|
|
|
@Slot()
|
|
def on_new_connection(self):
|
|
# 获取客户端连接的 socket 对象
|
|
client_socket = self.server.nextPendingConnection()
|
|
|
|
# 连接信号
|
|
client_socket.readyRead.connect(lambda: self.on_ready_read(client_socket))
|
|
client_socket.disconnected.connect(lambda: self.on_disconnected(client_socket))
|
|
|
|
print(f"New connection from {client_socket.peerAddress().toString()}:{client_socket.peerPort()}")
|
|
|
|
# 发送欢迎消息给客户端
|
|
client_socket.write(b"Hello from server!")
|
|
client_socket.flush() # 确保数据已发送
|
|
print("Welcome message sent to client.")
|
|
|
|
@Slot()
|
|
def on_ready_read(self, client_socket: QTcpSocket):
|
|
# 读取客户端发送的数据
|
|
data = client_socket.readAll()
|
|
print(f"Received from client: {data.data().decode()}")
|
|
|
|
# 发送响应给客户端
|
|
response = "Server has received your message."
|
|
client_socket.write(response.encode())
|
|
client_socket.flush()
|
|
print(f"Sent to client: {response}")
|
|
|
|
@Slot()
|
|
def on_disconnected(self, client_socket: QTcpSocket):
|
|
print(f"Connection from {client_socket.peerAddress().toString()}:{client_socket.peerPort()} closed.")
|
|
client_socket.deleteLater() # 清理套接字资源
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = TcpServer("127.0.0.1", 1234)
|
|
sys.exit(app.exec())
|