Ⅳ. 강사님 제공 예제코드
라이브러리 설치
1) pip install channels-redis==2.4.2
2) pip install channels==3.0.4
3) python manage.py shell
4)
import channels.layers
channel_layer = channels.layers.get_channel_layer()
from asgiref.sync import async_to_sync
async_to_sync(channel_layer.send)('test_channel', {'type': 'hello'})
async_to_sync(channel_layer.receive)('test_channel')
--------------------
# {'type': 'hello'}
5) pip install daphne
Plain Text
복사
•
위 순서 지켜서 해야 한다고 합니다.
•
daphne : django용 asgi middleware
•
channel : django에서 채팅 websocket을 구현할 수 있게 해주는 라이브러리
•
channel-redis : redis와 연결해주는 라이브러리
Redis 설치(윈도우)
releases ← 여기에서 최신 버전 .msi 설치customers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
self.room_group_name = "chat_%s" % self.room_name
# Join room group
await self.channel_layer.group_add(self.room_group_name, self.channel_name)
await self.accept()
async def disconnect(self, close_code):
# Leave room group
await self.channel_layer.group_discard(self.room_group_name, self.channel_name)
# Receive message from WebSocket
async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json["message"]
username = text_data_json["username"] # username을 추출
# Send message to room group
await self.channel_layer.group_send(
self.room_group_name, {
"type": "chat_message",
"message": message,
"username": username # username을 함께 전송
}
)
# Receive message from room group
async def chat_message(self, event):
message = event["message"]
username = event["username"] # username을 추출
# Send message and username to WebSocket
await self.send(text_data=json.dumps({
"message": message,
"username": username # username도 함께 전송
}))
JavaScript
복사
routing.py
from django.urls import re_path
from . import customers
websocket_urlpatterns = [
re_path(r"ws/chat/(?P<room_name>\w+)/$", customers.ChatConsumer.as_asgi()),
]
Python
복사
urls.py
from django.contrib import admin
from django.urls import path, include
urlpatterns = [
path('admin/', admin.site.urls),
path('chat/', include('chat_app.urls'))
]
Python
복사
from django.urls import path
from . import views
urlpatterns = [
path("", views.index, name='index'),
path("<str:room_name>/<str:username>/", views.room, name='room')
]
Python
복사
views.py
from django.shortcuts import render
from django.contrib.auth import login
from django.contrib.auth.models import User
def index(request):
return render(request, 'index.html')
def room(request, room_name, username):
user, created = User.objects.get_or_create(username=username)
login(request, user)
# 채팅 방으로 리디렉션
return render(request, "room.html", {"room_name": room_name, "user": user})
Python
복사
asgi.py
"""
ASGI config for chat_project project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/4.2/howto/deployment/asgi/
"""
import os
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
# Initialize Django ASGI application early to ensure the AppRegistry
# is populated before importing code that may import ORM models.
django_asgi_app = get_asgi_application()
import chat_app.routing
application = ProtocolTypeRouter(
{
"http": django_asgi_app,
"websocket": AllowedHostsOriginValidator(
AuthMiddlewareStack(URLRouter(chat_app.routing.websocket_urlpatterns))
),
}
)
Python
복사
settings.py
INSTALLED_APPS = [
'daphne',
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'chat_app'
]
Python
복사
ASGI_APPLICATION = "chat_project.asgi.application"
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {
"hosts": [("127.0.0.1", 6379)],
},
},
}
Python
복사
index.html
<!-- chat/templates/chat/index.html -->
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8"/>
<title>Chat Rooms</title>
</head>
<body>
Enter your username:<br>
<input id="username-input" type="text" size="100"><br>
What chat room would you like to enter?<br>
<input id="room-name-input" type="text" size="100"><br>
<input id="room-name-submit" type="button" value="Enter">
<script>
document.querySelector('#username-input').focus();
document.querySelector('#username-input').onkeyup = function (e) {
if (e.keyCode === 13) { // enter, return
document.querySelector('#room-name-submit').click();
}
};
document.querySelector('#room-name-submit').onclick = function (e) {
var username = document.querySelector('#username-input').value;
var roomName = document.querySelector('#room-name-input').value;
window.location.pathname = '/chat/' + roomName + '/' + username + '/';
};
</script>
</body>
</html>
HTML
복사
room.html
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8" />
<title>Chat Room</title>
</head>
<body>
<textarea id="chat-log" cols="100" rows="20"></textarea><br>
<input id="chat-message-input" type="text" size="100"><br>
<input id="chat-message-submit" type="button" value="Send">
{{ room_name|json_script:"room-name" }}
<script>
const roomName = JSON.parse(document.getElementById('room-name').textContent);
const chatSocket = new WebSocket(
'ws://'
+ window.location.host
+ '/ws/chat/'
+ roomName
+ '/'
);
chatSocket.onmessage = function (e) {
const data = JSON.parse(e.data);
document.querySelector('#chat-log').value += (data.message + '\n');
};
chatSocket.onclose = function (e) {
console.error('Chat socket closed unexpectedly');
};
document.querySelector('#chat-message-input').focus();
document.querySelector('#chat-message-input').onkeyup = function (e) {
if (e.keyCode === 13) { // enter, return
document.querySelector('#chat-message-submit').click();
}
};
document.querySelector('#chat-message-submit').onclick = function (e) {
const messageInputDom = document.querySelector('#chat-message-input');
const message = messageInputDom.value;
chatSocket.send(JSON.stringify({
'message': message
}));
messageInputDom.value = '';
};
</script>
</body>
</html>
HTML
복사
tests.py
from django.test import TestCase
# Create your tests here.
from channels.testing import ChannelsLiveServerTestCase
from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys # 키보드의 키 명령 모음
from selenium.webdriver.support.wait import WebDriverWait
# Create your tests here.
class ChatTests(ChannelsLiveServerTestCase):
serve_static = True # emulate StaticLiveServerTestCase
@classmethod
def setUpClass(cls):
"""
Set up configuration
"""
super().setUpClass()
try:
# NOTE: Requires "chromedriver" binary to be installed in $PATH
cls.driver = webdriver.Chrome("chromedriver가 깔린 path 입력")
except:
super().tearDownClass()
raise
@classmethod
def tearDownClass(cls) -> None:
"""
Delete configuration
"""
cls.driver.quit()
super().tearDownClass()
def test_when_chat_message_posted_then_seen_by_everyone_in_same_room(self):
try:
self._enter_chat_room("room_1")
self._open_new_window()
self._enter_chat_room("room_1")
self._switch_to_window(0)
self._post_message("hello")
WebDriverWait(self.driver, 2).until(
lambda _: "hello" in self._chat_log_value,
"Message was not received by window 1 from window 1",
)
self._switch_to_window(1)
WebDriverWait(self.driver, 2).until(
lambda _: "hello" in self._chat_log_value,
"Message was not received by window 2 from window 1",
)
finally:
self._close_all_new_windows()
def test_when_chat_message_posted_then_not_seen_by_anyone_in_different_room(self):
try:
self._enter_chat_room("room_1")
self._open_new_window()
self._enter_chat_room("room_2")
self._switch_to_window(0)
self._post_message("hello")
WebDriverWait(self.driver, 2).until(
lambda _: "hello" in self._chat_log_value,
"Message was not received by window 1 from window 1",
)
self._switch_to_window(1)
self._post_message("world")
WebDriverWait(self.driver, 2).until(
lambda _: "world" in self._chat_log_value,
"Message was not received by window 2 from window 2",
)
self.assertTrue(
"hello" not in self._chat_log_value,
"Message was improperly received by window 2 from window 1",
)
finally:
self._close_all_new_windows()
# === Utility ===
def _enter_chat_room(self, room_name):
self.driver.get(self.live_server_url + "/chat/")
ActionChains(self.driver).send_keys(room_name, Keys.ENTER).perform()
WebDriverWait(self.driver, 2).until(
lambda _: room_name in self.driver.current_url
)
def _open_new_window(self):
self.driver.execute_script('window.open("about:blank", "_blank");')
self._switch_to_window(-1)
def _close_all_new_windows(self):
while len(self.driver.window_handles) > 1:
self._switch_to_window(-1)
self.driver.execute_script("window.close();")
if len(self.driver.window_handles) == 1:
self._switch_to_window(0)
def _switch_to_window(self, window_index):
self.driver.switch_to.window(self.driver.window_handles[window_index])
def _post_message(self, message):
ActionChains(self.driver).send_keys(message, Keys.ENTER).perform()
@property
def _chat_log_value(self):
return self.driver.find_element(
by=By.CSS_SELECTOR, value="#chat-log"
).get_property("value")
Python
복사