CODEWITHSUNDEEP
stompspring-rabbitrabbitmq-stomp
Ritesh Kaushik
Ritesh Kaushik

Reputation: 735

RabbitMQ Stomp not able to get continuous large messages

I am using Spring RabbitMQ with Stomp to build an event streaming application in which server puts large messages of around 100KB each continuously to a TopicExchange amp.topic with binding key as test, and the stomp client has subscribed to /topic/test.

Here is the code:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {

private static final int PORT = 1234;

/**
 * Methos to configure the Message Broker, in this case StompMessageBroker
 * @param config
 */
@Override
public void configureMessageBroker(MessageBrokerRegistry config) {

     config.enableStompBrokerRelay("/topic", "/queue/")
             .setRelayHost(host)
             .setRelayPort(PORT)
             .setClientLogin(username)
             .setClientPasscode(password)      
     config.setApplicationDestinationPrefixes("/app");
}

/**
 * Method to configure Stomp endpoints used by the client to connect using websockets.
 * @param registry
 */
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/test-websocket")
            .setAllowedOrigins("*").withSockJS();
  }
}

Exchange info: 

public static final String EXCHANGE_NAME="amq.topic";
// This call is done in a loop 100 times and puts 500KB of message each time to exchange.
 Rabbittemplate.convertAndSend(EXCHANGE_NAME, "test_user", "Loop Counter_"+i+" : " + str); // str is a string message of size 500KB.

My front end app uses:

sockJs = new SockJS('test-websocket');
stompClient = Stomp.over(sockJs);
stompClient.connect('username','password', (frame: any)=> {

var subscription_id =   this.stompClient1.subscribe('/topic/test_user',  (greeting: any) => {
      var message_id = greeting.headers['message-id'];
      stompClient.ack(message_id, subscription_id);
  },{ack: 'client'});
});

Q1. When i am sending larger messages (> 100KB each) to exchange in loop (100 times continuously), the stomp client is not able to get console message as 

Whoops! Lost connection to test-websocket

Any help is appreciated.

Upvotes: 1

Views: 857

Answers (1)

Ruslan Stelmachenko
Ruslan Stelmachenko

Reputation: 5420

This can be caused by exceeded buffer size for websocket queue. You sending too fast to your websocket. Try to increase websocket buffer size on spring-websocket transport configuration:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketMessageBrokerConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
        registry.setSendBufferSizeLimit(100 * 1024 * 1024); // 100MB
        registry.setSendTimeLimit(60_000);
    }

}

But before doing this, enable DEBUG level logging for org.springframework.web.socket.messaging logger, you should see the message like:

2019-04-08 19:16:40.171 DEBUG 15428 --- [clientOutboundChannel-5] o.s.w.s.m.SubProtocolWebSocketHandler    : Terminating 'WebSocketServerSockJsSession[id=f46296cfb74949b3af04b53739dc83ba]'

org.springframework.web.socket.handler.SessionLimitExceededException: Buffer size 524450 bytes for session 'f46296cfb74949b3af04b53739dc83ba' exceeds the allowed limit 524288
    at org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.limitExceeded(ConcurrentWebSocketSessionDecorator.java:227) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.checkSessionLimits(ConcurrentWebSocketSessionDecorator.java:197) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.sendMessage(ConcurrentWebSocketSessionDecorator.java:150) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.messaging.StompSubProtocolHandler.sendToClient(StompSubProtocolHandler.java:455) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.messaging.StompSubProtocolHandler.handleMessageToClient(StompSubProtocolHandler.java:442) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.messaging.SubProtocolWebSocketHandler.handleMessage(SubProtocolWebSocketHandler.java:355) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.support.ExecutorSubscribableChannel$SendTask.run(ExecutorSubscribableChannel.java:144) [spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_152]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_152]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_152]

This clearly describes the problem.

Upvotes: 2

Related Questions