Ufdjrw

I have this in my WebSocketHandler implementation:

@Override
public Mono<Void> handle(WebSocketSession session) 

 return session.send(
 session.receive()
 .flatMap(webSocketMessage -> 
 int id = Integer.parseInt(webSocketMessage.getPayloadAsText());

 Flux<EfficiencyData> flux = service.subscribeToEfficiencyData(id);
 var publisher = flux
 .<String>handle((o, sink) -> 
 try 
 sink.next(objectMapper.writeValueAsString(o));
 catch (JsonProcessingException e) 
 e.printStackTrace(); 
 
 )
 .map(session::textMessage);

 return publisher;
 )
 );

The Flux<EfficiencyData> is currently generated for testing in the service as follows:

public Flux<EfficiencyData> subscribeToEfficiencyData(long weavingLoomId) 
 return Flux.interval(Duration.ofSeconds(1))
 .map(aLong -> 
 longAdder.increment();
 return new EfficiencyData(new MachineSpeed(
 RotationSpeed.ofRpm(longAdder.intValue()),
 RotationSpeed.ofRpm(0),
 RotationSpeed.ofRpm(400)));
 ).publish().autoConnect();

I am using publish().autoConnect() to make it a hot stream. I created a unit test that starts 2 threads that do this on the returned Flux:

flux.log().handle((s, sink) -> 
 LOGGER.info("", s.getMachineSpeed().getCurrent());
 ).subscribe();

In this case, I see both threads printing out the same value every second.

However, when I open 2 browser tabs, I don't see the same values in both my web pages. The more websocket clients that connect, the more the values are apart (So each value from the original Flux seems to be sent to a different client, instead of sent to all of them).

Managed to fix this thanks to Brian Clozel on twitter.

The issue is that for each connecting websocket client, I call the service.subscribeToEfficiencyData(id) method, which returns a new Flux every time it is called. So of course, those independent Flux'es are not being shared between the different websocket clients.

To fix the issue, I create the Flux instance in the constructor or a PostConstruct method of my service so the subscribeToEfficiencyData returns the same Flux instance every time.

Note that .publish().autoConnect() on the Flux remains important, because without that websocket clients will again see different values!