从Tornado WebSocket服务器主动发送消息
从Tornado WebSocket服务器主动发送消息
提问于 2017-08-18 19:02:26
我有一台Python WebSocket服务器。这可以在收到消息时返回响应。
import tornado.web
import tornado.websocket
import tornado.ioloop
class WebSocketHandler(tornado.websocket.WebSocketHandler):
def open(self):
print("New client connected")
self.write_message("You are connected")
def on_message(self, message):
self.write_message(u"You said: " + message)
def on_close(self):
print("Client disconnected")
def check_origin(self, origin):
return True
application = tornado.web.Application([
(r"/", WebSocketHandler),
])
if __name__ == "__main__":
application.listen(8888)
tornado.ioloop.IOLoop.instance().start()但是，在收到消息之前，它不能发送消息。如何主动发送消息？例如，它测量时间，如果它在10秒内没有收到消息，它会发送“你在睡觉吗？”。
我想用WebSocket制作聊天机器人。我使用tornado和websocket，因为我只知道这一点，如果你知道在这方面使用更好的方法，我会很感兴趣。
回答 1
Stack Overflow用户
回答已采纳
发布于 2017-08-18 20:53:59
您可以使用tornado.ioloop.PeriodicCallback添加一个时间表，并简单地根据时间检查是否有活动，如下所示(我根据您的代码和this answer进行了调整)：
import tornado.web
import tornado.websocket
import tornado.ioloop
import time
class WebSocketHandler(tornado.websocket.WebSocketHandler):
def simple_init(self):
self.last = time.time()
self.stop = False
def open(self):
self.simple_init()
print("New client connected")
self.write_message("You are connected")
self.loop = tornado.ioloop.PeriodicCallback(self.check_ten_seconds, 1000, io_loop=tornado.ioloop.IOLoop.instance())
self.loop.start()
def on_message(self, message):
self.write_message(u"You said: " + message)
self.last = time.time()
def on_close(self):
print("Client disconnected")
self.loop.stop()
def check_origin(self, origin):
return True
def check_ten_seconds(self):
print("Just checking")
if (time.time() - self.last > 10):
self.write_message("You sleeping mate?")
self.last = time.time()现在，客户端在空闲和写入之间交替(改编自here)：
class Client(object):
def __init__(self, url, timeout):
self.url = url
self.timeout = timeout
self.ioloop = IOLoop.instance()
self.ws = None
self.connect()
PeriodicCallback(self.keep_alive, 20000, io_loop=self.ioloop).start()
self.ioloop.start()
@gen.coroutine
def connect(self):
print("trying to connect")
try:
self.ws = yield websocket_connect(self.url)
except Exception as e:
print("connection error")
else:
print("connected")
self.run()
@gen.coroutine
def run(self):
once = False
while True:
msg = yield self.ws.read_message()
print(msg)
if once:
time.sleep(11)
once = False
else:
time.sleep(1)
once = True
self.ws.write_message("Hello matey")
if msg is None:
print("connection closed")
self.ws = None
break
def keep_alive(self):
if self.ws is None:
self.connect()
else:
self.ws.write_message("keep alive")我们得到预期的输出：
trying to connect
connected
You are connected
You said: Hello matey
You sleeping mate?
You said: Hello matey
You said: Hello matey
You sleeping mate?页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/45755114
复制相关文章
相似问题
腾讯云开发者
