search

vaadin / flow

Vaadin Flow is a Java framework binding Vaadin web components to Java. This is part of Vaadin 10+.
Apache License 2.0
619 stars 167 forks source link

Double wrapped JSON message sent on websocket reconnect #6289

Open Artur- opened 5 years ago

Artur- commented 5 years ago

To reproduce, start project with push:

git clone https://github.com/Artur-/flow-lit.git
cd flow-lit
mvn jetty:run

Create a SSH tunnel

ssh localhost -L 1234:localhost:8080

Open app through tunnel

open http://localhost:1234

Kill SSH tunnel using exit + ctrl-c

Re-establish tunnel using 

ssh localhost -L 1234:localhost:8080

The first message delivered to the application contains an extra length| prefix, e.g.

302|298|for(;;);[{syncId: 5, clientId: 1, meta: {async: true},…}]

when it should be

298|for(;;);[{syncId: 5, clientId: 1, meta: {async: true},…}]
Artur- commented 5 years ago

I think this exact case is tested by https://github.com/vaadin/framework/blob/master/uitest/src/test/java/com/vaadin/tests/push/ReconnectWebsocketTest.java for FW8

denis-anisimov commented 4 years ago

I think this exact case is tested by https://github.com/vaadin/framework/blob/master/uitest/src/test/java/com/vaadin/tests/push/ReconnectWebsocketTest.java for FW8

Apparently it's not tested by this test since the test is migrated and works with Flow.

denis-anisimov commented 4 years ago

We should retest this with the latest versions especially considering the fact that we have passing ReconnectWebsocketTest.java test.

Artur- commented 4 years ago

Upgraded example to 14.2.0.alpha7. Still reproduces

denis-anisimov commented 4 years ago

Alright, thanks for the info.

Artur- commented 4 years ago

With Atmosphere trace logging you can see what is happening here:

  1. The browser receives push message 8

    Received push (websocket) message: for(;;);[{"syncId":8,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-4.342586269301345,"latitude":-33.80385646477082},"return $0.personUpdated($1)"]],"timings":[385,10]}]

  2. The websocket connection is disconnected

  3. The server creates and pushes message 9. Atmosphere happily writes it to the closed connection and does not notice anything is wrong. The message is lost forever. 

    [Atmosphere-Shared-101] TRACE org.atmosphere.websocket.WebSocket - WebSocket.write() for(;;);[{"syncId":9,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":1,"longitude":165.69204276168028,"latitude":-12.009253616658},"return $0.personUpdated($1)"]],"timings":[385,10]}]
[Atmosphere-Shared-101] TRACE org.atmosphere.cpr.AtmosphereResourceImpl - Invoking listener [com.vaadin.flow.server.communication.PushAtmosphereHandler$AtmosphereResourceListener@11f7e790] for 25dfa7c1-98d7-4559-840e-e2004f64168f
[Atmosphere-Shared-101] TRACE org.atmosphere.cpr.AtmosphereResourceEventListenerAdapter - AtmosphereResourceEventImpl{ isCancelled=false isClosedByClient=false isClosedByApplication=false isResumedOnTimeout=false throwable=null resource=null}
[Atmosphere-Shared-101] TRACE org.atmosphere.cpr.BroadcasterListenerAdapter - onComplete /*
[Thread-21] DEBUG com.vaadin.flow.server.communication.UidlWriter - * Creating response to client
[Atmosphere-Shared-0] TRACE org.atmosphere.cpr.DefaultBroadcaster - /* is about to broadcast Entry{message=for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}], type=RESOURCE, future=org.atmosphere.cpr.BroadcasterFuture@5a3c362d}
[Atmosphere-Shared-0] TRACE org.atmosphere.cache.UUIDBroadcasterCache - Adding for AtmosphereResource 25dfa7c1-98d7-4559-840e-e2004f64168f cached messages for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}]
[Atmosphere-Shared-0] TRACE org.atmosphere.cache.UUIDBroadcasterCache - Active clients {25dfa7c1-98d7-4559-840e-e2004f64168f=1584532781663, 89f371b2-c4cc-4424-b2ec-49d47556b70b=1584532761618}
[Atmosphere-Shared-0] TRACE org.atmosphere.cpr.BroadcasterListenerAdapter - onMessage for broadcaster /* for Entry{message=for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}], type=RESOURCE, future=org.atmosphere.cpr.BroadcasterFuture@5a3c362d}
[Atmosphere-Shared-0] TRACE org.atmosphere.cpr.DefaultBroadcaster - AtmosphereResource 25dfa7c1-98d7-4559-840e-e2004f64168f available for for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}]
[Atmosphere-Shared-0] TRACE org.atmosphere.cpr.DefaultBroadcaster - Broadcaster /* is about to queueWriteIO for AtmosphereResource 25dfa7c1-98d7-4559-840e-e2004f64168f

    This is one problem.

  4. The server creates and pushes message 10. Atmosphere notices that the connection does not work and caches the message. 

    [Atmosphere-Shared-102] TRACE org.atmosphere.cpr.DefaultBroadcaster - About to write to AtmosphereResource{
 uuid=25dfa7c1-98d7-4559-840e-e2004f64168f,
 transport=WEBSOCKET,
 isInScope=true,
 isResumed=false,
 isCancelled=false,
 isSuspended=true,
 broadcasters=/*,
 isClosedByClient=false,
 isClosedByApplication=false,
 action=Action{timeout=-1, type=SUSPEND}}
[Atmosphere-Shared-102] TRACE org.atmosphere.cache.UUIDBroadcasterCache - Removing for AtmosphereResource 25dfa7c1-98d7-4559-840e-e2004f64168f cached message for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}]
[Atmosphere-Shared-102] TRACE org.atmosphere.cpr.DefaultBroadcaster - /* is broadcasting to 25dfa7c1-98d7-4559-840e-e2004f64168f
[Atmosphere-Shared-102] TRACE org.atmosphere.websocket.WebSocket - WebSocket.write() for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}]
[Atmosphere-Shared-102] TRACE org.atmosphere.container.version.JSR356WebSocket - WebSocket AtmosphereResource{
 uuid=25dfa7c1-98d7-4559-840e-e2004f64168f,
 transport=WEBSOCKET,
 isInScope=true,
 isResumed=false,
 isCancelled=false,
 isSuspended=true,
 broadcasters=/*,
 isClosedByClient=false,
 isClosedByApplication=false,
 action=Action{timeout=-1, type=SUSPEND}} failed to write 197|for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}]
[Atmosphere-Shared-102] TRACE org.atmosphere.cache.UUIDBroadcasterCache - Adding for AtmosphereResource 25dfa7c1-98d7-4559-840e-e2004f64168f cached messages 197|for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}]

    Caching happens AFTER TrackMessageSizeInterceptor has been invoked and thus the cached message starts with the message length: 197|

  5. Channel is reconnected

  6. Atmosphere pushes the cached message but passes it AGAIN through TrackMessageSizeInterceptor, which adds an extra 201| so the message becomes 201|197|for(;;);[{"syncId":10,"clientId":0,"meta":{"async":true},"execute":[[[0,3],{"id":8,"longitude":-3.5774276745521707,"latitude":-32.5840109255126},"return $0.personUpdated($1)"]],"timings":[385,10]}].

  7. Client side fails because it gets a double wrapped message. This it the problem you first see