FC2ブログ

記事一覧

django で Websocket

django で Websocketを試します。

(参考)
http://www.denzow.me/entry/2018/03/25/235952

環境は以下の通り
python : 3.7.2
django : 2.2.4

インストール

django で websocket を使用する際は、channels というライブラリを使用します。

$ pip install channels channels_redis


プロジェクト/アプリ作成

websocket を試すためのプロジェクトを作成します。

$ django-admin startproject websocket_proj
$ python manage.py startapp websocket_app


channelsの有効化

プロジェクトディレクトリ下のsettings.pyの以下太字箇所を修正します。

INSTALLED_APPS = [
...
'websocket_app',
'channels',

...
ASGI_APPLICATION = 'websocket_proj.routing.application'
]


View作成

プロジェクトディレクトリ下の urls.py の以下太字箇所を修正します。

from django.contrib import admin
from django.urls import path
from django.conf.urls import url

import websocket_app.views as websocket_view


urlpatterns = [
path('admin/', admin.site.urls),
url(r'^websocket/', websocket_view.WebsocketView.as_view())
]


続いてアプリディレクトリ直下のviews.pyの以下の内容を追記します。

from django.views.generic import TemplateView

class WebsocketView(TemplateView):
template_name = "websocket_view.html"

def get(self, request, *args, **kwargs):
context = super(WebsocketView, self).get_context_data(*kwargs)
return render(self.request, self.template_name, context)


続いて、アプリディレクトリ下にtemplatesディレクトリを作成し、websocket_view.htmlファイルを作成します。

<!DOCTYPE html>
<html lang='ja'>
<head>
<meta charset="utf-8">
<script type="text/javascript" src="https://ajax.googleapis.com/ajax/libs/jquery/1.9.1/jquery.min.js"></script>
<title>websocket test</title>
</head>
<body>
<textarea style="width:300px; height:100px;" id="message_area" readonly></textarea>
</body>
<script>
var ws = new WebSocket('ws://' + window.location.host + '/ws/');
ws.onmessage = function(e) {
var data = JSON.parse(e.data);
var msg = data['message'];
$("#message_area").append(msg + '\n');
};
ws.onclose = function(e) {
$("#message_area").append('websocket closed');
}
</script>
</html>


サーバー側の実装

アプリディレクトリにconsumers.pyを作成します。
Websocket接続されたら、1秒ごとにメッセージをpublishします。

from channels.generic.websocket import WebsocketConsumer
import json
import threading
import time

class TestConsumer(WebsocketConsumer):
def connect(self):
self.accept()
self.start_publish()

def disconnect(self, close_code):
self.stop_publish()

def start_publish(self):
self.publishing = True
self.t = threading.Thread(target=self.interval_publish)
self.t.start()

def stop_publish(self):
self.publishing = False
self.t.join()

def interval_publish(self):
while True:
if self.publishing == False:
break
self.send(text_data=json.dumps({
'message': 'test'
}))
time.sleep(1)


続いて、アプリディレクトリ下にrouting.pyを作成して以下の内容を記載します。

from django.urls import path
from . import consumers

websocket_urlpatterns = [
path('ws/', consumers.TestConsumer),
]


続いて、プロジェクトディレクトリ下にrouting.pyを作成して以下の内容を記載します。

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from websocket_app import routing as ws_routing

application = ProtocolTypeRouter({
'websocket': AuthMiddlewareStack(
URLRouter(
ws_routing.websocket_urlpatterns
)
),
})


実行すると、テキストエリアにメッセージが追加されていきます。

Websocket_1.png
スポンサーサイト



コメント

コメントの投稿

非公開コメント