Django SSE 이용해서 알람 기능 구현 (1)
WebSocket과 HTTP, SSE WebSocket과 HTTP,SSE 모두 응용계층에서 사용되는 프로토콜이다. HTTP는 클라이언트-서버 간의 통신을 위한 단방향 통신이고 SSE는 서버에서 클라이언트 통신을 위한 일방향 통신다.
jongseoung.tistory.com
위 글과 이어지는 내용입니다.
이제 ASGI/Channels 설정을 했으니 본격적으로 기능을 구현할 차례이다.
앞서 작성했던 alarms 앱으로 들어가보자.
나는 아래의 코드를 보고 내가 필요한 부분을 변경해서 사용한 것이다. 커스텀마이징해서 사용할꺼면 아래 깃허브를 참고하는게 더 도움이 될 것이다.
GitHub - fanout/django-eventstream: Server-Sent Events for Django
Server-Sent Events for Django. Contribute to fanout/django-eventstream development by creating an account on GitHub.
github.com
urls.py
from django.urls import path, include
import django_eventstream
from . import views
urlpatterns = [
path('events/', views.AlarmList.as_view()),
path('events/<user_id>/', include(django_eventstream.urls), {
'format-channels': ['user-{user_id}']
}),
]
AlarmList 클래스를 'events/' 와 연결해준다.
'events/<user_id>/' 로 들어오면 새로운 채널을 만들거나 기존의 채널이 있으면 연결해준다. - send_event 할때 여기로 연결
views.py
from __future__ import unicode_literals
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework_simplejwt.tokens import AccessToken
from rest_framework.generics import ListAPIView
from django_eventstream.models import Event
from alarms.serializers import EventSerializers
class AlarmList(ListAPIView):
queryset = Event.objects.all()
serializer_class = EventSerializers
authentication_classes = [JWTAuthentication]
permission_classes = [IsAuthenticatedOrReadOnly]
def get_user_instance(self, request, *args, **kwargs):
auth_header = self.request.headers.get('Authorization')
access_token = auth_header.split(' ')[1]
decoded = AccessToken(access_token)
user_id = decoded['user_id']
return user_id
def get(self, request, *args, **kwargs):
user_id = self.get_user_instance(self.request)
try:
room = Event.objects.filter(channel='user-{}'.format(user_id)).order_by('-created')[:10]
self.queryset = room
except Event.DoesNotExist:
# 예외처리
return self.list(request, *args, **kwargs)
get_user_instance를 통해서 JWT토큰을 파싱하여 사용자 ID를 반환한다.
GET 요청으로 사용자 ID를 가져와 채널이름이 user-<user_id> 인 이벤트를 가지고 온다.
serializers.py
import json
from django_eventstream.models import Event
from rest_framework import serializers
class EventSerializers(serializers.ModelSerializer):
class Meta:
model = Event
fields = [
"channel",
"type",
"data",
"eid",
"created",
]
data = serializers.SerializerMethodField("get_data")
created = serializers.SerializerMethodField("get_created")
def get_data(self, obj):
try:
data = json.loads(obj.data)
except json.JSONDecodeError:
data = {}
return data
def get_created(self, obj):
date_format = "%Y-%m-%d %H:%M"
created = obj.created.strftime(date_format)
return created
데이터를 직렬화하는 부분이다.
send_event
from django_eventstream import send_event
send_message = f'{user.nickname}님이 댓글을 남겼습니다.\n"{content}"\n{post.title}'
send_event('user-{}'.format(post.author_id), 'message', send_message, json_encode=False)
send_event를 호출해서 채널에 메시지를 전달해줄 수 있다.
첫 인수는 전송할 채널, 두번째는 이벤트 유형, 세번째는 이벤트 데이터이다. 공식 문서에는 여기까지만 나와 았다.
https://github.com/fanout/django-eventstream/blob/master/django_eventstream/eventstream.py
위 링크로 들어가보면 send_event(channel, event_type, data, skip_user_ids=None, async_publish=True, json_encode=True) 로 되어 있어서 나는 json으로 인코딩을 안하고 진하도록 설정하였다.
후기
공식 문서만을 무작정 믿고 진행하는게 아니라 코드를 하나하나 분해해서 확인해보는게 조금 더 도움이 되었다.
비록 공식 문서가 잘못되어있는 부분이 많아서 헤매긴했지만 그로 인해 코드를 하나하나 해석하는 능력을 키울 수 있었다.
Reference
GitHub - fanout/django-eventstream: Server-Sent Events for Django
Server-Sent Events for Django. Contribute to fanout/django-eventstream development by creating an account on GitHub.
github.com
django-eventstream
Server-Sent Events for Django
pypi.org
'Django > DRF' 카테고리의 다른 글
Django postgresql 연동 및 설치 - 맥북 (0) | 2023.11.25 |
---|---|
Django 프로젝트 단계별 가이드라인 1 (0) | 2023.11.23 |
Django SSE 이용해서 알람 기능 구현 (1) (0) | 2023.10.18 |
DRF 구글 소셜로그인 TypeError: string indices must be integers (0) | 2023.10.12 |
DRF Paginator를 이용한 페이지네이션 (0) | 2023.10.11 |