tornadoでのwhile loopの抜け方

Question

server.pyはPython Web Frameworkであるtornadoを使っています。

client.htmlのプルダウンメニューから選択した銘柄をserver.pyに送り、対応する銘柄をSQLから取得し、client.htmlに送り返すプログラムを作りたい。

SQLが随時更新されるので、def on_messageを(*1)のようにwhile Trueを使って1銘柄のときは随時更新していました。

しかし、(*2)の注意書きを目にしたので、どうやってプルダウンメニューで別の銘柄を選んだ時に前のloopを切断し、新しい銘柄での随時更新情報をclient.htmlに送るのか、どなたか分かる方はいらっしゃいませんでしょうか?
それとも、他の良い方法があるのならばご教授いただければ幸いです。

(*1)

import psycopg2, os, pandas as pd, time, json, random
import tornado.websocket
import tornado.ioloop

class RealTimeDataHandler(tornado.websocket.WebSocketHandler):
    def open(self):
        print ('Session Opened. IP:' + self.request.remote_ip)
        conn = psycopg2.connect(
            dbname='postgres',
            user='postgres',
            password=os.environ.get('PostgreSQL_PASS'),
            host='localhost',
        )
        self.cur = conn.cursor()

        self.ioloop = tornado.ioloop.IOLoop.instance()

    def get_data_from_sql(self):
        self.cur.execute("SELECT * FROM public.random_time_multi")
        m = self.cur.fetchall()
        column_names = [desc[0] for desc in self.cur.description]
        data = pd.DataFrame(m, columns=column_names)
        return data

    def on_message(self, message):
        column_name = json.loads(message)['symbol']
        while True:
            data = self.get_data_from_sql()
            message = json.dumps(data[column_name].to_dict())
            self.write_message(message)
            time.sleep(5)

    def on_close(self):
        print("Session closed")
        self.cur.close()
        self.ioloop.stop()
        
    def check_origin(self, origin):
        return True

def stop_tornado():
    tornado.ioloop.IOLoop.current().stop()

app = tornado.web.Application([(r"/ws/display", RealTimeDataHandler)])

if __name__ == '__main__':
    app.listen(8080)
    try:
        tornado.ioloop.IOLoop.current().start()
    except:
        print("KeyboardInterrupt received. Stopping Tornado.")
        stop_tornado()

(*2)
on_messageメソッド内で無限ループ(while Trueなど)が実行されている場合、新しいメッセージを受け取ることができません。無限ループが実行されている間は、on_messageメソッドがブロックされてしまい、新しいメッセージを処理するためのイベントループが実行されないためです。

Answer 1

クライアントからsymbolを受け取ったタイミングでtornado.ioloop.PeriodicCallbackを起動し、一定間隔でクライアントへwrite_messageすればよいのではないでしょうか？またはスレッドを生成してそこで無限ループするとか。

Comments

[LeonhardThird]

tornado.ioloop.PeriodicCallbackを使ってうまくいきました。ありがとうございました。

Answer 2

import psycopg2, os, pandas as pd, time, json, random
import tornado.websocket
import tornado.ioloop

class RealTimeDataHandler(tornado.websocket.WebSocketHandler):
    def open(self):
        """WebSocket接続が開かれた際の処理を実装
        """
        print ('Session Opened. IP:' + self.request.remote_ip)
        conn = psycopg2.connect(
            dbname='postgres',
            user='postgres',
            password=os.environ.get('PostgreSQL_PASS'),
            host='localhost',
        )
        self.cur = conn.cursor()

        self.column_name = None

        # PeriodicCallbackを使って定期的にデータを取得し、クライアントに送信する処理を設定
        self.periodic_callback = tornado.ioloop.PeriodicCallback(self.send_data_to_client, 5000)    # 5000は5000[ms]
        self.periodic_callback.start()

    def on_message(self, message):
        self.column_name = json.loads(message)['symbol']
        print('RecieveMessage of %s' % self.column_name)

    def get_data_from_sql(self):
        """SQLからデータを取得する。table 全体を返すようにする。

        Returns:
            pandas.DataFrame: 
        """
        self.cur.execute("SELECT * FROM public.random_time_multi")
        m = self.cur.fetchall()
        column_names = [desc[0] for desc in self.cur.description]
        data = pd.DataFrame(m, columns=column_names)
        return data

    def send_data_to_client(self):
        print('send data of %s' % self.column_name)
        if self.column_name:
            data = self.get_data_from_sql()
            message = json.dumps(data[['unix_time', self.column_name]].to_dict())
            self.write_message(message)

    def on_close(self):
        print("Session closed")
        self.cur.close()
        self.periodic_callback.stop()
        
    def check_origin(self, origin):
        return True

def stop_tornado():
    tornado.ioloop.IOLoop.current().stop()

app = tornado.web.Application([(r"/ws/display", RealTimeDataHandler)])

if __name__ == '__main__':
    app.listen(8080)
    try:
        tornado.ioloop.IOLoop.current().start()
    except:
        print("KeyboardInterrupt received. Stopping Tornado.")
        stop_tornado()