problem: stuck if upstream returned null

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt

@@ -24,6 +24,7 @@ import io.infinitape.etherjar.rpc.*
 import io.infinitape.etherjar.rpc.json.ResponseJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
+import reactor.core.publisher.switchIfEmpty
 import java.time.Duration
 
 open class DirectEthereumApi(
@@ -41,6 +42,10 @@ open class DirectEthereumApi(
             targets.isAllowed(method)   -> callUpstream(method, params)
             else -> Mono.error(RpcException(-32601, "Method not allowed or not found"))
         }
+        return processResult(id, method, result)
+    }
+
+    public fun processResult(id: Int, method: String, result: Mono<out Any>): Mono<ByteArray> {
         return result
                 .doOnError { t ->
                     log.warn("Upstream error: [${t.message}] for $method")
@@ -49,7 +54,18 @@ open class DirectEthereumApi(
                     val resp = ResponseJson<Any, Int>()
                     resp.id = id
                     resp.result = it
-                    objectMapper.writer().writeValueAsBytes(resp)
+                    resp
+                }
+                .switchIfEmpty(
+                        Mono.fromCallable {
+                            val resp = ResponseJson<Any, Int>()
+                            resp.id = id
+                            resp.result = null
+                            resp
+                        }
+                )
+                .map {
+                    objectMapper.writer().writeValueAsBytes(it)
                 }
                 .onErrorResume(StatusRuntimeException::class.java) { t ->
                     if (t.status.code == Status.Code.CANCELLED) {

src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy

@@ -60,6 +60,14 @@ class TestingCommons {
         return new EthereumUpstreamMock(Chain.ETHEREUM, api)
     }
 
+    static EthereumUpstreamMock upstream(DirectEthereumApi api, String method) {
+        return upstream(api, [method])
+    }
+
+    static EthereumUpstreamMock upstream(DirectEthereumApi api, List<String> methods) {
+        return new EthereumUpstreamMock(Chain.ETHEREUM, api, new DirectCallMethods(methods))
+    }
+
     static AggregatedUpstream aggregatedUpstream(DirectEthereumApi api) {
         return aggregatedUpstream(upstream(api))
     }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy

@@ -0,0 +1,92 @@
+/**
+ * Copyright (c) 2019 ETCDEV GmbH
+ *
+ * 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.ethereum
+
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.DirectCallMethods
+import io.infinitape.etherjar.rpc.ReactorRpcClient
+import io.infinitape.etherjar.rpc.RpcException
+import io.infinitape.etherjar.rpc.RpcResponseError
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+
+class DirectEthereumApiSpec extends Specification {
+
+    DirectEthereumApi api = new DirectEthereumApi(Stub(ReactorRpcClient), TestingCommons.objectMapper(), new DirectCallMethods())
+
+    def "Process successful result"() {
+        setup:
+        def result = Mono.just("hello")
+        when:
+        def act = api.processResult(1, "eth_test", result)
+            .map { new String(it) }
+
+        then:
+        StepVerifier.create(act)
+            .expectNext('{"jsonrpc":"2.0","id":1,"result":"hello"}')
+            .expectComplete()
+            .verify(Duration.ofSeconds(1))
+
+    }
+
+    def "Process empty result"() {
+        setup:
+        def result = Mono.empty()
+        when:
+        def act = api.processResult(1, "eth_test", result)
+                .map { new String(it) }
+
+        then:
+        StepVerifier.create(act)
+                .expectNext('{"jsonrpc":"2.0","id":1,"result":null}')
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+
+    }
+
+    def "Process standard RPC error"() {
+        setup:
+        def result = Mono.error(new RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Test Error", Map.of("foo", "bar")))
+        when:
+        def act = api.processResult(1, "eth_test", result)
+                .map { new String(it) }
+
+        then:
+        StepVerifier.create(act)
+                .expectNext('{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"Test Error","data":{"foo":"bar"}}}')
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+
+    }
+
+    def "Process internal exception"() {
+        setup:
+        def result = Mono.error(new InterruptedException("test"))
+        when:
+        def act = api.processResult(1, "eth_test", result)
+                .map { new String(it) }
+
+        then:
+        StepVerifier.create(act)
+                .expectNext('{"jsonrpc":"2.0","id":1,"error":{"code":-32020,"message":"Error reading from upstream"}}')
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+
+    }
+}