从 Tornado WebSocket 服务器主动发送消息
Sending messages actively from Tornado WebSocket server
我有一个 Python WebSocket 服务器。这可以在收到消息时return进行响应。
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 秒内没有收到消息,它会发送 "Are you sleeping?".
我想使用 WebSocket 制作聊天机器人。我使用 tornado 和 websocket 因为我只知道这些,如果你知道这方面更好的方法,我会很感兴趣。
您可以使用 tornado.ioloop.PeriodicCallback
添加时间表,并根据时间简单地检查是否有 activity,如下所示(我改编自您的代码和 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?
我有一个 Python WebSocket 服务器。这可以在收到消息时return进行响应。
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 秒内没有收到消息,它会发送 "Are you sleeping?".
我想使用 WebSocket 制作聊天机器人。我使用 tornado 和 websocket 因为我只知道这些,如果你知道这方面更好的方法,我会很感兴趣。
您可以使用 tornado.ioloop.PeriodicCallback
添加时间表,并根据时间简单地检查是否有 activity,如下所示(我改编自您的代码和 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?