diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt index 8acff9e2..b525b1c2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt @@ -25,6 +25,8 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError +import io.grpc.StatusRuntimeException +import org.apache.commons.lang3.time.StopWatch import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.util.concurrent.TimeUnit @@ -32,7 +34,7 @@ import java.util.concurrent.TimeUnit class JsonRpcGrpcClient( private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val chain: Chain, - private val metrics: RpcMetrics + private val metrics: RpcMetrics?, ) { companion object { @@ -46,11 +48,11 @@ class JsonRpcGrpcClient( class Executor( private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val chain: Chain, - private val metrics: RpcMetrics + private val metrics: RpcMetrics? ) : Reader { override fun read(key: JsonRpcRequest): Mono { - var startTime: Long = 0 + val timer = StopWatch() val req = BlockchainOuterClass.NativeCallRequest.newBuilder() .setChainValue(chain.id) @@ -66,39 +68,58 @@ class JsonRpcGrpcClient( req.addItems(reqItem.build()) return Mono.just(key) - .doOnNext { - startTime = System.nanoTime() - }.flatMap { + .doOnNext { timer.start() } + .flatMap { stub.nativeCall(req.build()) .single() - .flatMap { resp -> - if (resp.succeed) { - val bytes = resp.payload.toByteArray() - val signature = if (resp.hasSignature()) { - extractSignature(resp.signature) - } else { - null - } - Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature)) - } else { - metrics.fails.increment() - Mono.error( - RpcException( - RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, - resp.errorMessage - ) - ) - } - } + .onErrorResume(::handleError) + .flatMap(::handleResponse) } .doOnNext { - if (startTime > 0) { - val now = System.nanoTime() - metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS) + if (timer.isStarted) { + metrics?.timer?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS) } } } + fun handleResponse(resp: BlockchainOuterClass.NativeCallReplyItem): Mono = + if (resp.succeed) { + val bytes = resp.payload.toByteArray() + val signature = if (resp.hasSignature()) { + extractSignature(resp.signature) + } else { + null + } + Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature)) + } else { + metrics?.fails?.increment() + Mono.error( + RpcException( + RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, + resp.errorMessage + ) + ) + } + + fun handleError(t: Throwable): Mono { + metrics?.fails?.increment() + return when (t) { + is StatusRuntimeException -> Mono.error( + RpcException( + RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, + "Remote status code: ${t.status.code.name}" + ) + ) + + else -> Mono.error( + RpcException( + RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, + "Other connection error" + ) + ) + } + } + fun extractSignature(resp: NativeCallReplySignature?): ResponseSigner.Signature? { if (resp == null || resp.signature == null || resp.signature.isEmpty || resp.upstreamId == null || resp.upstreamId.isEmpty()) { return null diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MockGrpcServer.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MockGrpcServer.groovy index 3e046e9b..f1d3689f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MockGrpcServer.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MockGrpcServer.groovy @@ -26,14 +26,6 @@ class MockGrpcServer { GrpcCleanupRule grpcCleanup = new GrpcCleanupRule() - ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(ReactorBlockchainGrpc.BlockchainImplBase impl) { - String serverName = InProcessServerBuilder.generateName() - grpcCleanup.register(InProcessServerBuilder - .forName(serverName).directExecutor().addService(impl).build().start()); - def channel = grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()) - return ReactorBlockchainGrpc.newReactorStub(channel) - } - ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(BlockchainGrpc.BlockchainImplBase impl){ String serverName = InProcessServerBuilder.generateName() grpcCleanup.register(InProcessServerBuilder diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClientSpec.groovy new file mode 100644 index 00000000..318a436f --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClientSpec.groovy @@ -0,0 +1,88 @@ +package io.emeraldpay.dshackle.upstream.rpcclient + +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainGrpc +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.test.MockGrpcServer +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.etherjar.rpc.RpcException +import io.emeraldpay.etherjar.rpc.RpcResponseError +import io.emeraldpay.dshackle.Chain +import io.grpc.stub.StreamObserver +import spock.lang.Specification + +import java.time.Duration +import java.util.concurrent.atomic.AtomicReference + +class JsonRpcGrpcClientSpec extends Specification { + + def "Makes a request"() { + setup: + def mockGrpc = new MockGrpcServer() + def requested = new AtomicReference() + + def grpc = mockGrpc.clientForServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { + requested.set(request) + responseObserver.onNext(BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(1) + .setSucceed(true) + .setPayload(ByteString.copyFromUtf8("\"hello world!\"")) + .build()) + responseObserver.onCompleted() + } + }) + def client = new JsonRpcGrpcClient( + grpc, Chain.BITCOIN, null + ).getReader() + + when: + def act = client.read( + new JsonRpcRequest("test", []) + ).block(Duration.ofSeconds(1)) + + then: + !act.hasError() + act.resultAsProcessedString == "hello world!" + + requested.get() == BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(Common.ChainRef.CHAIN_BITCOIN) + .addAllItems([ + BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(1) + .setMethod("test") + .setPayload(ByteString.copyFromUtf8("[]")) + .build() + ]) + .build() + } + + def "Return error on HTTP error"() { + setup: + def mockGrpc = new MockGrpcServer() + def grpc = mockGrpc.clientForServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { + responseObserver.onError(new IllegalStateException("fail")) + } + }) + def client = new JsonRpcGrpcClient( + grpc, Chain.BITCOIN, null + ).getReader() + + when: + client.read( + new JsonRpcRequest("test", []) + ).block(Duration.ofSeconds(1)) + + then: + def t = thrown(RpcException) + with(t.error) { + message == "Remote status code: UNKNOWN" + it.code == RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR + } + + } +}