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

Question

我有一个使用STOMP的SockJSJava客户端。基于此,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);

如果消息是字符串类对象，则服务器端引发下一个转换错误：

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)

观察错误：

MessageConversion异常：无法从[[B]转换为[org.gasmart.websocket.消息]的GenericMessage[有效负载=字节[33]，标头={simpMessageType=MESSAGE，Stump命令=SEND，nativeHeaders={目的地=[/app/聊天频道]，内容类型=[text/普通；charset=UTF-8]，内容长度=[33]}，simpSessionAtinites={ip=/127.0.0.1:59629}，simpHeartbeat=[J@147a2bf，内容类型=文本/普通；charset=UTF-8，lookupDestation=/chat Channel，simpSessionId=79431feb8b5f4a9497492cc64f8965f，simpDestation=/app/chat Channel}]

请参见内容类型头是由SockJS Java客户端创建的。

如果我将stompClient配置为：

stompClient.setMessageConverter(new SimpleMessageConverter());

我发出一个信息：

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

其中消息是字节数组，服务器不抛出错误。但是我需要在发送之前将所有的字符串转换为字节数组。我想了解为什么服务器可以转换相同的JSON对象发送与SimpleMessageConzer和一个发送与StringMessageConzer.

我比较了从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}]

观察差异：内容类型=[text/普通；字符集=UTF-8]

Spring消息传递中有更多转换器：

您可以在此软件包中观察到SimpleMessageConzer和StringMessageConzer。但是有ByteArrayMessageConzer、CompositeMessageConzer、SmartMessageConzer等。这个转换器是怎么工作的？

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

为什么服务器可以转换JSON消息{“from”：“cliente1”，“text”：“KIKO”}是从JScript客户端发送的，而不是从Java客户端发送的？

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... 
}

然后以这种方式设置消息转换器

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

我没有看到在方法“afterConnected（StompSession会话，StompHeaders connectedHeaders）”中为您的客户机调用任何“subscribe”方法

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

然后使用前面定义的SimpleBean实例调用“send”方法：

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

这应该会有所帮助。