WebSocketService leaks subscriptions and requests after disconnect

[Shohou]

Subscription leak

WebSocketService has a field subscriptionForId which is map with subscription objects. The only way for subscription to be removed from this map is through dispose call. If websocket gets disconnected onWebSocketClose method is called and all subscriptions are notified through onError method, but they are not removed from map. When subsriptions receives onError call it cannot call dispose anymore as it is already in cancelled status and object in subscriptionForId field stays there forever. Now if I call WebSocketService.connect to reconnect websocket and do another subscription and repeat disconnect-reconnect-subscribe multiple times, each cycle create another object in subscriptionForId map and during disconnect all of them are notified in closeOutstandingSubscriptions() and as that BehaviorSubject is already cancelled it spams errors in console:

io.reactivex.exceptions.UndeliverableException: The exception could not be delivered to the consumer because it has already canceled/disposed the flow or the exception has nowhere to go to begin with. Further reading: https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling | java.io.IOException: Connection was closed
        at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
        at io.reactivex.subjects.BehaviorSubject.onError(BehaviorSubject.java:276)
        at org.web3j.protocol.websocket.WebSocketService.lambda$closeOutstandingSubscriptions$9(WebSocketService.java:569)
        at java.base/java.util.concurrent.ConcurrentHashMap$ValuesView.forEach(ConcurrentHashMap.java:4783)
        at org.web3j.protocol.websocket.WebSocketService.closeOutstandingSubscriptions(WebSocketService.java:565)
        at org.web3j.protocol.websocket.WebSocketService.onWebSocketClose(WebSocketService.java:549)
        at org.web3j.protocol.websocket.WebSocketService$1.onClose(WebSocketService.java:167)
        at java.base/java.util.Optional.ifPresent(Optional.java:178)
        at org.web3j.protocol.websocket.WebSocketClient.onClose(WebSocketClient.java:67)
        at org.java_websocket.client.WebSocketClient.onWebsocketClose(WebSocketClient.java:671)
        at org.java_websocket.WebSocketImpl.closeConnection(WebSocketImpl.java:562)
        at org.java_websocket.WebSocketImpl.closeConnection(WebSocketImpl.java:586)
        at org.java_websocket.client.WebSocketClient.run(WebSocketClient.java:496)
        at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: java.io.IOException: Connection was closed
        at org.web3j.protocol.websocket.WebSocketService.lambda$closeOutstandingSubscriptions$9(WebSocketService.java:568)
        ... 11 more
Exception in thread "WebSocketConnectReadThread-386" io.reactivex.exceptions.UndeliverableException: The exception could not be delivered to the consumer because it has already canceled/disposed the flow or the exception has nowhere to go to begin with. Further reading: https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling | java.io.IOException: Connection was closed
        at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
        at io.reactivex.subjects.BehaviorSubject.onError(BehaviorSubject.java:276)
        at org.web3j.protocol.websocket.WebSocketService.lambda$closeOutstandingSubscriptions$9(WebSocketService.java:569)
        at java.base/java.util.concurrent.ConcurrentHashMap$ValuesView.forEach(ConcurrentHashMap.java:4783)
        at org.web3j.protocol.websocket.WebSocketService.closeOutstandingSubscriptions(WebSocketService.java:565)
        at org.web3j.protocol.websocket.WebSocketService.onWebSocketClose(WebSocketService.java:549)
        at org.web3j.protocol.websocket.WebSocketService$1.onClose(WebSocketService.java:167)
        at java.base/java.util.Optional.ifPresent(Optional.java:178)
        at org.web3j.protocol.websocket.WebSocketClient.onClose(WebSocketClient.java:67)
        at org.java_websocket.client.WebSocketClient.onWebsocketClose(WebSocketClient.java:671)
        at org.java_websocket.WebSocketImpl.closeConnection(WebSocketImpl.java:562)
        at org.java_websocket.WebSocketImpl.closeConnection(WebSocketImpl.java:586)
        at org.java_websocket.client.WebSocketClient.run(WebSocketClient.java:496)
        at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: java.io.IOException: Connection was closed
        at org.web3j.protocol.websocket.WebSocketService.lambda$closeOutstandingSubscriptions$9(WebSocketService.java:568)
        ... 11 more

There is also requestForId map which I think behaves in a similar way, but as requests are not long living objects it's harder to reproduce and so I didn't check.

Steps To Reproduce

subscribe with WebSocketService by calling eth_subscribe for newHeads
disable your internet connection
reconnect
resubsribe
check exceptions and see how there is more and more exception spam after each disconnect-connect-subsribe

Expected behavior

I expect subscription objects to be cleared after websocket disconnects

Actual behavior

spam in console during disconnects, some lost memory I guess

Environment

• Web3j version - 4.12.2

Additional context

Ask me if more information is needed