problem: in some cases WS may produce empty or broken messages that breaks the connection

solution: better handling for invalid JSON

(cherry picked from commit 1b6a995c4e746f0704643426fc20f7496fc09cef)

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt

@@ -21,6 +21,7 @@ import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.core.JsonToken
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.etherjar.rpc.RpcResponseError
+import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import java.io.IOException
 
@@ -58,15 +59,18 @@ abstract class ResponseParser<T> {
         } catch (e: JsonParseException) {
             log.warn("Failed to parse JSON from upstream: ${e.message}")
         }
-        if (state.isReady) {
-            return state
-        }
-        return Preparsed(
-            error = JsonRpcError(
-                RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
-                "Invalid JSON structure: never finalized"
+        return if (state.isReady) {
+            state
+        } else {
+            log.debug("Failed to parse `${StringUtils.abbreviateMiddle(String(json), "...", 200)}` JSON")
+            state.copy(
+                result = null,
+                error = JsonRpcError(
+                    RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
+                    "Invalid JSON structure: never finalized"
+                )
             )
-        )
+        }
     }
 
     open fun process(parser: JsonParser, json: ByteArray, field: String, state: Preparsed): Preparsed {

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt

@@ -44,6 +44,16 @@ class ResponseWSParser : ResponseParser<ResponseWSParser.WsResponse>() {
                 state.error
             )
         }
+        if (state.error != null) {
+            return WsResponse(
+                // we don't have any real option because it's just an invalid value and can be anything,
+                // so let's suppose its Type as RPC as a most likely scenario
+                Type.RPC,
+                state.id ?: JsonRpcResponse.Id.from(0),
+                null,
+                state.error
+            )
+        }
         throw IllegalStateException("State is not ready")
     }
 

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParserSpec.groovy

@@ -223,4 +223,27 @@ class ResponseRpcParserSpec extends Specification {
         !act.hasResult()
     }
 
+    def "Keep provided ID even if JSON is not full"() {
+        setup:
+        def json = '{"jsonrpc": "2.0", "id": 1}'
+        when:
+        def act = parser.parse(json.getBytes())
+        then:
+        act.id.asNumber() == 1
+        act.error != null
+        act.hasError()
+        !act.hasResult()
+    }
+
+    def "Keep provided ID even if JSON is broken"() {
+        setup:
+        def json = '{"jsonrpc": "2.0", "id": 101, "resu'
+        when:
+        def act = parser.parse(json.getBytes())
+        then:
+        act.id.asNumber() == 101
+        act.error != null
+        act.hasError()
+        !act.hasResult()
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy

@@ -103,4 +103,24 @@ class ResponseWSParserSpec extends Specification {
         act.error == null
         new String(act.value) == "null"
     }
+
+    def "Keep provided ID even if JSON is not full"() {
+        setup:
+        def json = '{"jsonrpc": "2.0", "id": 1}'
+        when:
+        def act = parser.parse(json.getBytes())
+        then:
+        act.id.asNumber() == 1
+        act.error != null
+    }
+
+    def "Keep provided ID even if JSON is broken"() {
+        setup:
+        def json = '{"jsonrpc": "2.0", "id": 101, "resu'
+        when:
+        def act = parser.parse(json.getBytes())
+        then:
+        act.id.asNumber() == 101
+        act.error != null
+    }
 }