如何配置Spring Sockjs Java客户端消息转换器

Question

我有一个使用STOMP的SockJS Java客户端，基于这个https://github.com/rstoyanchev/spring-websocket-portfolio/blob/master/src/test/java/org/springframework/samples/portfolio/web/load/StompWebSocketLoadTestClient.java。

我的代码：

package mx.intercommunication.websocket.stompclient;
import org.springframework.messaging.converter.StringMessageConverter;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.sockjs.client.SockJsClient;
import org.springframework.web.socket.sockjs.client.Transport;
import org.springframework.web.socket.sockjs.client.WebSocketTransport;
public class StompClient {
    public StompClient(){
        cliente();
    }
    public void cliente() {
        String host = "localhost";

        int port = 8080;
        String stompUrl = "ws://{host}:{port}/Server/chat";

        StandardWebSocketClient webSocketClient = new StandardWebSocketClient();

        List<Transport> transports = new ArrayList<>(2);

        /*
         * The WebSocketTransport can be configured with:
         *      + StandardWebSocketClient in a JSR-356 runtime
         *      + JettyWebSocketClient using the Jetty 9+ native WebSocket API
         *      + Any implementation of Spring’s WebSocketClient
         */
        transports.add(new WebSocketTransport(webSocketClient));

        SockJsClient sockJsClient = new SockJsClient(transports);
        WebSocketStompClient stompClient = new WebSocketStompClient(sockJsClient);


        //Configure a scheduler to use for heartbeats and for receipt tracking. 
        //stompClient.setTaskScheduler(taskScheduler);
        //stompClient.setDefaultHeartbeat(new long[] {0, 0});

        /*
         * Set the MessageConverter to use to convert the payload of incoming and
         *  outgoing messages to and from byte[] based on object type and the "content-type" header. 
         * By default, SimpleMessageConverter is configured.
         */
        stompClient.setMessageConverter(new StringMessageConverter());


        ProducerStompSessionHandler producer = new ProducerStompSessionHandler();


         /*
         * Connect to the given WebSocket URL and notify the given
         *   org.springframework.messaging.simp.stomp.StompSessionHandler when connected on
         *   the STOMP level after the CONNECTED frame is received.
         *   
         *   Parameters:
         *          url the url to connect to
         *          handler the session handler
         *          uriVars URI variables to expand into the URL
         *   Returns:
         *          ListenableFuture for access to the session when ready for use
         * 
         */
        stompClient.connect(stompUrl, producer, host, port);


    }

    private static class ProducerStompSessionHandler extends StompSessionHandlerAdapter {

        //private final AtomicReference<Throwable> failure;

        private StompSession session;


        @Override
        public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
            this.session =  session;


            try {
                Thread.sleep(6000);
            } catch (InterruptedException e) {

                e.printStackTrace();
            }

            Json m = Json.object()
                .set("from", "cliente1")
                .set("text", "KIKO");

            String message = m.toString();

            //byte messageByteArr[] = message.getBytes();

            /*
             * Send a message to the specified destination, converting the payload to a
             *  byte[] with the help of a MessageConverter.
             *  
             * Parameters:
             *      destination:     the destination to send a message to
             *      payload:        the message payload
             * Returns:
             *      a Receiptable for tracking receipts
             */
            try {
                session.send("/app/chatchannel", message);
                //session.send("/app/chatchannel", messageByteArr);

                System.out.println("Sending message HELLO: "+message);

            } catch (Throwable t) {
                System.out.println("Message sending failed: "+t);

                //logger.error("Message sending failed at " + i, t);
                //failure.set(t);
            }


        }

        /**
         * This implementation returns String as the expected payload type
         * for STOMP ERROR frames.
         */
        @Override
        public Type getPayloadType(StompHeaders headers) {
            return String.class;
        }

        @Override
        public void handleFrame(StompHeaders headers, Object payload) {
            Exception ex = new Exception(headers.toString());
            System.out.println("STOMP ERROR frame: "+ex);
        }

        @Override
        public void handleException(StompSession session, StompCommand command, StompHeaders headers,
                byte[] payload, Throwable exception) {

            System.out.println("Handling exception: "+exception);
        }

        @Override
        public void handleTransportError(StompSession session, Throwable exception) {

            System.out.println("Transport error: "+exception);
        }


        @Override
        public String toString() {
            //return "ConsumerStompSessionHandler[messageCount=" + this.messageCount + "]";
            return "ConsumerStompSessionHandler to String....";
        }

    }
}

请检查当我将stompClient配置为以下内容时：

stompClient.setMessageConverter(new StringMessageConverter());

我发送了一条消息：

session.send("/app/chatchannel", message);

如果message是一个String类对象，服务器端会抛出下一个转换错误：

