如何将 ros2 消息从 websocket 服务器发送到龙卷风中连接的客户端

Question

我有一个 ros2 发布者脚本，可以从 ros2 节点发送自定义消息。我需要做的是让订阅者（这也是我的 websocket 服务器）收听 pulisher 发送的消息，然后将其转换为字典并将其作为 json 从 websocket 服务器发送到连接的 websocket 客户端。我已经检查了 rosbridge repo，但我无法让它工作。它没有足够的文档，而且我是 ros 新手。 我需要这样的东西：

import rclpy
import sys
from rclpy.node import Node
import tornado.ioloop

import tornado.httpserver
import tornado.web
import threading

from custom.msg import CustomMsg


from .convert import message_to_ordereddict





wss = []
class wsHandler(tornado.websocket.WebSocketHandler):
    def open(self):
        print 'Online'
        if self not in wss:
            wss.append(self)

    def on_close(self):
        print 'Offline'
        if self in wss:
            wss.remove(self)

def wsSend(message):
    for ws in wss:
        ws.write_message(message)


class MinimalSubscriber(Node):

    def __init__(self):
        super().__init__('minimal_subscriber')
        self.subscription = self.create_subscription(CustomMsg, 'topic', self.CustomMsg_callback, 10)
        self.subscription  # prevent unused variable warning

    def CustomMsg_callback(self, msg):
        ws_message = message_to_ordereddict(msg)
        wsSend(ws_message)






if __name__ == "__main__":
    http_server = tornado.httpserver.HTTPServer(tornado.web.Application(wsHandler))
    http_server.listen(8888)
    main_loop = tornado.ioloop.IOLoop.instance()
    # Start main loop
    main_loop.start()

所以MinimalSubscriber类中的回调函数，接收ros消息，将其转换为字典并发送给websocket客户端。我有点困惑如何让这两个线程（ros和websocket）相互通信。

Answer 1

所以我想我自己在处理线程时有点困惑。所以我改变了我的方法，并使用龙卷风周期性回调和 rclpy 的spin_once 函数作为回调函数使其工作。我会发布我的解决方案，因为它可能会帮助一些遇到同样问题的人。

import queue
import rclpy
from rclpy.node import Node
import tornado.ioloop

import tornado.httpserver
import tornado.web


from custom.msg import CustomMsg


from .convert import message_to_ordereddict





wss = []
class wsHandler(tornado.websocket.WebSocketHandler):
    @classmethod
    def route_urls(cls):
        return [(r'/',cls, {}),]

    def open(self):
        print 'Online'
        if self not in wss:
            wss.append(self)

    def on_close(self):
        print 'Offline'
        if self in wss:
            wss.remove(self)


def make_app():
    myWebHandler = wsHandler.route_urls()
    return tornado.web.Application(myWebHandler)


message_queue = queue.Queue
class MinimalSubscriber(Node):

    def __init__(self):
        super().__init__('minimal_subscriber')
        self.subscription = self.create_subscription(CustomMsg, 'topic', self.CustomMsg_callback, 10)
        self.subscription  # prevent unused variable warning

    def CustomMsg_callback(self, msg):
        msg_dict = message_to_ordereddict(msg)
        msg_queue.put(msg_dict)






if __name__ == "__main__":
    rclpy.init(args=args)

    minimal_subscriber = MinimalSubscriber()


    def send_ros_to_clients():
        rclpy.spin_once(minimal_subscriber)
        my_msg = msg_queue.get()

        for client in ws_clients:
            client.write_message(my_msg)


    app = make_app()
    server = tornado.httpserver.HTTPServer(app)
    server.listen(8888)
    tornado.ioloop.PeriodicCallback(send_ros_to_clients, 1).start()
    tornado.ioloop.IOLoop.current().start()



    minimal_subscriber.destroy_node()
    rclpy.shutdown()

我还在send_ros_to_clients 函数中实现了wsSend 函数。有人可能会说使用全局队列不是最佳实践，但我想不出另一种解决方案。如果对我的解决方案有任何建议或更正，我将不胜感激。