我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用flask_socketio.emit()。
def __init__(self, socketio, hashtag=None, filter_by_hashtag=False): """ Args: socketio (SocketIO): Used for emitting the tweet data to the client filter_by_hashtag (bool): When filtering the stream with a bounding box, an extra filter needs to be performed to emit only the tweets with the desired hashtag hashtag (string): If 'filter_by_hashtag' is specified, this is required to assist the internal filter """ super().__init__() self.socketio = socketio self.filter_by_hashtag = filter_by_hashtag self.hashtag = hashtag
def logSucess(ctp): emit('rsp_login', {"ErrorID": 0, "ErrorMsg": "??"}) emit('rsp_account', ctp.t.Account.__dict__) rtn = [] for p in ctp.t.DicInstrument: rtn.append(ctp.t.DicInstrument[p].__dict__) emit('rsp_instrument', rtn) rtn = [] for p in ctp.t.DicPositionField: rtn.append(ctp.t.DicPositionField[p].__dict__) emit('rsp_position', rtn) for p in ctp.t.DicOrderField: emit('rtn_order', ctp.t.DicOrderField[p].__dict__) for p in ctp.t.DicTradeField: emit('rtn_trade', ctp.t.DicTradeField[p].__dict__)
def emitQueueInfo(queue): """Emit information about queue. Specifically, this will emit the following information: (1) time to resolution (2) number of requests """ from quupod.models import Inquiry emit( 'update student page', { 'ttr': queue.ttr(), 'nor': Inquiry.query.filter_by( queue_id=queue.id, status='unresolved').count() }, broadcast=True, namespace=QUEUE_SOCKET_FORMAT % queue.id)
def general_info(): info = {'hostname' : NodeDefender.hostname, 'release' : NodeDefender.release, 'date_loaded' : str(NodeDefender.date_loaded), 'run_mode' : NodeDefender.config.general.run_mode()} emit('general', info) return True
def logging(): info = {'enabled' : NodeDefender.config.logging.enabled(), 'type' : NodeDefender.config.logging.type(), 'name' : NodeDefender.config.logging.name(), 'server' : NodeDefender.config.logging.server(), 'port' : NodeDefender.config.logging.port()} return emit('logging', info)
def database(): info = {'enabled' : NodeDefender.config.database.enabled(), 'engine' : NodeDefender.config.database.engine(), 'server' : NodeDefender.config.database.server(), 'port' : NodeDefender.config.database.port(), 'database' : NodeDefender.config.database.db(), 'file' : NodeDefender.config.database.file()} return emit('database', info)
def celery(): info = {'enabled' : NodeDefender.config.celery.enabled(), 'broker' : NodeDefender.config.celery.broker(), 'server' : NodeDefender.config.celery.server(), 'port' : NodeDefender.config.celery.port(), 'database' : NodeDefender.config.celery.database()} return emit('celery', info)
def mail(): info = {'enabled' : NodeDefender.config.mail.enabled(), 'server' : NodeDefender.config.mail.server(), 'port' : NodeDefender.config.mail.port(), 'tls' : NodeDefender.config.mail.tls(), 'ssl' : NodeDefender.config.mail.ssl(), 'username' : NodeDefender.config.mail.username(), 'password' : NodeDefender.config.mail.password()} return emit('mail', info)
def list(group): emit('list', NodeDefender.db.mqtt.list(group)) return True
def info(host, port): mqtt = NodeDefender.db.mqtt.get_redis(host, port) sql_mqtt = NodeDefender.db.mqtt.get_sql(host, port) mqtt['icpes'] = [icpe.mac_address for icpe in sql_mqtt.icpes] mqtt['groups'] = [group.name for group in sql_mqtt.groups] emit('mqttInfo', mqtt) return True
def update_mqtt(current_host, new_host): mqtt = NodeDefender.db.mqtt.get_sql(current_host['host'], current_host['port']) mqtt.host = new_host['host'] mqtt.port = new_host['port'] NodeDefender.db.mqtt.save_sql(mqtt) emit('reload', namespace='/general') return True
def delete_mqtt(host, port): NodeDefender.db.mqtt.delete(host, port) emit('reload', namespace='/general') return True
def desktops_stream(): with app.app_context(): for c in r.table('domains').get_all(current_user.username, index='user').merge({"table": "domains"}).changes(include_initial=False).run(db.conn): if c['new_val'] is not None: print('new event for user '+current_user.username+' desktop:'+c['new_val']['id']) emit('status_desktop',c['new_val']) #~ @app.route('/stream/desktops') #~ @login_required #~ def sse_request(): #~ return Response(event_stream(current_user.username), mimetype='text/event-stream') #~ def event_stream(username): #~ with app.app_context(): #~ #dom_started=set(['_jvinolas_asdf']) #~ dom_started=[] #~ for c in r.table('domains').get_all(username, index='user').merge({"table": "domains"}).changes(include_initial=False).union( #~ r.table('domains_status').get_all(r.args(dom_started)).merge({"table": "domains_status"}).changes(include_initial=False)).run(db.conn): #~ #get_all(username, index='user').filter({'kind': 'desktop'}) #~ #r.table('domains_status').filter(lambda domain: r.table('domains').get_all(username, index='user').filter({'status':'Started'})) #~ #r.table('domains_status').get_all(r.args(list(dom_started)), index='name') #~ #r.table('domains_status').filter(lambda domain: r.table('domains').get_all(username, index='user').filter({'status':'Started'})) #~ print(list(r.table('domains_status').get_all(r.args(list(dom_started)), index='name').merge({"table": "domains_status"}).run(db.conn))) #~ if (c['new_val'] is not None and c['new_val']['table'] == 'domains') or (c['old_val'] is not None and c['old_val']['table'] == 'domains'): #~ if c['new_val'] is None: #~ yield 'retry: 5000\nevent: %s\nid: %d\ndata: %s\n\n' % ('Deleted',time.time(),json.dumps(c['old_val'])) #~ continue #~ if c['old_val'] is None: #~ yield 'retry: 5000\nevent: %s\nid: %d\ndata: %s\n\n' % ('New',time.time(),json.dumps(app.isardapi.f.flatten_dict(c['new_val']))) #~ continue #~ #if 'detail' not in c['new_val']: c['new_val']['detail']='' #~ if c['old_val']['status']=='Starting' and c['new_val']['status']=='Started': #~ dom_started.append(c['new_val']['id']) #~ yield 'retry: 2000\nevent: %s\nid: %d\ndata: %s\n\n' % ('Status',time.time(),json.dumps(app.isardapi.f.flatten_dict(c['new_val']))) #~ else: #~ print('THIS IS A DOMAIN STATUS') #~ print(c['new_val']['name'],c['new_val']['status']['cpu_usage'])
def messenger(): while True: sleep(5) app.logger.debug("Inside Messenger Iteration") msgs = poll_topics() ads = dict() for topic in msgs.keys(): all_msgs = msgs[topic] ads[topic] = dict() for msg in all_msgs: offset = msg[2] ad = msg[-1] ads[topic][offset] = ad # offset = msg.offset # ad = msg.value # try: # ads[topic][offset] = ad # except KeyError: # ads[topic] = dict # ads[topic][offset] = ad for sid in client_dict.keys(): interest_list = client_dict[sid].keys() for interest in interest_list: regions = client_dict[sid][interest].keys() for region in regions: if client_dict[sid][interest][region][0]: offset = client_dict[sid][interest][region][1] while True: try: topic = interest+'_'+region app.logger.debug('Emitting to server with topic ' + topic) ad = ads[topic][offset] # if len(ad) == 0: # app.logger.debug('ad length was 0') # continue socketio.emit('server-message', {'topic': topic, 'ad': ad}, room=sid, callback=ack) offset += 1 except KeyError: app.logger.debug("Key Error Exception happened") break
def on_data(self, data): json_data = json.loads(data) # skip this tweet if it doesn't have the desired attributes if not all([key in json_data for key in ["text", "created_at"]]): return if self.filter_by_hashtag: if 'entities' in json_data and json_data['entities']['hashtags']: entities = json_data['entities'] hashtags = map(lambda x: x['text'], entities['hashtags']) if not self.hashtag in hashtags: return text = json_data['text'] sentiment = getSentiment(text) retweeted = True if "RT @" in text else False if retweeted: text = text.replace("RT @", "") self.socketio.emit("response", { "text": text, "created_at": json_data["created_at"], "sentiment": sentiment, "retweeted": retweeted }, namespace="/streaming")
def io_emit(self, rsp_type_str, data): socketio.emit(rsp_type_str, data, namespace='/ctp', room=self.Investor+self.PassWord)
def q_Tick(self, field=Tick): """""" socketio.emit('rtn_tick', field.__dict__, namespace='/ctp', room=field.Instrument)
def ctp_connect(): # client_sid = socketio.__dict__['server'].__dict__['manager'].__dict__['rooms']['/ctp'].keys() # ids = [id for id in client_sid if id not in ctps] # logger.info(ids) #??? new_sid = flask.request.sid logger.info('ctp connect:' + new_sid) #??emit?????????? emit('sid', new_sid)
def sub_inst(data): sid = flask.request.sid ctp = ctps[sid_ctpid[sid]] inst = data['instrument'] ctp.q.ReqSubscribeMarketData(inst) join_room(inst) tick = ctp.q.DicTick.get(inst) if tick: socketio.emit('rtn_tick', tick.__dict__, namespace='/ctp', room=inst)
def emitQueuePositions(inquiry): """Emit new queue positions.""" from quupod.models import Inquiry unresolved = (Inquiry.query.filter_by( queue_id=inquiry.queue_id, status='unresolved') .order_by(asc(Inquiry.created_at)) .all()) indices = list(enumerate([i.id for i in unresolved], start=1)) emit( 'update position', {'positions': json.dumps(indices)}, broadcast=True, namespace=QUEUE_SOCKET_FORMAT % inquiry.queue_id)
def receive_message(message): """"Receive a SocketIO request""" (request, response) = _process_message(message) print 'Socket:\t{} to `/{}` resulted in {}: {}'.format( request.method, request.path, response.status['code'], response.status['description'] ) sys.stdout.flush() flask_socketio.emit('response', response.to_JSON(), json=True)
def post_resources(Model, data): with orm.db_session: setattr(g, 'in_celery', True) # this is just a virtual test # emit("user created!") print "get data: ", data send(data) # where can I put my celery task:(
def desktops_template(): msg=True if request.method == 'POST': if float(app.isardapi.get_user_quotas(current_user.username)['tqp']) >= 100: flash('Quota for creating new templates is full','danger') return redirect(url_for('desktops')) #~ # if app.isardapi.is_domain_id_unique original=app.isardapi.get_domain(request.form['id']) domain_dict=app.isardapi.f.unflatten_dict(original) res=app.isardapi.new_tmpl_from_domain(current_user.username, request.form['name'], request.form['description'], request.form['kind'], domain_dict) if res is True: flash('Template creation queued, wait to complete','success') return redirect(url_for('desktops')) else: flash('Could not create template now','danger') return redirect(url_for('desktops')) #~ @socketio.on('connect', namespace='/test') #~ def test_connect(): #~ print(current_user.username) #~ global thread #~ if thread is None: #~ thread = socketio.start_background_task(target=background_thread) #~ emit('my_response', {'data': 'Connected', 'count': 0}) #~ def background_thread(): #~ """Example of how to send server generated events to clients.""" #~ count = 0 #~ while True: #~ socketio.sleep(10) #~ count += 1 #~ socketio.emit('my_response', #~ {'data': 'Server generated event', 'count': count}, #~ namespace='/test') #~ @socketio.on('connect') #~ def desktops_connect_handler(): #~ if current_user.is_authenticated: #~ log.info('authenticated: '+current_user.username) #~ with app.app_context(): #~ for c in r.table('domains').get_all(current_user.username, index='user').merge({"table": "domains"}).changes(include_initial=False).run(db.conn): #~ if c['new_val'] is not None: #~ log.info('new event for user '+current_user.username+' desktop:'+c['new_val']['id']) #~ emit('status_desktop',c['new_val']) #~ else: #~ log.info('not authenticated') #~ return False # not allowed here #~ @socketio.on('disconnect') #~ def disconnect(): #~ print("%s disconnected" % (current_user.username))
def handle_disconnect(): """ Disconnect handler that removes the client from the room list :return: """ app.logger.debug("Client disconnected: " + str(request.sid)) with lock: sid = str(request.sid) try: client_dict.pop(sid) except KeyError: pass # @socketio.on('client-message') # def handle_client_message(msg): # """ # Custom event name example # :param msg: # :return: # """ # # emit message on server-message channel and set a callback for handling delivery # emit('server-message', ('lele', 'theeke'), callback=ack) # app.logger.debug('Client message received: ' + msg) # # return acknowledgement: can be processed as args i client callback # return 'got it', 'carry on' # def ack(): # """ # Callback for acknowledging whether # client received the message or not # :return: # """ # print "ack" # def messenger(): # """ # Simple stupid test # :return: # """ # for i in range(0,100): # if len(client_dict) > 0: # idx = i % len(client_dict) # app.logger.info('Sending message to client in room: ') # socketio.emit('server-message', {'data': 'Message sent at time: ' + str(i)}) # app.logger.info('Messenger in iteration: ' + str(i)) # sleep(5) #
def new_server(viewer_queue, stop_page, port, secret_key): # create server app = Flask(__name__, static_url_path='/static') app.config['SECRET_KEY'] = secret_key # must be 'threading' for broadcast emitting socketio = SocketIO(app, async_mode='threading', logger=False, engineio_logger=False) # rooting @app.route('/') def __index(): logger.info('Render viewer page') return render_template('index.html', script="index.js") if stop_page: @app.route('/stop') def __stop(): socketio.stop() logger.info('Server stop request') return 'This server is stopped' @socketio.on('connect', namespace=IO_NAMESPACE) def __on_viewer_connect(): logger.info('New viewer connection is established') @socketio.on('disconnect', namespace=IO_NAMESPACE) def __on_viewer_disconnect(): logger.info('Viewer connection is closed') @socketio.on('update', namespace=IO_NAMESPACE) def __on_update(): logger.info('Image updating request is received') # get all of current data emit_data = buffering_thread.get_data_all() # emit all logger.debug('Emit for update all') emit('update', emit_data, namespace=IO_NAMESPACE) def update_event(tab, name, data): emit_data = [[tab, name, data]] # single data # broadcast emit logger.debug('Broadcast emit for update (tab: %s, name: %s)' % (str(tab), str(name))) socketio.emit('update', emit_data, namespace=IO_NAMESPACE) # create image updating thread if viewer_queue: logger.info('Start image buffering thread') buffering_thread = ImageBufferingThread(viewer_queue) buffering_thread.daemon = True buffering_thread.start() buffering_thread.register_update_event_func(update_event) # start server logger.info('Start server on port %d' % port) socketio.run(app, host='0.0.0.0', port=port, debug=False, log_output=False) logger.info('Stop server on port %d' % port)
