problem: doesn't pass error details from upstream

fix: #42
2020-08-10 22:18:54 -04:00
parent 2abe3023b6
commit b5daceeff5
30 changed files with 325 additions and 103 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt
@@ -86,7 +86,7 @@ class ProxyServer(
                .addAllItems(call.items)
                .build()
        val jsons = nativeCall
-                .nativeCall(Mono.just(request))
+                .nativeCallResult(Mono.just(request))
                .transform(writeRpcJson.toJsons(call))
        return if (call.type == ProxyCall.RpcType.SINGLE) {
            jsons.next()
--- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt
@@ -42,7 +42,7 @@ open class WriteRpcJson() {
    /**
     * Convert Dshackle protobuf based responses to JSON RPC formatted as strings
     */
-    open fun toJsons(call: ProxyCall): Function<Flux<BlockchainOuterClass.NativeCallReplyItem>, Flux<String>> {
+    open fun toJsons(call: ProxyCall): Function<Flux<NativeCall.CallResult>, Flux<String>> {
        return Function { flux ->
            flux
                    .flatMap { response ->
@@ -70,19 +70,24 @@ open class WriteRpcJson() {
        }
    }

-    open fun toJson(call: ProxyCall, response: BlockchainOuterClass.NativeCallReplyItem): String? {
-        val id = call.ids[response.id] ?: return null;
-        val json = if (response.succeed) {
-            JsonRpcResponse.ok(response.payload.toByteArray(), JsonRpcResponse.Id.from(id))
+    open fun toJson(call: ProxyCall, response: NativeCall.CallResult): String? {
+        val id = call.ids[response.id]?.let {
+            JsonRpcResponse.Id.from(it)
+        } ?: return null;
+        val json = if (response.isError()) {
+            val error = response.error!!
+            error.upstreamError?.let { upstreamError ->
+                JsonRpcResponse.error(upstreamError, id)
+            } ?: JsonRpcResponse.error(-32002, error.message, id)
        } else {
-            JsonRpcResponse.error(-32002, response.errorMessage, JsonRpcResponse.Id.from(id))
+            JsonRpcResponse.ok(response.result!!, id)
        }
        return objectMapper.writeValueAsString(json)
    }

    fun toJson(call: ProxyCall, error: NativeCall.CallFailure): String? {
        val id = call.ids[error.id] ?: return null;
-        val json = JsonRpcResponse.error(-32002, error.reason.message ?: "", JsonRpcResponse.Id.from(id))
+        val json = JsonRpcResponse.error(-32003, error.reason.message ?: "", JsonRpcResponse.Id.from(id))
        return objectMapper.writeValueAsString(json)
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
@@ -18,12 +18,15 @@ package io.emeraldpay.dshackle.quorum

 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.infinitape.etherjar.rpc.RpcException

 open class AlwaysQuorum: CallQuorum {

    private var resolved = false
    private var result: ByteArray? = null
+    private var rpcError: JsonRpcError? = null

    override fun init(head: Head) {
    }
@@ -42,10 +45,15 @@ open class AlwaysQuorum: CallQuorum {
        return true
    }

-    override fun record(error: RpcException, upstream: Upstream) {
+    override fun record(error: JsonRpcException, upstream: Upstream) {
+        this.rpcError = error.error
    }

    override fun getResult(): ByteArray? {
        return result
    }
+
+    override fun getError(): JsonRpcError? {
+        return rpcError
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
@@ -18,6 +18,8 @@ package io.emeraldpay.dshackle.quorum

 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.json.BlockJson
@@ -34,8 +36,9 @@ interface CallQuorum {
    fun isFailed(): Boolean

    fun record(response: ByteArray, upstream: Upstream): Boolean
-    fun record(error: RpcException, upstream: Upstream)
+    fun record(error: JsonRpcException, upstream: Upstream)
    fun getResult(): ByteArray?
+    fun getError(): JsonRpcError?

    companion object {
        fun untilResolved(cq: CallQuorum): Predicate<Any> {
--- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt
@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.quorum
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.infinitape.etherjar.rpc.JacksonRpcConverter
 import io.infinitape.etherjar.rpc.RpcException

@@ -55,7 +56,7 @@ open class NonEmptyQuorum(
        tries++
    }

-    override fun record(error: RpcException, upstream: Upstream) {
+    override fun record(error: JsonRpcException, upstream: Upstream) {
        tries++
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
@@ -18,6 +18,8 @@ package io.emeraldpay.dshackle.quorum

 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.infinitape.etherjar.rpc.RpcException
 import java.util.concurrent.atomic.AtomicReference

@@ -25,6 +27,7 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {

    private val result: AtomicReference<ByteArray> = AtomicReference()
    private val failed = AtomicReference(false)
+    private var rpcError: JsonRpcError? = null

    override fun init(head: Head) {
    }
@@ -46,7 +49,8 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
        return false
    }

-    override fun record(error: RpcException, upstream: Upstream) {
+    override fun record(error: JsonRpcException, upstream: Upstream) {
+        this.rpcError = error.error
        val lagging = upstream.getLag() > maxLag
        if (!lagging && result.get() == null) {
            failed.set(true)
@@ -56,4 +60,8 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
    override fun getResult(): ByteArray {
        return result.get()
    }
+
+    override fun getError(): JsonRpcError? {
+        return rpcError
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.quorum

 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.ApiSource
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.infinitape.etherjar.rpc.RpcException
@@ -56,8 +57,10 @@ class QuorumRpcReader(

        val defaultResult: Mono<Result> = Mono.just(quorum).flatMap { q ->
            if (q.isFailed()) {
-                //TODO record and return actual error details
-                Mono.error<Result>(RpcException(-32000, "Upstream error"))
+                Mono.error<Result>(
+                        q.getError()?.asException(JsonRpcResponse.IntId(1))
+                                ?: RpcException(-32000, "Unknown Upstream error")
+                )
            } else {
                log.warn("Empty result for ${key.method} as ${q}")
                Mono.empty<Result>()
@@ -74,9 +77,14 @@ class QuorumRpcReader(
                            .flatMap { response ->
                                response.requireResult()
                                        .onErrorResume { err ->
-                                            if (err is RpcException) {
+                                            if (err is RpcException || err is JsonRpcException) {
                                                // on error notify quorum, it may use error message or other details
-                                                quorum.record(err, api)
+                                                val cleanErr: JsonRpcException = when (err) {
+                                                    is RpcException -> JsonRpcException.from(err)
+                                                    is JsonRpcException -> err
+                                                    else -> throw IllegalStateException("Cannot convert from exception", err)
+                                                }
+                                                quorum.record(cleanErr, api)
                                                // it it's failed after that, then we don't need more calls, stop api source
                                                if (quorum.isFailed()) {
                                                    apis.resolve()
--- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt
@@ -19,6 +19,8 @@ package io.emeraldpay.dshackle.quorum
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.infinitape.etherjar.rpc.JacksonRpcConverter
 import io.infinitape.etherjar.rpc.RpcException
 import org.slf4j.LoggerFactory
@@ -28,6 +30,7 @@ abstract class ValueAwareQuorum<T>(
 ): CallQuorum {

    private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
+    private var rpcError: JsonRpcError? = null

    fun extractValue(response: ByteArray, clazz: Class<T>): T? {
        return Global.objectMapper.readValue(response.inputStream(), clazz)
@@ -45,12 +48,16 @@ abstract class ValueAwareQuorum<T>(
        return isResolved();
    }

-    override fun record(error: RpcException, upstream: Upstream) {
-        recordError(null, error.rpcMessage, upstream)
+    override fun record(error: JsonRpcException, upstream: Upstream) {
+        this.rpcError = error.error
+        recordError(null, error.error.message, upstream)
    }

    abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream)

    abstract fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream)

+    override fun getError(): JsonRpcError? {
+        return rpcError
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
@@ -25,6 +25,8 @@ import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
 import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
@@ -48,6 +50,12 @@ open class NativeCall(
    var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()

    open fun nativeCall(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
+        return nativeCallResult(requestMono)
+                .map(this::buildResponse)
+                .onErrorResume(this::processException)
+    }
+
+    open fun nativeCallResult(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<CallResult> {
        return requestMono.flatMapMany(this::prepareCall)
                .map(this::parseParams)
                .parallel()
@@ -56,8 +64,6 @@ open class NativeCall(
                            .doOnError { e -> log.warn("Error during native call: ${e.message}") }
                }
                .sequential()
-                .map(this::buildResponse)
-                .onErrorResume(this::processException)
    }

    fun parseParams(it: CallContext<RawCallDetails>): CallContext<ParsedCallDetails> {
@@ -201,13 +207,14 @@ open class NativeCall(

    open class CallFailure(val id: Int, val reason: Throwable) : Exception("Failed to call $id: ${reason.message}")

-    open class CallError(val id: Int, val message: String) {
+    open class CallError(val id: Int, val message: String, val upstreamError: JsonRpcError?) {
        companion object {
            fun from(t: Throwable): CallError {
                return when (t) {
-                    is RpcException -> CallError(t.code, t.rpcMessage)
-                    is CallFailure -> CallError(t.id, t.reason.message ?: "Upstream Error")
-                    else -> CallError(1, t.message ?: "Upstream Error")
+                    is JsonRpcException -> CallError(t.id.asInt(), t.error.message, t.error)
+                    is RpcException -> CallError(t.code, t.rpcMessage, null)
+                    is CallFailure -> CallError(t.id, t.reason.message ?: "Upstream Error", null)
+                    else -> CallError(1, t.message ?: "Upstream Error", null)
                }
            }
        }
@@ -220,7 +227,7 @@ open class NativeCall(
            }

            fun fail(id: Int, errorCore: Int, errorMessage: String): CallResult {
-                return CallResult(id, null, CallError(errorCore, errorMessage))
+                return CallResult(id, null, CallError(errorCore, errorMessage, null))
            }

            fun fail(id: Int, error: Throwable): CallResult {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
@@ -56,7 +56,7 @@ class EthereumRpcHead(
                            .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
                            .flatMap {
                                if (it.error != null) {
-                                    Mono.error(it.error.asException())
+                                    Mono.error(it.error.asException(null))
                                } else {
                                    val value = it.getResultAsProcessedString()
                                    Mono.just(HexQuantity.from(value))
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt
@@ -0,0 +1,56 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+import io.infinitape.etherjar.rpc.RpcException
+
+class JsonRpcError(val code: Int, val message: String, val details: Any?) {
+
+    constructor(code: Int, message: String) : this(code, message, null)
+
+    companion object {
+        @JvmStatic
+        fun from(err: RpcException): JsonRpcError {
+            return JsonRpcError(
+                    err.code, err.rpcMessage, err.details
+            )
+        }
+    }
+
+    fun asException(id: JsonRpcResponse.Id?): JsonRpcException {
+        return JsonRpcException(id ?: JsonRpcResponse.IntId(-1), this)
+    }
+
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (other !is JsonRpcError) return false
+
+        if (code != other.code) return false
+        if (message != other.message) return false
+        if (details != other.details) return false
+
+        return true
+    }
+
+    override fun hashCode(): Int {
+        var result = code
+        result = 31 * result + message.hashCode()
+        result = 31 * result + (details?.hashCode() ?: 0)
+        return result
+    }
+
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcException.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcException.kt
@@ -0,0 +1,41 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+import io.infinitape.etherjar.rpc.RpcException
+
+class JsonRpcException(
+        val id: JsonRpcResponse.Id,
+        val error: JsonRpcError
+) : Exception(error.message) {
+
+    constructor(id: Int, message: String) : this(JsonRpcResponse.IntId(id), JsonRpcError(-32005, message))
+
+    companion object {
+        fun from(err: RpcException): JsonRpcException {
+            val id = err.details?.let {
+                if (it is JsonRpcResponse.Id) {
+                    it
+                } else {
+                    JsonRpcResponse.IntId(-3)
+                }
+            } ?: JsonRpcResponse.IntId(-4)
+            return JsonRpcException(
+                    id, JsonRpcError.from(err)
+            )
+        }
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
@@ -88,7 +88,9 @@ class JsonRpcHttpClient(

        return response.response { header, bytes ->
            if (header.status().code() != 200) {
-                Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "HTTP Code: ${header.status().code()}"))
+                Mono.error(JsonRpcException(JsonRpcResponse.IntId(-2),
+                        JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "HTTP Code: ${header.status().code()}"))
+                )
            } else {
                bytes.aggregate().asByteArray()
            }
@@ -101,10 +103,10 @@ class JsonRpcHttpClient(
                .flatMap(this@JsonRpcHttpClient::execute)
                .map(parser::parse)
                .onErrorResume { t ->
-                    val err = if (t is RpcException) {
-                        JsonRpcResponse.error(t.code, t.rpcMessage)
-                    } else {
-                        JsonRpcResponse.error(1, t.message ?: t.javaClass.name)
+                    val err = when (t) {
+                        is RpcException -> JsonRpcResponse.error(t.code, t.rpcMessage)
+                        is JsonRpcException -> JsonRpcResponse.error(t.error, JsonRpcResponse.IntId(1))
+                        else -> JsonRpcResponse.error(1, t.message ?: t.javaClass.name)
                    }
                    Mono.just(err)
                }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt
@@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.JsonFactory
 import com.fasterxml.jackson.core.JsonParseException
 import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.core.JsonToken
+import io.emeraldpay.dshackle.Global
 import io.infinitape.etherjar.rpc.RpcResponseError
 import org.slf4j.LoggerFactory

@@ -35,14 +36,14 @@ class JsonRpcParser() {
            val parser: JsonParser = jsonFactory.createParser(json)
            parser.nextToken()
            if (parser.currentToken != JsonToken.START_OBJECT) {
-                return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON"))
+                return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON"))
            }
            var nullResponse: JsonRpcResponse? = null
            while (parser.nextToken() != JsonToken.END_OBJECT) {
                val field = parser.currentName
                if (field == "jsonrpc" || field == "id") {
                    if (!parser.nextToken().isScalarValue) {
-                        return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id/type)"))
+                        return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id or jsonrpc value)"))
                    }
                    // just skip the field
                } else if (field == "result") {
@@ -78,12 +79,13 @@ class JsonRpcParser() {
        } catch (e: JsonParseException) {
            log.warn("Failed to parse JSON from upstream: ${e.message}")
        }
-        return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure"))
+        return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure"))
    }

-    fun readError(parser: JsonParser): JsonRpcResponse.ResponseError? {
+    fun readError(parser: JsonParser): JsonRpcError? {
        var code = 0
        var message = ""
+        var details: Any? = null

        while (parser.nextToken() != JsonToken.END_OBJECT) {
            if (parser.currentToken() == JsonToken.VALUE_NULL) {
@@ -95,9 +97,15 @@ class JsonRpcParser() {
                code = parser.intValue
            } else if (field == "message" && parser.currentToken == JsonToken.VALUE_STRING) {
                message = parser.valueAsString
+            } else if (field == "data") {
+                when (val value = parser.nextToken()) {
+                    JsonToken.VALUE_NULL -> details = null
+                    JsonToken.VALUE_STRING -> details = parser.valueAsString
+                    JsonToken.START_OBJECT -> details = Global.objectMapper.readValue(parser, java.util.Map::class.java)
+                    else -> log.warn("Unsupported error data type $value")
+                }
            }
        }
-
-        return JsonRpcResponse.ResponseError(code, message)
+        return JsonRpcError(code, message, details)
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt
@@ -18,16 +18,15 @@ package io.emeraldpay.dshackle.upstream.rpcclient
 import com.fasterxml.jackson.core.JsonGenerator
 import com.fasterxml.jackson.databind.JsonSerializer
 import com.fasterxml.jackson.databind.SerializerProvider
-import io.infinitape.etherjar.rpc.RpcException
 import reactor.core.publisher.Mono

 class JsonRpcResponse(
        private val result: ByteArray?,
-        val error: ResponseError?,
+        val error: JsonRpcError?,
        val id: Id
 ) {

-    constructor(result: ByteArray?, error: ResponseError?) : this(result, error, IntId(0))
+    constructor(result: ByteArray?, error: JsonRpcError?) : this(result, error, IntId(0))

    companion object {
        private val NULL_VALUE = "null".toByteArray()
@@ -49,12 +48,17 @@ class JsonRpcResponse(

        @JvmStatic
        fun error(code: Int, msg: String): JsonRpcResponse {
-            return JsonRpcResponse(null, ResponseError(code, msg))
+            return JsonRpcResponse(null, JsonRpcError(code, msg))
+        }
+
+        @JvmStatic
+        fun error(error: JsonRpcError, id: Id): JsonRpcResponse {
+            return JsonRpcResponse(null, error, id)
        }

        @JvmStatic
        fun error(code: Int, msg: String, id: Id): JsonRpcResponse {
-            return JsonRpcResponse(null, ResponseError(code, msg), id)
+            return JsonRpcResponse(null, JsonRpcError(code, msg), id)
        }
    }

@@ -88,7 +92,7 @@ class JsonRpcResponse(

    fun requireResult(): Mono<ByteArray> {
        return if (error != null) {
-            Mono.error(error.asException())
+            Mono.error(error.asException(id))
        } else {
            Mono.just(getResult())
        }
@@ -96,7 +100,7 @@ class JsonRpcResponse(

    fun requireStringResult(): Mono<String> {
        return if (error != null) {
-            Mono.error(error.asException())
+            Mono.error(error.asException(id))
        } else {
            Mono.just(getResultAsProcessedString())
        }
@@ -121,12 +125,6 @@ class JsonRpcResponse(
        return result1
    }

-    class ResponseError(val code: Int, val message: String) {
-        fun asException(): RpcException {
-            return RpcException(code, message)
-        }
-    }
-
    /**
     * JSON RPC wrapper. Makes sure that the id is either Int or String
     */
@@ -220,6 +218,13 @@ class JsonRpcResponse(
                gen.writeObjectFieldStart("error")
                gen.writeNumberField("code", value.error.code)
                gen.writeStringField("message", value.error.message)
+                value.error.details?.let { details ->
+                    when (details) {
+                        is String -> gen.writeStringField("data", details)
+                        is Number -> gen.writeNumberField("data", details.toInt())
+                        else -> gen.writeObjectField("data", details)
+                    }
+                }
                gen.writeEndObject()
            } else {
                if (value.result == null) {