Support STOMP receipts with the simple broker [SPR-17315]

spencercw opened SPR-17315 and commented

I'm trying to set up a project using WebSockets with the 'subscribe and snapshot' pattern (i.e., subscribing to a stream of updates and requesting a snapshot for initialisation).

From what I've been able to gather, the intended way to get the initial snapshot is to use @SubscribeMapping. I have put together a simple test class:

@Controller
public class GreetingController {
    private Logger logger = LoggerFactory.getLogger(GreetingController.class);

    private final SimpMessagingTemplate messagingTemplate;

    private AtomicInteger value = new AtomicInteger();

    @Autowired
    public GreetingController(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @SubscribeMapping("/greetings")
    public int init() {
        int x = value.get();
        logger.info("init " + x);
        return x;
    }

    @Scheduled(fixedRate = 1000)
    public void poll() {
        int x = value.incrementAndGet();
        logger.info("pushing " + x);
        messagingTemplate.convertAndSend("/topic/greetings", x);
    }
}

My client first subscribes to /topic/greetings and then /app/greetings. By doing it in this order, updates may be delivered to the client before the initial snapshot (which is fine), but the reverse order would create a brief period where an update could be generated after the snapshot but not delivered to the client.

You can see in the log below, both subscription messages are received in thread 'http-nio-8080-exec-9'. From there they are passed to the 'clientInboundChannel' thread pool. The topic subscription is handled by 'clientInboundChannel-5' while the app subscription is handled by 'clientInboundChannel-7'.

There is a race condition here because the topic subscription could be delayed long enough for an update to be generated after the app subscription has completed but before the topic subscription has completed, which would not be delivered to the client. I can artificially induce this by putting a breakpoint in SimpleBrokerMessageHandler.handleMessageInternal() and suspending only that thread.

As far as I can tell, it should be safe if I limit the thread pool to one thread, but that's obviously not ideal. Am I doing something wrong; is there some better way to do this?

2018-09-30 15:46:27.330 TRACE 32092 --- [nio-8080-exec-9] o.s.messaging.simp.stomp.StompDecoder    : Decoded SUBSCRIBE {id=[sub-0], destination=[/topic/greetings]} session=null
2018-09-30 15:46:27.330 TRACE 32092 --- [nio-8080-exec-9] o.s.w.s.m.StompSubProtocolHandler        : From client: SUBSCRIBE /topic/greetings id=sub-0 session=ok5b3h1f
2018-09-30 15:46:27.331 TRACE 32092 --- [nio-8080-exec-9] ConfigServletWebServerApplicationContext : Publishing event in org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@536dbea0: SessionSubscribeEvent[GenericMessage [payload=byte[0], headers={simpMessageType=SUBSCRIBE, stompCommand=SUBSCRIBE, nativeHeaders={id=[sub-0], destination=[/topic/greetings]}, simpSessionAttributes={}, simpHeartbeat=[J@63b1df68, simpSubscriptionId=sub-0, simpSessionId=ok5b3h1f, simpDestination=/topic/greetings}]]
2018-09-30 15:46:27.331 DEBUG 32092 --- [nboundChannel-5] o.s.m.s.b.SimpleBrokerMessageHandler     : Processing SUBSCRIBE /topic/greetings id=sub-0 session=ok5b3h1f
2018-09-30 15:46:27.331 DEBUG 32092 --- [nio-8080-exec-9] o.a.t.websocket.server.WsFrameServer     : WebSocket frame received. fin [true], rsv [4], OpCode [1], payload length [13]
2018-09-30 15:46:27.332 TRACE 32092 --- [nio-8080-exec-9] s.w.s.h.LoggingWebSocketHandlerDecorator : Handling TextMessage payload=[SUBSCRIBE
..], byteCount=48, last=true] in WebSocketServerSockJsSession[id=ok5b3h1f]
2018-09-30 15:46:27.332 TRACE 32092 --- [nio-8080-exec-9] o.s.messaging.simp.stomp.StompDecoder    : Decoded SUBSCRIBE {id=[sub-1], destination=[/app/greetings]} session=null
2018-09-30 15:46:27.332 TRACE 32092 --- [nio-8080-exec-9] o.s.w.s.m.StompSubProtocolHandler        : From client: SUBSCRIBE /app/greetings id=sub-1 session=ok5b3h1f
2018-09-30 15:46:27.332 TRACE 32092 --- [nio-8080-exec-9] ConfigServletWebServerApplicationContext : Publishing event in org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@536dbea0: SessionSubscribeEvent[GenericMessage [payload=byte[0], headers={simpMessageType=SUBSCRIBE, stompCommand=SUBSCRIBE, nativeHeaders={id=[sub-1], destination=[/app/greetings]}, simpSessionAttributes={}, simpHeartbeat=[J@6148f8ed, simpSubscriptionId=sub-1, simpSessionId=ok5b3h1f, simpDestination=/app/greetings}]]
2018-09-30 15:46:27.332 DEBUG 32092 --- [nio-8080-exec-9] o.a.tomcat.util.net.SocketWrapperBase    : Socket: [org.apache.tomcat.util.net.NioEndpoint$NioSocketWrapper@3fdb1e3c:org.apache.tomcat.util.net.NioChannel@311ce66f:java.nio.channels.SocketChannel[connected local=0:0:0:0:0:0:0:1/0:0:0:0:0:0:0:1:8080 remote=/0:0:0:0:0:0:0:1:32205]], Read from buffer: [0]
2018-09-30 15:46:27.332 DEBUG 32092 --- [nio-8080-exec-9] org.apache.tomcat.util.net.NioEndpoint   : Socket: [org.apache.tomcat.util.net.NioEndpoint$NioSocketWrapper@3fdb1e3c:org.apache.tomcat.util.net.NioChannel@311ce66f:java.nio.channels.SocketChannel[connected local=0:0:0:0:0:0:0:1/0:0:0:0:0:0:0:1:8080 remote=/0:0:0:0:0:0:0:1:32205]], Read direct from socket: [0]
2018-09-30 15:46:27.332 DEBUG 32092 --- [nboundChannel-7] .WebSocketAnnotationMethodMessageHandler : Searching methods to handle SUBSCRIBE /app/greetings id=sub-1 session=ok5b3h1f, lookupDestination='/greetings'
2018-09-30 15:46:27.334 TRACE 32092 --- [nboundChannel-7] .WebSocketAnnotationMethodMessageHandler : Found 1 handler methods: [{[/greetings],messageType=[SUBSCRIBE]}]
2018-09-30 15:46:27.334 DEBUG 32092 --- [nboundChannel-7] .WebSocketAnnotationMethodMessageHandler : Invoking hello.GreetingController#init[0 args]
2018-09-30 15:46:27.334 DEBUG 32092 --- [nboundChannel-7] o.s.b.f.s.DefaultListableBeanFactory     : Returning cached instance of singleton bean 'greetingController'
2018-09-30 15:46:27.334 TRACE 32092 --- [nboundChannel-7] o.s.m.h.i.InvocableHandlerMethod         : Invoking 'hello.GreetingController.init' with arguments []
2018-09-30 15:46:27.334  INFO 32092 --- [nboundChannel-7] hello.GreetingController                 : init 9
2018-09-30 15:46:27.334 TRACE 32092 --- [nboundChannel-7] o.s.m.h.i.InvocableHandlerMethod         : Method [hello.GreetingController.init] returned [9]
2018-09-30 15:46:27.336 TRACE 32092 --- [nboundChannel-7] HandlerMethodReturnValueHandlerComposite : Processing return value with org.springframework.messaging.simp.annotation.support.SubscriptionMethodReturnValueHandler@4cb10416
2018-09-30 15:46:27.336 DEBUG 32092 --- [nboundChannel-7] a.s.SubscriptionMethodReturnValueHandler : Reply to @SubscribeMapping: 9

Affects: 5.0.9

Start with SimpleBrokerMessageHandler.handleMessageInternal() and trace how messages from the clientInboundChannel are processed alongside @SubscribeMapping replies. Reproduce the subscription-order race described in the issue and inspect existing STOMP handling before defining the expected receipt or ordering behavior; the issue does not name tests or a specific completion criterion.