42강 귓속말
42강 구현 중심 ⏱ 약 20분
0. 학습 목표
→ 특정 한 명에게만 전달되는 귓속말을 추가합니다.
더보기
0.1 이번 글에서 다룰 내용
이번 글은 구현 중심 강의입니다.
41강까지 만든 서버는 메시지를 항상 전체에 브로드캐스트합니다.
귓속말은 그 반대입니다. 보낸 사람과 받는 사람 단 둘에게만 전달해야 합니다.
이번 강의에서 프로토콜에 whisper 타입을 추가하고,
닉네임으로 소켓을 역조회해 해당 한 명에게만 메시지를 보내는 기능을 구현합니다.
대상이 없을 때 오류 처리도 함께 다룹니다.
| 구분 | 내용 |
| 이해할 것 | 닉네임으로 소켓을 역조회하는 방식, 전체 전달과 1:1 전달의 차이 |
| 만들 것 | server/app.py 수정, server/test_client.html 수정 |
| 확인할 것 | 귓속말이 보낸 사람과 받는 사람에게만 전달되는 것, 대상이 없을 때 오류 안내가 보낸 사람에게만 오는 것 |
0.2 이번 강의에서 수정하는 구조
fastapi-chat/
├── server/
│ ├── app.py ← ✏️ get_socket_by_nickname() 추가, whisper 처리 추가
│ └── test_client.html ← ✏️ 귓속말 입력 형식 안내, whisper 수신 표시 추가
└── desktop/ ← 43강에서 사용 예정(✏️ 수정 · 표시 없음은 변경 없음)
1. 서버 - 소켓 역조회와 귓속말 처리 추가하기
→ 닉네임으로 소켓을 찾는 메서드를 추가하고, whisper 타입을 처리합니다.
더보기
브로드캐스트는 active_connections의 모든 소켓에 전송하면 끝입니다.
귓속말은 다릅니다. target 닉네임을 받아서 그 닉네임을 가진 소켓 하나를 찾아야 합니다.
41강에서 예고한 역조회 메서드 get_socket_by_nickname()을 먼저 추가합니다.
class ConnectionManager:
def __init__(self):
self.active_connections: Dict[WebSocket, str] = {}
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections[websocket] = ""
def disconnect(self, websocket: WebSocket):
self.active_connections.pop(websocket, None)
def set_nickname(self, websocket: WebSocket, nickname: str):
self.active_connections[websocket] = nickname
def get_nickname(self, websocket: WebSocket) -> str:
return self.active_connections.get(websocket, "")
def get_nicknames(self) -> list:
return [n for n in self.active_connections.values() if n]
def get_socket_by_nickname(self, nickname: str): # ➕ 닉네임 → 소켓 역조회
for ws, name in self.active_connections.items():
if name == nickname:
return ws
return None # ➕ 없으면 None 반환
async def broadcast(self, message: dict):
for connection in self.active_connections:
await connection.send_json(message)Dict[WebSocket, str]는 소켓 → 닉네임 방향으로 설계되어 있습니다.
역방향 조회는 딕셔너리를 순회하며 값이 일치하는 키를 찾습니다. 접속자 수가 적은 채팅 서버에서는 충분히 빠릅니다.
이제 websocket_endpoint에 whisper 처리를 추가합니다. 기존 if/elif 분기에 하나를 더 붙입니다.
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_json()
msg_type = data.get("type", "")
if msg_type == "join":
nickname = data.get("sender", "익명")
manager.set_nickname(websocket, nickname)
await manager.broadcast({
"type": "system",
"text": f"{nickname} 님이 입장했습니다."
})
await manager.broadcast({
"type": "user_list",
"users": manager.get_nicknames()
})
elif msg_type == "chat":
await manager.broadcast({
"type": "chat",
"sender": manager.get_nickname(websocket),
"text": data.get("text", "")
})
elif msg_type == "whisper": # ➕ 귓속말 처리
target_name = data.get("target", "")
text = data.get("text", "")
sender_name = manager.get_nickname(websocket)
target_ws = manager.get_socket_by_nickname(target_name) # ➕ 대상 소켓 조회
if target_ws is None: # ➕ 대상이 없을 때
await websocket.send_json({
"type": "system",
"text": f"'{target_name}' 님은 현재 접속해 있지 않습니다."
})
else:
whisper_msg = { # ➕ 귓속말 메시지 구성
"type": "whisper",
"sender": sender_name,
"text": text
}
await target_ws.send_json(whisper_msg) # ➕ 받는 사람에게 전송
await websocket.send_json(whisper_msg) # ➕ 보낸 사람에게도 전송 (확인용)
except Exception:
nickname = manager.get_nickname(websocket)
manager.disconnect(websocket)
if nickname:
await manager.broadcast({
"type": "system",
"text": f"{nickname} 님이 퇴장했습니다."
})
await manager.broadcast({
"type": "user_list",
"users": manager.get_nicknames()
})귓속말에서 보낸 사람에게도 같은 메시지를 돌려보내는 이유가 있습니다. 전송이 성공했는지 보낸 사람이 확인할 방법이 없으면 메시지가 전달됐는지 알 수 없습니다. 같은 whisper_msg 객체를 두 소켓에 보내면 보낸 사람 화면에도 귓속말이 표시되어 확인이 됩니다.
오류 메시지는 manager.broadcast()가 아니라 websocket.send_json()으로 보냅니다. 대상이 없다는 안내는 귓속말을 시도한 사람에게만 전달해야 하기 때문입니다.
2. 테스트 클라이언트 - 귓속말 입력과 표시 추가하기
→ /귓속말 형식을 파싱해 whisper 메시지로 전송하고, 받은 귓속말을 구분해 표시합니다.
더보기
귓속말 전송 방식은 IRC 스타일을 그대로 씁니다. 입력창에 /w 김철수 안녕처럼 입력하면 sendMsg()가 슬래시 명령어인지 판단해 whisper 메시지로 변환해 보냅니다. 일반 채팅 전송 로직과 같은 입력창을 공유해서 UI를 추가하지 않아도 됩니다.
바뀌는 부분은 sendMsg()와 ws.onmessage 두 곳입니다.
function sendMsg() {
const input = document.getElementById("msg");
const text = input.value.trim();
if (!text) return;
if (text.startsWith("/w ")) { // ➕ 귓속말 명령어 감지
const parts = text.slice(3).split(" "); // ➕ "/w " 이후 파싱
const target = parts[0]; // ➕ 첫 단어 = 대상 닉네임
const whisperText = parts.slice(1).join(" "); // ➕ 나머지 = 전달 내용
if (target && whisperText) {
ws.send(JSON.stringify({
type: "whisper",
target: target,
text: whisperText
}));
}
} else {
ws.send(JSON.stringify({ type: "chat", sender: myNickname, text: text }));
}
input.value = "";
}ws.onmessage에 whisper 타입 처리를 추가합니다.
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === "system") {
addLog(`[시스템] ${data.text}`);
} else if (data.type === "chat") {
addLog(`${data.sender}: ${data.text}`);
} else if (data.type === "user_list") {
updateUserList(data.users);
} else if (data.type === "whisper") { // ➕ 귓속말 수신 표시
addLog(`[귓속말 from ${data.sender}] ${data.text}`);
}
};입력창 아래에 사용 방법 한 줄을 추가해두면 학생들이 명령어를 기억하기 쉽습니다.
<p style="margin:4px 0; font-size:12px; color:#888;">귓속말: /w 닉네임 내용</p> <!-- ➕ -->
3. 실행 결과 확인하기
→ 세 창으로 귓속말이 당사자에게만 전달되는지, 대상 없는 귓속말 오류를 확인합니다.
더보기
3.1 귓속말 전달 확인
브라우저 세 개를 열고 홍길동·김철수·이영희로 각각 입장합니다. 홍길동 창에서 /w 김철수 안녕하세요를 입력하고 보냅니다.
[창 1 — 홍길동] [창 2 — 김철수] [창 3 — 이영희]
/w 김철수 안녕하세요 → 전송
[귓속말 from 홍길동] 안녕하세요 [귓속말 from 홍길동] 안녕하세요 (아무것도 표시되지 않음)이영희 창에는 아무것도 표시되지 않는 것이 정상입니다. 서버가 target_ws 하나와 발신자 소켓에만 전송했기 때문입니다.
✔ 확인 기준: 귓속말이 보낸 사람과 받는 사람 두 창에만 나타나고, 세 번째 창에는 아무것도 표시되지 않으면 완료.
3.2 대상 없는 귓속말 오류 확인
접속하지 않은 닉네임으로 귓속말을 보냅니다. 예를 들어 /w 박민수 안녕처럼 입력합니다.
[창 1 — 홍길동]
/w 박민수 안녕 → 전송
[시스템] '박민수' 님은 현재 접속해 있지 않습니다.오류 안내는 홍길동 창에만 표시됩니다. 다른 창에는 아무것도 나타나지 않습니다.
✔ 확인 기준: 존재하지 않는 닉네임으로 귓속말을 보냈을 때 오류 안내가 보낸 사람 창에만 표시되면 완료. 오류가 전체에 브로드캐스트되면 websocket.send_json()이 아니라 manager.broadcast()로 잘못 작성한 것입니다.
4. 최종 코드 정리하기
→ 이번 강의에서 완성한 두 파일 전체를 정리합니다.
더보기
4.1 server/app.py
from fastapi import FastAPI, WebSocket
from typing import Dict
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections: Dict[WebSocket, str] = {}
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections[websocket] = ""
def disconnect(self, websocket: WebSocket):
self.active_connections.pop(websocket, None)
def set_nickname(self, websocket: WebSocket, nickname: str):
self.active_connections[websocket] = nickname
def get_nickname(self, websocket: WebSocket) -> str:
return self.active_connections.get(websocket, "")
def get_nicknames(self) -> list:
return [n for n in self.active_connections.values() if n]
def get_socket_by_nickname(self, nickname: str):
for ws, name in self.active_connections.items():
if name == nickname:
return ws
return None
async def broadcast(self, message: dict):
for connection in self.active_connections:
await connection.send_json(message)
manager = ConnectionManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_json()
msg_type = data.get("type", "")
if msg_type == "join":
nickname = data.get("sender", "익명")
manager.set_nickname(websocket, nickname)
await manager.broadcast({
"type": "system",
"text": f"{nickname} 님이 입장했습니다."
})
await manager.broadcast({
"type": "user_list",
"users": manager.get_nicknames()
})
elif msg_type == "chat":
await manager.broadcast({
"type": "chat",
"sender": manager.get_nickname(websocket),
"text": data.get("text", "")
})
elif msg_type == "whisper":
target_name = data.get("target", "")
text = data.get("text", "")
sender_name = manager.get_nickname(websocket)
target_ws = manager.get_socket_by_nickname(target_name)
if target_ws is None:
await websocket.send_json({
"type": "system",
"text": f"'{target_name}' 님은 현재 접속해 있지 않습니다."
})
else:
whisper_msg = {
"type": "whisper",
"sender": sender_name,
"text": text
}
await target_ws.send_json(whisper_msg)
await websocket.send_json(whisper_msg)
except Exception:
nickname = manager.get_nickname(websocket)
manager.disconnect(websocket)
if nickname:
await manager.broadcast({
"type": "system",
"text": f"{nickname} 님이 퇴장했습니다."
})
await manager.broadcast({
"type": "user_list",
"users": manager.get_nicknames()
})
4.2 server/test_client.html
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<title>WebSocket 테스트</title>
</head>
<body>
<h3>WebSocket 채팅 테스트</h3>
<div id="join-area">
<input id="nickname" type="text" placeholder="닉네임 입력" style="width:200px;">
<button onclick="joinChat()">입장</button>
</div>
<div id="chat-area" style="display:none;">
<div style="display:flex; gap:16px;">
<div style="flex:1;">
<div id="log" style="border:1px solid #ccc; height:200px; overflow-y:auto; padding:8px;"></div>
<br>
<input id="msg" type="text" placeholder="메시지 입력" style="width:260px;">
<button onclick="sendMsg()">보내기</button>
<p style="margin:4px 0; font-size:12px; color:#888;">귓속말: /w 닉네임 내용</p>
</div>
<div style="width:140px;">
<b>접속자</b>
<ul id="user-list" style="padding-left:16px; margin-top:4px;"></ul>
</div>
</div>
</div>
<script>
let ws;
let myNickname = "";
function joinChat() {
myNickname = document.getElementById("nickname").value.trim();
if (!myNickname) return;
ws = new WebSocket("ws://127.0.0.1:8000/ws");
ws.onopen = () => {
ws.send(JSON.stringify({ type: "join", sender: myNickname }));
document.getElementById("join-area").style.display = "none";
document.getElementById("chat-area").style.display = "block";
};
ws.onclose = () => addLog("[연결 종료]");
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === "system") {
addLog(`[시스템] ${data.text}`);
} else if (data.type === "chat") {
addLog(`${data.sender}: ${data.text}`);
} else if (data.type === "user_list") {
updateUserList(data.users);
} else if (data.type === "whisper") {
addLog(`[귓속말 from ${data.sender}] ${data.text}`);
}
};
}
function updateUserList(users) {
const ul = document.getElementById("user-list");
ul.innerHTML = users.map(u => `<li>${u}</li>`).join("");
}
function sendMsg() {
const input = document.getElementById("msg");
const text = input.value.trim();
if (!text) return;
if (text.startsWith("/w ")) {
const parts = text.slice(3).split(" ");
const target = parts[0];
const whisperText = parts.slice(1).join(" ");
if (target && whisperText) {
ws.send(JSON.stringify({ type: "whisper", target: target, text: whisperText }));
}
} else {
ws.send(JSON.stringify({ type: "chat", sender: myNickname, text: text }));
}
input.value = "";
}
function addLog(text) {
const log = document.getElementById("log");
log.innerHTML += text + "<br>";
log.scrollTop = log.scrollHeight;
}
document.addEventListener("keydown", (e) => {
if (e.key === "Enter") {
if (document.getElementById("chat-area").style.display === "block") {
sendMsg();
} else {
joinChat();
}
}
});
</script>
</body>
</html>
4.3 최종 확인 표
| 코드 | 역할 |
get_socket_by_nickname(nickname) |
닉네임으로 소켓을 역조회 — 없으면 None 반환 |
if target_ws is None |
대상 없음 — 보낸 사람에게만 오류 안내 |
await target_ws.send_json(whisper_msg) |
받는 사람에게 귓속말 전송 |
await websocket.send_json(whisper_msg) |
보낸 사람에게도 전송 — 전달 성공 확인용 |
text.startsWith("/w ") |
귓속말 명령어 감지 |
text.slice(3).split(" ") |
"/w " 이후를 공백으로 나눠 닉네임과 내용 분리 |
[귓속말 from ${data.sender}] |
일반 채팅과 시각적으로 구분되는 표시 |
→ 다음 강의 (43강): 39~42강에서 브라우저로 검증한 이 서버에 PySide6 데스크톱 앱을 연결합니다. 앱의 소켓 통신을 WebSocket으로 교체하는 것이 핵심입니다. PySide6의 Qt 이벤트 루프와 asyncio 이벤트 루프를 함께 돌리는 방법을 다룹니다.