我有一个使用 STOMP 的 SockJS Java 客户端。是基于这个https://github.com/rstoyanchev/spring-websocket-portfolio/blob/master/src/test/java/org/springframework/samples/portfolio/web/load/StompWebSocketLoadTestClient.java https://github.com/rstoyanchev/spring-websocket-portfolio/blob/master/src/test/java/org/springframework/samples/portfolio/web/load/StompWebSocketLoadTestClient.java.
My code:
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 as:
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：无法将 GenericMessage [payload=byte[33]、headers={simpMessageType=MESSAGE、stompCommand=SEND、nativeHeaders={destination=[/app/chatchannel] 从 [[B] 转换为 [org.gasmart.websocket.Message] ],内容类型=[文本/纯文本；字符集=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}]
看到内容类型标头由 SockJS Java 客户端创建。
如果我配置stompClient as:
stompClient.setMessageConverter(new SimpleMessageConverter());
我发送一条消息：
...
String message = m.toString();
byte messageByteArr[] = message.getBytes();
session.send("/app/chatchannel", message);
其中消息是字节数组，服务器不会抛出错误。
但我需要在发送之前将所有字符串转换为字节数组。
我想了解为什么服务器可以转换相同的 JSON 对象发送
SimpleMessageConverter 和一个使用 StringMessageConverter 发送的程序。
我对从 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}]
观察差异：内容类型=[文本/纯文本；字符集=UTF-8]
Spring消息传递中有更多转换器：
您可以观察此包中的 SimpleMessageConverter 和 StringMessageConverter。但还有ByteArrayMessageConverter、CompositeMessageConverter、SmartMessageConverter等。
这个转换器如何工作？
如果我想从 Java 客户端发送字符串，我需要什么转换器？
为什么服务器可以转换从 JScript 客户端发送的 JSON 消息 {"from":"cliente1","text":"KIKO"} 而从 Java 客户端发送时却不能？
