使用Apache Camel向Websocket发送RabbitMQ消息

Question

如标题所述，我想使用RabbitMQ向Websocket发送消息。使用AngularJS前端，我想从Websocket读取RabbitMQ消息，并将它们打印到控制台。原则上，我的代码似乎是可行的，尽管我不知道如何获得消息的实际（字符串）内容？

后端：为了创建Websocket并执行路由，我使用Spring Boot和Apache Camel：http://camel.Apache.org/Spring-boot.html。

使用Camel路由很容易,我的完整Java代码如下所示：

@Component
public final class CamelRoute extends RouteBuilder {

    @Override
    public final void configure() throws Exception {
        from("rabbitmq:localhost/myExchange")
                .to("websocket://myEndpoint?sendToAll=true");
    }
}

运行SpringBoot应用程序不会出错,日志看起来也不错：

Route: route1 started and consuming from: rabbitmq://localhost/myExchange

RabbitMQ：RabbitMQ运行在我的本地主机上。当我转到http：//localhost:15672/时，我可以看到SpringBoot创建了myExchange并且可以使用UI中的Publish Message向它发送消息。例如，我指定“HelloWorld！”作为有效负载（没有任何报头等）并点击发送。

前端：为了从websocket读取消息，我基本上在AngularJS控制器中执行以下操作：

var socket = {
            start: function () {
                var location = "ws://localhost:9292/myEndpoint";
                this._ws = new WebSocket(location);
                this._ws.onmessage = this._onmessage;
                this._ws.onclose = this._onclose;
            },

            _onmessage: function (m) {
                //Log received RabbitMQ messages from the Websocket
                console.log(m);
            },

            _onclose: function (m) {
                if (this._ws) {
                    this._ws.close();
                }
            }
        };

原则上，接收信息似乎是可行的。当我转到我的网页(localhost:8080/)并打开Chrome中的控制台时，它会打印一个带有许多属性的“MessageEvent”对象。这些属性似乎都不包含实际的消息字符串，例如：“HelloWorld！”？

更新：我没有使用RabbitMQ传递消息，而是尝试用Apache Kafka执行同样的操作，并且成功了。以下是我所做的：

为了测试，我只需使用Docker-Image在本地Kafka代理上设置一个名为“测试”的主题。我的路由配置现在如下所示：

from("kafka:localhost:9092?topic=test&groupId=camelgroupid&autoOffsetReset=earliest&consumersCount=1")
            .to("websocket://dashboard?sendToAll=true");

要发送消息，我使用官方的Kafka客户端库。使用上面发布的相同的JS代码，我现在可以看到MessageObject.Data中的消息。

我仍然想知道为什么同样的方法在RabbitMQ中不起作用？有人有主意吗？

Answer 1

终于解决了我的问题。

问题出在客户端。RabbitMQ通过byte[]发送消息。从WebSocket读取消息时，需要正确解码字节[]。

以下是我更新的和工作的客户端代码：

     var socket = {
        start: function () {
            var location = "ws://localhost:9292/myEndpoint";
            this._ws = new WebSocket(location);
            // RabbitMQ transmits a byte[], therefore we need to change the binary type:
            this._ws.binaryType = 'arraybuffer';
            this._ws.onmessage = this._onmessage;
            this._ws.onclose = this._onclose;
        },

        _onmessage: function (m) {
            // Decode the byte[]:
            var messageString = new TextDecoder('UTF-8').decode(m.data);
            console.log(messageString);
        },

        _onclose: function (m) {
            if (this._ws) {
                this._ws.close();
            }
        }
    };