from flask import Flask, render_template from flask_session import Session from flask_socketio import SocketIO, emit,send import datetime import time import json app = Flask(__name__) Session(app) socketio = SocketIO(app, manage_session=False) @app.route('/') def index(): return render_template('index.html') @app.route('/session', methods=['GET', 'POST']) def session_access(): pass @socketio.on('get-session') def get_session(): pass @socketio.on('set-session') def set_session(data): pass @socketio.on('my event') def handle_message(message): print('received message: ' + str(message)) while True: current_date = str(datetime.datetime.now()) dic = {'current_date': current_date} res_json = json.dumps(dic) time.sleep(1) emit('server_response',res_json) if __name__ == '__main__': socketio.run(app)
<html> <head> <link rel="stylesheet" href="https://unpkg.com/purecss@1.0.0/build/pure-min.css"> <meta name="viewport" content="width=device-width, initial-scale=1"> </head> <body> <div id="current_time"> </div> <script type="text/javascript" src="https://cdn.bootcss.com/jquery/3.4.1/jquery.min.js"></script> <script type="text/javascript" src="//cdnjs.cloudflare.com/ajax/libs/socket.io/1.3.6/socket.io.min.js"></script> <script type="text/javascript" charset="utf-8"> var socket = io.connect('http://' + document.domain + ':' + location.port); socket.on('connect', function() { socket.emit('my event', {data: '建立连接成功!'}); socket.on('server_response',function (data) { $('#current_time').text(data); }) }); </script> </body> </html>
import datetime import json from flask_socketio import Namespace, emit,SocketIO,disconnect from flask import Flask, render_template app = Flask(__name__) socketio = SocketIO(app) @app.route('/') def index(): return render_template('index.html') class MyCustomNamespace(Namespace): def on_connect(self): print('建立连接成功！') def on_disconnect(self): print('客户端断开连接！') def on_my_event(self,data): print(data) current_date = str(datetime.datetime.now()) dic = {'current_date': current_date} res_json = json.dumps(dic) emit('server_response', res_json) socketio.on_namespace(MyCustomNamespace('/test')) if __name__ == '__main__': socketio.run(app)