背景
PythonでSocketIO通信を行う必要があったため、今後の開発の効率化のためにSocketIO通信を行うクラスのテンプレートを作成する。
記事の目的
PythonでSocketIOサーバーを作成する
python-socketio
python-socketioを使用したSocketIOサーバーを作成する
選定理由
python-socketioの選定理由は、下記の通りである。導入方法
python-socketioの導入手順は、下記の通りである。- pipを利用してインストールする
$ pip install requests eventlet python-socketioテンプレート
python-socketioの使用テンプレートは、下記の通りである。# socketioライブラリのインポート
import socketio
# ログ出力用ライブラリのインポート
from logging import getLogger, StreamHandler, DEBUG, INFO, ERROR
# 時間関係ライブラリのインポート
import time
# 終了シグナルをキャッチするライブラリのインポート
import signal
# eventletライブラリのインポート
import eventlet
# マルチスレッドライブラリをインポート
import threading
# ロガーのインスタンス作成
logger = getLogger(__name__)
stream_handler = StreamHandler()
# ログレベルを設定
log_level = INFO
logger.setLevel(log_level)
stream_handler.setLevel(log_level)
# ロガーにハンドラーをセット
logger.addHandler(stream_handler)
# SocketIOのテンプレート
class SocketIOServer:
# namespaceの設定用クラス
class NamespaceClass(socketio.Namespace):
def on_connect(self, sid, environ):
pass
def on_disconnect(self, sid):
pass
def on_message(self, sid, data):
pass
def on_client_to_server(self, sid, data):
pass
# 接続時に呼ばれるイベント
def on_connect(self, sid, environ):
logger.info('Connected to client (sid: %s, environ: %s)',\
str(sid), str(environ))
self.is_connect_ = True
# 切断時に呼ばれるイベント
def on_disconnect(self, sid):
logger.info('Disconnected from client (sid: %s)',str(sid))
self.is_connect_ = False
# サーバーからデータ送信(send)する
def send_data(self, data):
try:
self.sio_.send(data, namespace=self.namespace_)
except:
logger.error('Has not connected to client')
else:
logger.info('Send message %s (namespace="%s")',\
str(data), self.namespace_)
# 独自定義のイベント名「server_to_client」で、サーバーからデータ送信(emit)する
def emit_data(self, data):
try:
self.sio_.emit('server_to_client', data, namespace=self.namespace_)
except:
logger.error('Has not connected to client')
else:
logger.info('Emit message %s (namespace="%s")',\
str(data), self.namespace_)
def on_message(self, sid, data):
logger.info('Received message %s', str(data))
self.send_data({'test_ack': 'send_from_server'})
# クライアントからイベント名「on_client_to_server」でデータがemitされた時に呼ばれる
def on_client_to_server(self, sid, data):
logger.info('Received message %s', str(data))
self.emit_data({'test_ack': 'emit_from_server'})
# Namespaceクラス内の各イベントをオーバーライド
def overload_event(self):
self.Namespace.on_connect = self.on_connect
self.Namespace.on_disconnect = self.on_disconnect
self.Namespace.on_message = self.on_message
self.Namespace.on_client_to_server = self.on_client_to_server
# 初期化
def __init__(self,ip,port,namespace):
self.ip_ = ip
self.port_ = port
self.namespace_ = namespace
self.is_connect_ = False
self.sio_ = socketio.Server(async_mode='eventlet')
self.app_ = socketio.WSGIApp(self.sio_)
self.Namespace = self.NamespaceClass(self.namespace_)
self.overload_event()
self.sio_.register_namespace(self.Namespace)
# 接続確認
def isConnect(self):
return self.is_connect_
# 切断
def disconnect(self,sid):
try:
self.sio_.disconnect(sid,namespace=self.namespace_)
except:
logger.error('Cannot disconnect from Client(sid="%s", namespace="%s")',\
namespace=self.namespace_)
else:
self.is_connect_ = False
# 開始
def start(self):
try:
logger.info('Start listening(%s:%d, namespace="%s")',\
self.ip_,self.port_,self.namespace_)
eventlet.wsgi.server(eventlet.listen((self.ip_, self.port_)), self.app_)
except:
logger.error('Cannot start listening(%s:%d, namespace="%s")',\
self.ip_,self.port_,self.namespace_)
# メインの処理
def run(self):
p = threading.Thread(target=self.start)
p.setDaemon(True)
p.start()
if __name__ == '__main__':
# Ctrl + C (SIGINT) で終了
signal.signal(signal.SIGINT, signal.SIG_DFL)
# SocketIO Server インスタンスを生成
sio_server = SocketIOServer('localhost', 10000, '/test')
# SocketIO Server インスタンスを実行
sio_server.run()
# 接続待ち
while not sio_server.isConnect():
time.sleep(1)
# 切断待ち
while sio_server.isConnect():
time.sleep(1)
logger.info('Finish')
# 終了
exit()備考
- namespaceは、必ず「/」(スラッシュ)から始まる必要がある
- eventletを使用した場合、他の作業スレッドから、SocketIO内のEmitやSendを呼び出せない
- 上記テンプレートは、クライアントと通信テストができる
まとめ
- Python-SocketIOを利用したサーバーモジュールを作成した
参考文献
変更履歴
- 2019/08/17: 新規作成
0 件のコメント:
コメントを投稿