08:51:06,746 ERROR [org.springframework.web.socket.messaging.WebSocketAnnotationMethodMessageHandler] (clientInboundChannel-4) Unhandled exception from message handler method: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [org.gasmart.websocket.Message] for GenericMessage [payload=byte[33], headers={simpMessageType=MESSAGE, stompCommand=SEND, nativeHeaders={destination=[/app/chatchannel], content-type=[text/plain;charset=UTF-8], content-length=[33]}, simpSessionAttributes={ip=/127.0.0.1:59629}, simpHeartbeat=[J@147a2bf, contentType=text/plain;charset=UTF-8, lookupDestination=/chatchannel, simpSessionId=79431feb8b5f4a9497492ccc64f8965f, simpDestination=/app/chatchannel}]
    at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:124)
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:112)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:138)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:107)
    at org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler.handleMatch(AbstractMethodMessageHandler.java:502)
    at org.springframework.messaging.simp.annotation.support.SimpAnnotationMethodMessageHandler.handleMatch(SimpAnnotationMethodMessageHandler.java:497)
    at org.springframework.messaging.simp.annotation.support.SimpAnnotationMethodMessageHandler.handleMatch(SimpAnnotationMethodMessageHandler.java:87)
    at org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler.handleMessageInternal(AbstractMethodMessageHandler.java:461)
    at org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler.handleMessage(AbstractMethodMessageHandler.java:399)
    at org.springframework.messaging.support.ExecutorSubscribableChannel$SendTask.run(ExecutorSubscribableChannel.java:135)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

观察错误：

MessageConversionException：无法将[[B]转换为[org.gasmart.websocket.Message]，用于GenericMessage [payload = byte [33]，headers = {simpMessageType = MESSAGE，stompCommand = SEND，nativeHeaders = {destination = [/ app / chatchannel]， content-type = [text / plain; charset = UTF-8] ，content-length = [33]}，simpSessionAttributes = {ip = / 127.0.0.1：59629}，simpHeartbeat = [J @ 147a2bf，contentType = text / plain; charset = UTF-8，lookupDestination = /chatchannel，simpSessionId = 79431feb8b5f4a9497492ccc64f8965f，simpDestination = / app / chatchannel}]

注意到 content-type 标头由SockJS Java客户端创建。

如果我将 stompClient 配置为：

stompClient.setMessageConverter(new SimpleMessageConverter());

我发送了一条消息：

...
String message = m.toString();      
byte messageByteArr[] = message.getBytes();
session.send("/app/chatchannel", message);

当消息是一个字节数组时，服务器不会抛出错误。但在发送之前，我需要将所有的字符串转换为字节数组。

我想了解为什么服务器可以将使用SimpleMessageConverter和StringMessageConverter发送的相同JSON对象进行转换。

我比较了来自JavaScript客户端和Java客户端的JSON发送消息。两者都发送相同的JSON消息。然后我实现了一个ChannelInterceptor，在发送到相应的控制器之前打印接收到的消息：

public class WebSocketTraceChannelInterceptor extends ChannelInterceptorAdapter {
    @Override
    public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
         String payload = new String((byte[]) message.getPayload());
                                System.out.println("WebSocketTraceChannelInterceptor::afterSendCompletion!!    payload: "+payload);
    }   
}

相应的服务器控制器是：

@MessageMapping("/chatchannel")
    @SendTo("/topic/messages")
    public OutputMessage send(SimpMessageHeaderAccessor ha,@Payload Message message) throws Exception {
.....
}

Java输出：

message: GenericMessage [payload=byte[33], headers={simpMessageType=MESSAGE, stompCommand=SEND, nativeHeaders={destination=[/app/chatchannel], **content-type=[text/plain;charset=UTF-8]**, content-length=[33]}, simpSessionAttributes={ip=/127.0.0.1:59629}, simpHeartbeat=[J@147a2bf, contentType=text/plain;charset=UTF-8, simpSessionId=79431feb8b5f4a9497492ccc64f8965f, simpDestination=/app/chatchannel}]

在 JavaScript 中，使用 SockJS + Stomp.js：

message: GenericMessage [payload=byte[33], headers={simpMessageType=MESSAGE, stompCommand=SEND, nativeHeaders={destination=[/app/chatchannel], content-length=[33]}, simpSessionAttributes={ip=/127.0.0.1:57890}, simpHeartbeat=[J@d154f0, simpSessionId=innyvfme, simpDestination=/app/chatchannel}]

观察区别：content-type=[text/plain;charset=UTF-8] Spring Messaging 中有更多的转换器：

你可以在这个包中观察到SimpleMessageConverter和StringMessageConverter。但是还有ByteArrayMessageConverter、CompositeMessageConverter、SmartMessageConverter等。

这些转换器是如何工作的？

如果我想从Java客户端发送字符串，我需要哪种转换器？

为什么服务器可以将JSON消息{"from":"cliente1","text":"KIKO"}从JScript客户端发送，但不能从Java客户端发送？

- Sergio

Answer 1

尝试创建一个类似Bean的东西。

public class SimpleBean{
private String from;
private String text; 
public String getFrom(){return from;}
public String getText(){return text:}
//other methods like setters here... 
}

然后以这种方式设置messageConverter

            MappingJackson2MessageConverter m = new MappingJackson2MessageConverter();
            stompClient.setMessageConverter(m);

我在"afterConnected(StompSession session, StompHeaders connectedHeaders)"方法中没有看到您的客户端调用"subscribe"方法。

    session.subscribe(subscribeMethod, new StompFrameHandler() {
//.. overriding for public Type getPayloadType(StompHeaders headers) {} and
// public void handleFrame(StompHeaders headers, Object payload) {} here
});

然后调用"send"方法，使用刚刚定义的SimpleBean实例：

SimpleBean sb=new SimpleBean();
sb.setFrom("cliente1");
sb.setText("KIKO");
session.send("/app/chatchannel", sb);

这应该有所帮助。

- cristiano.salvati