4단계: 채팅 만들기
1. 학습 목표
더보기
(1) 1~3단계에서 무엇이 준비되었나
- 1단계에서 확보한 것 (개념/구조)
- 클라이언트/서버 모델
- IP/Port/Socket
- Listen → Accept → Send/Receive 실행 흐름
- PySide6에서의 대응 구조
- 서버: QTcpServer
- 클라이언트/연결: QTcpSocket
- 이벤트: newConnection, readyRead, disconnected
- 2단계에서 확보한 것 (송수신 동작)
- readyRead가 “데이터가 도착했음”을 알려줌
- readAll()로 들어온 데이터를 꺼냄
- write()로 데이터를 전송함
- 에코 서버는 결국 “수신 후 같은 소켓에 재전송”
- 3단계에서 확보한 것 (다중 클라이언트 처리 구조)
- client_socket 하나로는 다중 연결을 처리할 수 없음
- 서버는 “연결된 소켓들을 저장”해야 함
- 서버는 “어떤 소켓에서 readyRead가 발생했는지”로 발신자를 식별함
- 브로드캐스트: 저장된 소켓 전체에 write()
- 유니캐스트: 특정 소켓 1개에만 write()
(2) 4단계 학습 목표
- 채팅은 결국 아래 3가지를 합친 구조입니다.
- 여러 클라이언트가 서버에 접속한다 (다중 연결)
- 누가 보냈는지 구분한다 (식별: 닉네임)
- 메시지를 전달한다
- 전체 전달: 브로드캐스트
- 개인 전달: 유니캐스트
2. 서버 설계
더보기
(1) 서버가 들고 있는 자료구조
- clients: set[QTcpSocket]
연결된 소켓 목록 - buffers: dict[QTcpSocket, bytearray]
소켓별 수신 버퍼(라인 단위 처리용) - nicknames: dict[QTcpSocket, str]
소켓별 닉네임(식별자)
(2) 서버 이벤트 흐름
- newConnection
- 소켓 생성 nextPendingConnection()
- 자료구조에 저장
- 기본 닉네임 부여
- 시그널 연결(readyRead, disconnected)
- readyRead(socket)
- 해당 소켓 버퍼에 데이터 누적
- "\n" 기준으로 한 줄씩 꺼내 처리
- 명령이면 처리 (NICK, TO)
- 아니면 브로드캐스트
- disconnected(socket)
- 자료구조에서 제거
- 전체 공지(퇴장)
3. 서버
더보기
# chat_server.py
import sys
from PySide6.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout, QTextEdit
from PySide6.QtNetwork import QTcpServer, QTcpSocket, QHostAddress
class ChatServerWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("채팅 서버")
self.resize(560, 420)
# UI
central = QWidget(self)
self.setCentralWidget(central)
layout = QVBoxLayout(central)
self.log = QTextEdit(self)
self.log.setReadOnly(True)
layout.addWidget(self.log)
# Network
self.server = QTcpServer(self)
self.server.newConnection.connect(self.on_new_connection)
# 3단계(소켓 관리) 핵심: 다중 소켓 저장 구조
self.clients = set()
self.buffers = {} # socket -> bytearray
self.nicknames = {} # socket -> nickname
host = QHostAddress("127.0.0.1")
port = 8888
if self.server.listen(host, port):
self.append_log(f"서버 시작: {host.toString()}:{port}")
else:
self.append_log("서버 시작 실패")
def append_log(self, text: str):
self.log.append(text)
# 1) 새 연결 수락
def on_new_connection(self):
sock = self.server.nextPendingConnection()
if sock is None:
return
self.clients.add(sock)
self.buffers[sock] = bytearray()
# 기본 닉네임(학습용: 소켓 디스크립터로 유니크 값 부여)
default_nick = f"Guest-{int(sock.socketDescriptor())}"
self.nicknames[sock] = default_nick
# 중요: 어떤 소켓이 readyRead를 발생시키는지 구분해야 한다
sock.readyRead.connect(lambda s=sock: self.on_ready_read(s))
sock.disconnected.connect(lambda s=sock: self.on_disconnected(s))
peer = f"{sock.peerAddress().toString()}:{int(sock.peerPort())}"
self.append_log(f"접속: {default_nick} ({peer})")
# 접속자에게 안내
self.send_to(sock, "시스템: 닉네임 설정 예) NICK 홍길동")
self.broadcast(f"시스템: {default_nick} 님이 입장했습니다.")
# 2) 데이터 수신 처리 (라인 단위)
def on_ready_read(self, sock: QTcpSocket):
data = bytes(sock.readAll())
if not data:
return
buf = self.buffers.get(sock)
if buf is None:
self.buffers[sock] = bytearray()
buf = self.buffers[sock]
buf.extend(data)
# "\n" 기준으로 한 줄씩 처리
while b"\n" in buf:
line, _, rest = buf.partition(b"\n")
self.buffers[sock] = rest
buf = rest
text = line.decode("utf-8", errors="replace").strip()
if not text:
continue
self.handle_line(sock, text)
# 3) 프로토콜 처리
def handle_line(self, sock: QTcpSocket, text: str):
# 닉네임 변경: NICK <name>
if text.upper().startswith("NICK "):
new_nick = text[5:].strip()
if not new_nick:
self.send_to(sock, "시스템: 닉네임이 비어있습니다.")
return
old = self.nicknames.get(sock, "Unknown")
self.nicknames[sock] = new_nick
self.append_log(f"닉네임 변경: {old} -> {new_nick}")
self.broadcast(f"시스템: {old} 님이 {new_nick} 님으로 변경했습니다.")
return
# 유니캐스트: TO <target> <msg>
if text.upper().startswith("TO "):
parts = text.split(" ", 2) # ["TO", target, msg]
if len(parts) < 3:
self.send_to(sock, "시스템: 형식 오류. 예) TO 홍길동 안녕하세요")
return
target_nick = parts[1].strip()
msg = parts[2].strip()
if not target_nick or not msg:
self.send_to(sock, "시스템: 대상 또는 메시지가 비어있습니다.")
return
sender_nick = self.nicknames.get(sock, "Unknown")
target_sock = self.find_socket_by_nick(target_nick)
if target_sock is None:
self.send_to(sock, f"시스템: 대상 '{target_nick}' 을(를) 찾을 수 없습니다.")
return
# 대상에게만 전송
self.send_to(target_sock, f"(개인) {sender_nick} -> {target_nick}: {msg}")
# 발신자에게도 확인 메시지(학습용)
self.send_to(sock, f"(개인 전송 완료) {sender_nick} -> {target_nick}: {msg}")
return
# 일반 메시지: 브로드캐스트
sender_nick = self.nicknames.get(sock, "Unknown")
self.append_log(f"수신: {sender_nick}: {text}")
self.broadcast(f"{sender_nick}: {text}")
# 닉네임으로 소켓 찾기
def find_socket_by_nick(self, nickname: str):
for s, nick in self.nicknames.items():
if nick == nickname:
return s
return None
# 브로드캐스트
def broadcast(self, message: str):
payload = (message + "\n").encode("utf-8")
dead = []
for s in list(self.clients):
try:
s.write(payload)
except Exception:
dead.append(s)
# 예외 소켓 정리(안전)
for s in dead:
self.cleanup_socket(s)
# 유니캐스트(소켓 1개)
def send_to(self, sock: QTcpSocket, message: str):
try:
sock.write((message + "\n").encode("utf-8"))
except Exception:
self.cleanup_socket(sock)
# 연결 종료 처리
def on_disconnected(self, sock: QTcpSocket):
nick = self.nicknames.get(sock, "Unknown")
peer = f"{sock.peerAddress().toString()}:{int(sock.peerPort())}"
self.append_log(f"퇴장: {nick} ({peer})")
self.cleanup_socket(sock)
self.broadcast(f"시스템: {nick} 님이 퇴장했습니다.")
def cleanup_socket(self, sock: QTcpSocket):
if sock in self.clients:
self.clients.remove(sock)
self.buffers.pop(sock, None)
self.nicknames.pop(sock, None)
sock.deleteLater()
if __name__ == "__main__":
app = QApplication(sys.argv)
w = ChatServerWindow()
w.show()
sys.exit(app.exec())
4. 클라이언트
더보기
# chat_client.py
import sys
from PySide6.QtWidgets import (
QApplication, QMainWindow, QWidget,
QVBoxLayout, QHBoxLayout,
QTextEdit, QLineEdit, QPushButton
)
from PySide6.QtNetwork import QTcpSocket
class ChatClientWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("채팅 클라이언트")
self.resize(560, 460)
# UI
central = QWidget(self)
self.setCentralWidget(central)
root = QVBoxLayout(central)
self.chat = QTextEdit(self)
self.chat.setReadOnly(True)
root.addWidget(self.chat)
# 닉네임 설정 라인
row1 = QHBoxLayout()
self.nick_input = QLineEdit(self)
self.nick_input.setPlaceholderText("닉네임 입력 (예: 홍길동)")
self.btn_nick = QPushButton("닉네임 설정", self)
self.btn_nick.clicked.connect(self.send_nick)
row1.addWidget(self.nick_input)
row1.addWidget(self.btn_nick)
root.addLayout(row1)
# 유니캐스트 대상 라인
row2 = QHBoxLayout()
self.to_input = QLineEdit(self)
self.to_input.setPlaceholderText("유니캐스트 대상 닉네임 (예: 김철수)")
self.btn_to = QPushButton("Unicast 전송", self)
self.btn_to.clicked.connect(self.send_unicast)
row2.addWidget(self.to_input)
row2.addWidget(self.btn_to)
root.addLayout(row2)
# 메시지 입력 라인(브로드캐스트)
row3 = QHBoxLayout()
self.msg_input = QLineEdit(self)
self.msg_input.setPlaceholderText("메시지 입력 (일반 전송은 브로드캐스트)")
self.btn_send = QPushButton("Send", self)
self.btn_send.clicked.connect(self.send_broadcast)
row3.addWidget(self.msg_input)
row3.addWidget(self.btn_send)
root.addLayout(row3)
# Network
self.socket = QTcpSocket(self)
self.socket.connected.connect(self.on_connected)
self.socket.disconnected.connect(self.on_disconnected)
self.socket.readyRead.connect(self.on_ready_read)
self.buffer = bytearray()
self.socket.connectToHost("127.0.0.1", 8888)
def log(self, text: str):
self.chat.append(text)
def on_connected(self):
self.log("시스템: 서버에 연결되었습니다.")
def on_disconnected(self):
self.log("시스템: 서버 연결이 종료되었습니다.")
# 브로드캐스트 전송 (일반 메시지)
def send_broadcast(self):
text = self.msg_input.text().strip()
if not text:
return
self.socket.write((text + "\n").encode("utf-8"))
self.msg_input.clear()
# 닉네임 설정
def send_nick(self):
nick = self.nick_input.text().strip()
if not nick:
return
self.socket.write((f"NICK {nick}\n").encode("utf-8"))
# 유니캐스트 전송: TO <대상> <메시지>
def send_unicast(self):
target = self.to_input.text().strip()
msg = self.msg_input.text().strip()
if not target or not msg:
return
self.socket.write((f"TO {target} {msg}\n").encode("utf-8"))
self.msg_input.clear()
# 수신 처리(라인 단위)
def on_ready_read(self):
data = bytes(self.socket.readAll())
self.buffer.extend(data)
while b"\n" in self.buffer:
line, _, rest = self.buffer.partition(b"\n")
self.buffer = rest
text = line.decode("utf-8", errors="replace").strip()
if text:
self.log(text)
if __name__ == "__main__":
app = QApplication(sys.argv)
w = ChatClientWindow()
w.show()
sys.exit(app.exec())
5. 과제
더보기
닉네임 추가하기- 서버에서 접속자 수 표시
- 특정 단어 금지 필터
- 서버에서만 공지 메시지 보내기
- 시간 표시 추가