SocketIOとは？
リアルタイム通信に使われるWebSocketを高機能に実装した上位互換的な奴。HTTPと違い、1回コネクションを張るだけでサーバークライアント間で相互に非同期通信ができる低コストなもの。
簡単に言えば、リアルタイム相互通信できる技術。
特徴
- WebSocketが張れない場合、HTTPロングポールに切り替えてくれる
- 独自に「namespace」（名前空間）と「Room」（ルーム）機能があり、通信を分けることができる。
- 送信する際に、イベントを指定する。（デフォでmessage）
- WebSocket自体とは互換性はない
基本的な使い方
文法解説
スクリプトを見る前に、基本的な用語について。
room / namespace（ルームと名前空間）
上記も説明した通り、通信を分ける仕組み。namespaceは機能ごとに、roomはルームセッションごとに分けるのが良さそう。
また、namespaceを複数作る場合には、socketio.Namespaceを継承したクラスを作り、クラス内でイベントの関数を登録し、実行時にsio.register_namespace(作成したクラス)で名前空間として登録する。
sid
クライアントごとに割り当てられるセッションID。接続するたびに変わる。
sio.emit() / sio.send()
どちらもデータを送信する関数。違いはイベントをカスタム指定するかどうか。
Sendだと、イベントはデフォルトで「message」となる。
Emitだと、イベントは引数として指定する。
sio.send(data, to, room, ...)
sio.emit(event, data, to, room, ...)| 引数名 | 解説 |
|---|---|
| event | イベント名。sendだと”message”となる。 |
| data | 送信するデータ。文字列でも、辞書型でも、リストでも。多分なんでもよい気がする |
| to | 送信先。SIDまたはルーム名を指定する。指定しなければ全クライアントに送信される。 |
| room | toと同じ。 |
sio.enter_room()
特定のクライアントをルームに参加させる関数
sio.enter_room(sid, room, namespace)イベントの登録
基本的なイベントの登録はこんな感じ。
@sio.event
def イベント名(sid, rcvd_data):
sio.send(sid,f"recieved_data:{rcvd_data}")サーバー側のスクリプト
以下は基本的なSocketIOサーバーのコード。
import socketio
import eventlet
#準備的なところ
sio = socketio.Server(cors_allowed_origins="*")
app = socketio.WSGIApp(sio)
#クライアントが接続したときの処理
@sio.event
def connect(sid, environ):
print(f"クライアント {sid}が接続されました")
@sio.event
def disconnect(sid):
print(f"クライアント{sid}が切断されました。")
# "custom_event" というイベントを受信したときの処理
@sio.event
def custom_event(sid, data):
print(f"{sid}からデータを受信しました: {data}")
sio.emit("response_event", {"msg": "Hello, client!"}, to=sid)
# "message" というイベントを受信したときの処理
@sio.event
def message(sid, data):
print(f"クライアント {sid}からメッセージを受信！: {data}")
sio.send(f"送信された内容: {data}", to=sid)
if __name__ == "__main__":
#これでサーバーを起動
eventlet.wsgi.server(eventlet.listen(("localhost", 5000)), app)
これで基本的なサーバーとしては動作します。本格的に運営するならWSGIサーバーを別途用意するべきでしょう。
“connect”や”disconnect”、”connect_error”などのイベントで、接続時などのふるまいを指定出来るっぽい
クライアント側のスクリプト
import socketio
sio = socketio.Client()
@sio.event
def connect():
print("Connected to server")
@sio.event
def connect_error(data):
print("接続に失敗:", data) # 接続エラーの詳細を出力
@sio.event
def response(data):
print("サーバーから受信したデータ:", data)
sio.connect("http://localhost:5000/") # サーバーアドレスを確認
sio.send("Hello, server!")
sio.wait()
コメント