solution: use reactor based stubs for grpc server

2019-07-29 22:27:15 -04:00
parent 4c9343c146
commit 36e089f65f
18 changed files with 448 additions and 304 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt
@@ -1,6 +1,7 @@
 package io.emeraldpay.dshackle

 import com.fasterxml.jackson.core.Version
+import com.fasterxml.jackson.databind.DeserializationFeature
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.module.SimpleModule
 import org.springframework.context.annotation.Bean
@@ -22,6 +23,7 @@ open class Config {

        val objectMapper = ObjectMapper()
        objectMapper.registerModule(module)
+        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
        objectMapper
                .setDateFormat(SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss.SSS"))
                .setTimeZone(TimeZone.getTimeZone("UTC"))
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
@@ -1,15 +1,14 @@
 package io.emeraldpay.dshackle.rpc

-import io.emeraldpay.api.proto.BlockchainGrpc
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
-import io.emeraldpay.grpc.Chain
+import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.grpc.stub.StreamObserver
-import io.infinitape.etherjar.domain.TransactionId
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
-import java.time.Instant
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono

@Service
 class BlockchainRpc(
@@ -19,51 +18,35 @@ class BlockchainRpc(
        @Autowired private val trackAddress: TrackAddress,
        @Autowired private val describe: Describe,
        @Autowired private val subscribeStatus: SubscribeStatus
-): BlockchainGrpc.BlockchainImplBase() {
+): ReactorBlockchainGrpc.BlockchainImplBase() {

    private val log = LoggerFactory.getLogger(BlockchainRpc::class.java)

-    override fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver<BlockchainOuterClass.NativeCallReplyItem>) {
-        nativeCall.nativeCall(request, responseObserver)
+    override fun nativeCall(request: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
+        return nativeCall.nativeCall(request)
    }

-    override fun subscribeHead(request: Common.Chain, responseObserver: StreamObserver<BlockchainOuterClass.ChainHead>) {
-        streamHead.add(Chain.byId(request.type.number), responseObserver)
+    override fun subscribeHead(request: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
+        return streamHead.add(request)
    }

-    override fun subscribeTxStatus(request: BlockchainOuterClass.TxStatusRequest, responseObserver: StreamObserver<BlockchainOuterClass.TxStatus>) {
-        val tx = TrackTx.TrackedTx(
-                Chain.byId(request.chainValue),
-                StreamSender(responseObserver),
-                Instant.now(),
-                TransactionId.from(request.txId),
-                Math.min(Math.max(1, request.confirmationLimit), 100)
-        )
-        trackTx.add(tx)
+    override fun subscribeTxStatus(request: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
+        return trackTx.add(request)
    }

-    override fun subscribeBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>) {
-        trackAddress.add(request, responseObserver)
+    override fun subscribeBalance(request: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
+        return trackAddress.subscribe(request)
    }

-    override fun getBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>) {
-        val addresses = trackAddress.initializeFor(request, responseObserver)
-        trackAddress.send(request, addresses)
-                .doOnError { t ->
-                    log.error("Failed to process balance", t)
-                    responseObserver.onError(Exception("Internal error"))
-                }
-                .subscribe {
-                    responseObserver.onCompleted()
-                }
-
+    override fun getBalance(request: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
+        return trackAddress.getBalance(request)
    }

-    override fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver<BlockchainOuterClass.DescribeResponse>) {
-        describe.describe(request, responseObserver)
+    override fun describe(request: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {
+        return describe.describe(request)
    }

-    override fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver<BlockchainOuterClass.ChainStatus>) {
-        subscribeStatus.subscribeStatus(request, responseObserver)
+    override fun subscribeStatus(request: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
+        return subscribeStatus.subscribeStatus(request)
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
@@ -10,6 +10,7 @@ import io.emeraldpay.grpc.Chain
 import io.grpc.stub.StreamObserver
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
+import reactor.core.publisher.Mono

@Service
 class Describe(
@@ -17,25 +18,26 @@ class Describe(
        @Autowired private val subscribeStatus: SubscribeStatus
 ) {

-    fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver<BlockchainOuterClass.DescribeResponse>) {
-        val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
-        upstreams.getAvailable().forEach { chain ->
-            upstreams.getUpstream(chain)?.let { chainUpstreams ->
-                chainUpstreams.getAll().let { ups ->
-                    if (ups.isNotEmpty()) {
-                        val status = subscribeStatus.chainStatus(chain, ups)
-                        resp.addChains(
-                                BlockchainOuterClass.DescribeChain.newBuilder()
-                                        .setChain(Common.ChainRef.forNumber(chain.id))
-                                        .setStatus(status)
-                                        .build()
-                        )
+    fun describe(requestMono: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {
+        return requestMono.map { _ ->
+            val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
+            upstreams.getAvailable().forEach { chain ->
+                upstreams.getUpstream(chain)?.let { chainUpstreams ->
+                    chainUpstreams.getAll().let { ups ->
+                        if (ups.isNotEmpty()) {
+                            val status = subscribeStatus.chainStatus(chain, ups)
+                            resp.addChains(
+                                    BlockchainOuterClass.DescribeChain.newBuilder()
+                                            .setChain(Common.ChainRef.forNumber(chain.id))
+                                            .setStatus(status)
+                                            .build()
+                            )
+                        }
                    }
                }
            }
+            resp.build()
        }
-        responseObserver.onNext(resp.build())
-        responseObserver.onCompleted()
    }

 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
@@ -4,12 +4,15 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
+import io.emeraldpay.dshackle.upstream.EthereumApi
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
 import io.grpc.stub.StreamObserver
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 import reactor.core.publisher.toFlux
 import reactor.core.publisher.toMono
 import reactor.util.function.Tuples
@@ -23,55 +26,50 @@ class NativeCall(

    private val log = LoggerFactory.getLogger(NativeCall::class.java)

-    open fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver<BlockchainOuterClass.NativeCallReplyItem>) {
-        val chain= Chain.byId(request.chain.number)
-        if (chain == Chain.UNSPECIFIED) {
-            throw Exception("Invalid chain id: ${request.chain.number}")
+    open fun nativeCall(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
+        return requestMono.flatMapMany { request ->
+            val chain= Chain.byId(request.chain.number)
+            if (chain == Chain.UNSPECIFIED) {
+                throw Exception("Invalid chain id: ${request.chain.number}")
+            }
+            val upstream = upstreams.getUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable")
+            request.itemsList.toFlux().map {
+                val method = it.target
+                val params = it.payload.toStringUtf8()
+                CallContext(it.id, upstream, Tuples.of(method, params))
+            }
+        }
+        .map {
+            val params = extractParams(it.payload.t2)
+            it.withPayload(Tuples.of(it.payload.t1, params))
+        }
+        .flatMap { ctx ->
+            ctx.upstream.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { resp ->
+                ctx.withPayload(resp)
+            }.onErrorMap {
+                CallFailure(ctx.id, it)
+            }
+        }
+        .map {
+            BlockchainOuterClass.NativeCallReplyItem.newBuilder()
+                    .setSucceed(true)
+                    .setId(it.id)
+                    .setPayload(ByteString.copyFrom(it.payload))
+                    .build()
+        }
+        .onErrorResume() {
+            val id: Int = if (it != null && CallFailure::class.isInstance(it)) {
+                (it as CallFailure).id
+            } else {
+                log.error("Lost context for a native call", it)
+                0
+            }
+            BlockchainOuterClass.NativeCallReplyItem.newBuilder()
+                    .setSucceed(false)
+                    .setId(id)
+                    .build()
+                    .toMono()
        }
-        val upstream = upstreams.getUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable")
-        request.itemsList.toFlux()
-                .map {
-                    val method = it.target
-                    val params = it.payload.toStringUtf8()
-                    return@map CallContext(it.id, Tuples.of(method, params))
-                }
-                .map {
-                    val params = extractParams(it.payload.t2)
-                    return@map it.withPayload(Tuples.of(it.payload.t1, params))
-                }
-                .flatMap { ctx ->
-                    upstream.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { resp ->
-                        ctx.withPayload(resp)
-                    }.onErrorMap {
-                        CallFailure(ctx.id, it)
-                    }
-                }
-                .map {
-                    BlockchainOuterClass.NativeCallReplyItem.newBuilder()
-                            .setSucceed(true)
-                            .setId(it.id)
-                            .setPayload(ByteString.copyFrom(it.payload))
-                            .build()
-                }
-                .onErrorResume() {
-                    val id: Int = if (it != null && CallFailure::class.isInstance(it)) {
-                        (it as CallFailure).id
-                    } else {
-                        log.error("Lost context for a native call", it)
-                        0
-                    }
-                    return@onErrorResume BlockchainOuterClass.NativeCallReplyItem.newBuilder()
-                            .setSucceed(false)
-                            .setId(id)
-                            .build()
-                            .toMono()
-                }
-                .doOnComplete {
-                    responseObserver.onCompleted()
-                }
-                .subscribe {
-                    responseObserver.onNext(it)
-                }
    }

    private fun extractParams(jsonParams: String): List<Any> {
@@ -79,9 +77,9 @@ class NativeCall(
        return req as List<Any>
    }

-    private class CallContext<T>(val id: Int, val payload: T) {
+    private class CallContext<T>(val id: Int, val upstream: EthereumApi, val payload: T) {
        fun <X> withPayload(payload: X): CallContext<X> {
-            return CallContext(id, payload)
+            return CallContext(id, upstream, payload)
        }
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
@@ -2,15 +2,18 @@ package io.emeraldpay.dshackle.rpc

 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
-import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.upstream.AvailableChains
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
-import io.grpc.stub.StreamObserver
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toFlux
 import java.lang.Exception
 import java.util.concurrent.ConcurrentLinkedQueue
@@ -19,33 +22,36 @@ import kotlin.collections.HashMap

@Service
 class StreamHead(
-        @Autowired private val upstreams: Upstreams
+        @Autowired private val upstreams: Upstreams,
+        @Autowired private val availableChains: AvailableChains
 ) {

    private val log = LoggerFactory.getLogger(StreamHead::class.java)
-    private val clients = HashMap<Chain, ConcurrentLinkedQueue<StreamSender<BlockchainOuterClass.ChainHead>>>()
+    private val clients = HashMap<Chain, ConcurrentLinkedQueue<TopicProcessor<BlockchainOuterClass.ChainHead>>>()

    @PostConstruct
    fun init() {
-        listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.TESTNET_MORDEN, Chain.TESTNET_KOVAN).forEach { chain ->
-            if (upstreams.getUpstream(chain)?.getHead() != null) {
-                clients[chain] = ConcurrentLinkedQueue()
-                subscribe(chain)
-            }
+        availableChains.observe().subscribe { chain ->
+            clients[chain] = ConcurrentLinkedQueue()
+            subscribe(chain)
        }
    }

    private fun subscribe(chain: Chain) {
-        upstreams.getUpstream(chain)!!.getHead().getFlux()
+        upstreams.getUpstream(chain)?.let { up ->
+            up.getHead()
+                .getFlux()
                .doOnComplete {
                    log.info("Closing streams for ${chain.chainCode}")
                    clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client ->
                        try {
-                            client.stream.onCompleted()
-                        } catch (e: Throwable) {}
+                            client.dispose()
+                        } catch (e: Throwable) {
+                        }
                    }
                }
                .subscribe { block -> onBlock(chain, block) }
+        }
    }

    private fun onBlock(chain: Chain, block: BlockJson<TransactionId>) {
@@ -56,25 +62,28 @@ class StreamHead(
                }
    }

-    fun add(chain: Chain, client: StreamObserver<BlockchainOuterClass.ChainHead>) {
-        val sender = StreamSender(client)
-        if (!clients.containsKey(chain)) {
-            client.onError(Exception("Chain ${chain.chainCode} is not available for streaming"))
-            return
+    fun add(requestMono: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
+        return requestMono.map { request ->
+            Chain.byId(request.type.number)
+        }.filter {
+            it != Chain.UNSPECIFIED && clients.containsKey(it)
+        }.flatMapMany { chain ->
+            val sender = TopicProcessor.create<BlockchainOuterClass.ChainHead>()
+            clients[chain]!!.add(sender)
+            notify(chain, sender)
+            sender
        }
-        clients[chain]!!.add(sender)
-        process(chain, sender)
    }

-    fun process(chain: Chain, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {
-        val upstream = upstreams.getUpstream(chain) ?: return false
+    fun notify(chain: Chain, client: TopicProcessor<BlockchainOuterClass.ChainHead>) {
+        val upstream = upstreams.getUpstream(chain) ?: return
        val head = upstream.getHead().getHead()
-        return head.map {
+        head.subscribe {
            notify(chain, it, client)
-        }.defaultIfEmpty(false).block()!!
+        }
    }

-    fun notify(chain: Chain, block: BlockJson<TransactionId>, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {
+    fun notify(chain: Chain, block: BlockJson<TransactionId>, client: TopicProcessor<BlockchainOuterClass.ChainHead>) {
        val data = BlockchainOuterClass.ChainHead.newBuilder()
                .setChainValue(chain.id)
                .setHeight(block.number)
@@ -82,16 +91,8 @@ class StreamHead(
                .setWeight(ByteString.copyFrom(block.totalDifficulty.toByteArray()))
                .setBlockId(block.hash.toHex().substring(2))
                .build()
-        var sent: Boolean = false
-        try {
-            sent = client.send(data)
-            if (!sent) {
-                clients[chain]!!.remove(client)
-            }
-        } catch (e: Exception) {
-            log.error("Send error ${e.javaClass}: ${e.message}")
-        }
-        return sent
+        client.onNext(data)
    }

+
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamSender.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamSender.kt
@@ -1,26 +0,0 @@
-package io.emeraldpay.dshackle.rpc
-
-import io.grpc.Status
-import io.grpc.StatusRuntimeException
-import io.grpc.stub.StreamObserver
-import org.slf4j.LoggerFactory
-
-class StreamSender<T>(val stream: StreamObserver<T>) {
-
-    private val log = LoggerFactory.getLogger(StreamSender::class.java)
-
-    fun send(value: T): Boolean {
-        try {
-            stream.onNext(value)
-            return true
-        } catch (e: StatusRuntimeException) {
-            if (e.status.code != Status.CANCELLED.code) {
-                log.warn("Channel errored with ${e.status}: ${e.message}")
-            }
-        } catch (e: Exception) {
-            log.warn("Channel errored with ${e.javaClass.name}: ${e.message}")
-            stream.onError(e)
-        }
-        return false
-    }
-}
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt
@@ -2,34 +2,32 @@ package io.emeraldpay.dshackle.rpc

 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
-import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.UpstreamAvailability
-import io.emeraldpay.dshackle.upstream.Upstreams
+import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.grpc.Chain
-import io.grpc.StatusRuntimeException
-import io.grpc.stub.StreamObserver
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
-import reactor.core.Disposable
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono

@Service
 class SubscribeStatus(
-        @Autowired private val upstreams: Upstreams
+        @Autowired private val upstreams: Upstreams,
+        @Autowired private val availableChains: AvailableChains
 ) {

-    fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver<BlockchainOuterClass.ChainStatus>) {
-        upstreams.getAvailable().forEach { chain ->
-            var d: Disposable? = null
-            val chainUpstream = upstreams.getUpstream(chain)
-            d = chainUpstream?.observeStatus()?.subscribe { availability ->
-                val status = chainStatus(chain, chainUpstream.getAll())
-                try {
-                    responseObserver.onNext(status)
-                } catch (e: StatusRuntimeException) {
-                    // gRPC channel was closed
-                    d?.dispose()
+    fun subscribeStatus(requestMono: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
+        return requestMono.flatMapMany {
+            val ups = availableChains.getAll().mapNotNull { chain ->
+                val chainUpstream = upstreams.getUpstream(chain)
+                chainUpstream?.observeStatus()?.map { avail ->
+                    ChainSubscription(chain, chainUpstream, avail)
                }
            }
+
+            Flux.merge(ups)
+                    .map {
+                        chainStatus(it.chain, it.up.getAll())
+                    }
        }
    }

@@ -40,12 +38,13 @@ class SubscribeStatus(
        val quorum = ups.filter {
            it.getStatus() > UpstreamAvailability.UNAVAILABLE
        }.count()
-        val status = BlockchainOuterClass.ChainStatus.newBuilder()
+        return BlockchainOuterClass.ChainStatus.newBuilder()
                .setAvailability(BlockchainOuterClass.AvailabilityEnum.forNumber(available.grpcId))
                .setChain(Common.ChainRef.forNumber(chain.id))
                .setQuorum(quorum)
                .build()
-        return status
    }

+    class ChainSubscription(val chain: Chain, val up: AggregatedUpstreams, val avail: UpstreamAvailability)
+
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt
@@ -2,19 +2,20 @@ package io.emeraldpay.dshackle.rpc

 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
-import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
+import io.emeraldpay.dshackle.upstream.AvailableChains
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
-import io.grpc.stub.StreamObserver
 import io.infinitape.etherjar.domain.Address
 import io.infinitape.etherjar.domain.Wei
 import io.infinitape.etherjar.rpc.Commands
 import io.infinitape.etherjar.rpc.json.BlockTag
+import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toFlux
 import reactor.math.sum
 import java.lang.Exception
@@ -26,19 +27,21 @@ import javax.annotation.PostConstruct

@Service
 class TrackAddress(
-        @Autowired private val upstreams: Upstreams
+        @Autowired private val upstreams: Upstreams,
+        @Autowired private val availableChains: AvailableChains
 ) {

+    private val log = LoggerFactory.getLogger(TrackAddress::class.java)
    private val clients = HashMap<Chain, ConcurrentLinkedQueue<TrackedAddress>>()

-    private val allChains = listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN)
-
    @PostConstruct
    fun init() {
-        allChains.forEach { chain ->
-            clients[chain] = ConcurrentLinkedQueue()
-            upstreams.getUpstream(chain)?.getHead()?.let { head ->
-                head.getFlux().subscribe { verifyAll(chain) }
+        availableChains.observe().subscribe { chain ->
+            if (!clients.containsKey(chain)) {
+                clients[chain] = ConcurrentLinkedQueue()
+                upstreams.getUpstream(chain)?.getHead()?.let { head ->
+                    head.getFlux().subscribe { verifyAll(chain) }
+                }
            }
        }
    }
@@ -46,7 +49,7 @@ class TrackAddress(
    @Scheduled(fixedDelay = 120_000)
    fun pingOld() {
        val period = Duration.ofMinutes(15)
-        allChains.forEach { chain ->
+        availableChains.getAll().forEach { chain ->
            clients[chain]?.let { clients ->
                clients.toFlux().filter {
                    it.lastPing < Instant.now().minus(period)
@@ -57,26 +60,32 @@ class TrackAddress(
        }
    }

-    fun initializeFor(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>): List<TrackedAddress> {
+    fun initializeSimple(request: BlockchainOuterClass.BalanceRequest): Flux<SimpleAddress> {
        val chain = Chain.byId(request.asset.chainValue)
-        if (!allChains.contains(chain)) {
-            responseObserver.onError(Exception("Unsupported chain ${request.asset.chainValue}"))
-            return Collections.emptyList()
+        if (!availableChains.supports(chain)) {
+            return Flux.error(Exception("Unsupported chain ${request.asset.chainValue}"))
        }
        if (request.asset.code?.toLowerCase() != "ether") {
-            responseObserver.onError(Exception("Unsupported asset ${request.asset.code}"))
-            return Collections.emptyList()
+            return Flux.error(Exception("Unsupported asset ${request.asset.code}"))
        }
-        val new = java.util.ArrayList<TrackedAddress>()
-        val observer = StreamSender<BlockchainOuterClass.AddressBalance>(responseObserver)
-        if (request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_SINGLE) {
-            new.add(forAddress(request.address.addressSingle, chain, observer))
-        } else if (request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_MULTI) {
-            request.address.addressMulti.addressesList.forEach { address ->
-                new.add(forAddress(address, chain, observer))
+        return when {
+            request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_SINGLE ->
+                Flux.just(simpleAddress(request.address.addressSingle, chain))
+            request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_MULTI ->
+                Flux.fromIterable(request.address.addressMulti.addressesList)
+                        .map { simpleAddress(it, chain) }
+            else -> {
+                log.error("Unsupported address type: ${request.address.addrTypeCase}")
+                Flux.empty()
            }
        }
-        return new
+    }
+
+    fun initializeSubscription(request: BlockchainOuterClass.BalanceRequest, observer: TopicProcessor<BlockchainOuterClass.AddressBalance>): Flux<TrackedAddress> {
+        return initializeSimple(request)
+                .map {
+                    it.asTracked(observer)
+                }
    }

    fun send(request: BlockchainOuterClass.BalanceRequest, addresses: List<TrackedAddress>): Mono<Long> {
@@ -86,21 +95,28 @@ class TrackAddress(
                .sum()
    }

-    fun add(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver<BlockchainOuterClass.AddressBalance>) {
-        val chain = Chain.byId(request.asset.chainValue)
-        val new = initializeFor(request, responseObserver)
-        send(request, new)
-                .doFinally {
-                    clients[chain]?.addAll(new)
-                }
-                .subscribe()
+    fun subscribe(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
+        return requestMono.flatMapMany { request ->
+            val chain = Chain.byId(request.asset.chainValue)
+            val sender = TopicProcessor.create<BlockchainOuterClass.AddressBalance>()
+            initializeSubscription(request, sender)
+                    .doOnNext { tracked -> clients[chain]?.add(tracked) }
+                    .thenMany(sender)
+        }
    }

-    private fun forAddress(address: Common.SingleAddress, chain: Chain, observer: StreamSender<BlockchainOuterClass.AddressBalance>): TrackedAddress {
+    fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
+        return requestMono.flatMapMany { request ->
+            initializeSimple(request)
+                    .flatMap { getBalance(it) }
+                    .map { process(it) }
+        }
+    }
+
+    private fun simpleAddress(address: Common.SingleAddress, chain: Chain): SimpleAddress {
        val addressParsed = Address.from(address.address)
-        return TrackedAddress(
+        return SimpleAddress(
                chain,
-                observer,
                addressParsed
        )
    }
@@ -116,13 +132,21 @@ class TrackAddress(
        }
    }

+    fun getBalance(addr: SimpleAddress): Mono<SimpleAddress> {
+        val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(Exception("Unsupported chain: ${addr.chain}"))
+        return up.getApi()
+                .executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST))
+                .timeout(Duration.ofSeconds(15))
+                .map { value ->
+                    addr.withBalance(value)
+                }
+    }
+
    private fun verify(chain: Chain, group: List<TrackedAddress>): Flux<TrackedAddress> {
        val up = upstreams.getUpstream(chain) ?: return Flux.empty<TrackedAddress>()
        return group.toFlux()
                .flatMap { a ->
-                    up.getApi()
-                            .executeAndConvert(Commands.eth().getBalance(a.address, BlockTag.LATEST))
-                            .map { Update(a, it) }
+                    getBalance(a).map { Update(a, it.balance!!) }
                }
                .filter {
                    it.addr.balance == null || it.addr.balance != it.value
@@ -135,31 +159,38 @@ class TrackAddress(
                }
    }

-    private fun notify(address: TrackedAddress): Boolean {
-        val sent = address.stream.send(
-                BlockchainOuterClass.AddressBalance.newBuilder()
-                        .setBalance(address.balance!!.amount!!.toString(10))
-                        .setAsset(Common.Asset.newBuilder()
-                                .setChainValue(address.chain.id)
-                                .setCode("ETHER")
-                        )
-                        .setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex()))
-                        .build()
-        )
-        if (!sent) {
-            clients[address.chain]?.remove(address)
-        }
+    private fun process(address: SimpleAddress): BlockchainOuterClass.AddressBalance {
+        return BlockchainOuterClass.AddressBalance.newBuilder()
+                .setBalance(address.balance!!.amount!!.toString(10))
+                .setAsset(Common.Asset.newBuilder()
+                        .setChainValue(address.chain.id)
+                        .setCode("ETHER")
+                )
+                .setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex()))
+                .build()
+    }
+
+    private fun notify(address: TrackedAddress) {
+        address.stream.onNext(process(address))
        address.lastPing = Instant.now()
-        return sent
    }

    class Update(val addr: TrackedAddress, val value: Wei)

-    class TrackedAddress(val chain: Chain,
-                         val stream: StreamSender<BlockchainOuterClass.AddressBalance>,
-                         val address: Address,
-                         val since: Instant = Instant.now(),
+    open class SimpleAddress(val chain: Chain, val address: Address, var balance: Wei? = null) {
+        fun asTracked(stream: TopicProcessor<BlockchainOuterClass.AddressBalance>): TrackedAddress {
+            return TrackedAddress(chain, stream, address, balance = this.balance)
+        }
+
+        open fun withBalance(balance: Wei) = SimpleAddress(chain, address, balance)
+    }
+
+    class TrackedAddress(chain: Chain,
+                         val stream: TopicProcessor<BlockchainOuterClass.AddressBalance>,
+                         address: Address,
                         var lastPing: Instant = Instant.now(),
-                         var balance: Wei? = null
-    )
+                         balance: Wei? = null
+    ): SimpleAddress(chain, address, balance) {
+        override fun withBalance(balance: Wei) = TrackedAddress(chain, stream, address, lastPing, balance);
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.rpc
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.upstream.AvailableChains
 import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
@@ -12,7 +13,9 @@ import io.infinitape.etherjar.rpc.Commands
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toFlux
 import java.lang.Exception
 import java.math.BigInteger
@@ -20,10 +23,13 @@ import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.ConcurrentLinkedQueue
 import javax.annotation.PostConstruct
+import kotlin.math.max
+import kotlin.math.min

@Service
 class TrackTx(
-        @Autowired private val upstreams: Upstreams
+        @Autowired private val upstreams: Upstreams,
+        @Autowired private val availableChains: AvailableChains
 ) {

    private val ZERO_BLOCK = BlockHash.from("0x0000000000000000000000000000000000000000000000000000000000000000")
@@ -33,7 +39,7 @@ class TrackTx(

    @PostConstruct
    fun init() {
-        listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN).forEach { chain ->
+        availableChains.observe().subscribe { chain ->
            clients[chain] = ConcurrentLinkedQueue()
            upstreams.getUpstream(chain)?.getHead()?.let { head ->
                head.getFlux().subscribe { verifyAll(chain) }
@@ -41,18 +47,42 @@ class TrackTx(
        }
    }

-    private fun currentList(chain: Chain): ConcurrentLinkedQueue<TrackedTx> {
-        return clients[chain]!!
+    private fun currentList(chain: Chain): ConcurrentLinkedQueue<TrackedTx>? {
+        return clients[chain]
    }

-    fun add(tx: TrackedTx) {
-        currentList(tx.chain).add(tx)
-        verify(tx)
-        notify(tx)
+    fun add(requestMono: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
+        return requestMono.map { request ->
+            val sender = TopicProcessor.create<BlockchainOuterClass.TxStatus>()
+            TrackTx.TrackedTx(
+                    Chain.byId(request.chainValue),
+                    sender,
+                    Instant.now(),
+                    TransactionId.from(request.txId),
+                    min(max(1, request.confirmationLimit), 100)
+            )
+        }.filter {
+            clients.containsKey(it.chain)
+        }.map { tx ->
+            currentList(tx.chain)!!.let { list ->
+                list.add(tx)
+                tx.stream.doOnError {
+                    list.remove(tx)
+                    tx.stream.dispose()
+                }
+            }
+            tx
+        }.map { tx ->
+            verify(tx)
+            notify(tx)
+            tx
+        }.flatMapMany { tx ->
+            tx.stream
+        }
    }

    private fun verifyAll(chain: Chain) {
-        currentList(chain)
+        currentList(chain)!!
                .toFlux()
                .filter(this::verify)
                .subscribe {
@@ -61,7 +91,8 @@ class TrackTx(
    }

    private fun loadWeight(tx: TrackedTx): Mono<TrackedTx> {
-        val upstream = upstreams.getUpstream(tx.chain)!!
+        val upstream = upstreams.getUpstream(tx.chain)
+                ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}"))
        return upstream.getApi()
                .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash))
                .map { block ->
@@ -81,7 +112,7 @@ class TrackTx(
    private fun verify(tx: TrackedTx): Boolean {
        val found = tx.status.found
        val mined = tx.status.mined
-        val upstream = upstreams.getUpstream(tx.chain)!!
+        val upstream = upstreams.getUpstream(tx.chain) ?: return false
        val execution = upstream.getApi()
                .executeAndConvert(Commands.eth().getTransaction(tx.txid))
        val update = execution.flatMap {
@@ -122,7 +153,7 @@ class TrackTx(
        return true
    }

-    private fun notify(tx: TrackedTx): Boolean {
+    private fun notify(tx: TrackedTx) {
        val client = tx.stream
        val data = BlockchainOuterClass.TxStatus.newBuilder()
                .setTxId(tx.txid.toHex())
@@ -140,23 +171,11 @@ class TrackTx(
                            .setTimestamp(tx.status.blockTime!!.toEpochMilli())
            )
        }
-        var sent: Boolean = false
-        try {
-            sent = client.send(data.build())
-            if (!sent || tx.shouldClose()) {
-                if (sent) {
-                    client.stream.onCompleted()
-                }
-                currentList(tx.chain).remove(tx)
-            }
-        } catch (e: Exception) {
-            log.error("Send error ${e.javaClass}: ${e.message}")
-        }
-        return sent
+        client.onNext(data.build())
    }

    class TrackedTx(val chain: Chain,
-                    val stream: StreamSender<BlockchainOuterClass.TxStatus>,
+                    val stream: TopicProcessor<BlockchainOuterClass.TxStatus>,
                    val since: Instant,
                    val txid: TransactionId,
                    val maxConfirmations: Int,
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt
@@ -0,0 +1,32 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.grpc.Chain
+import org.springframework.stereotype.Repository
+import reactor.core.publisher.Flux
+import reactor.core.publisher.TopicProcessor
+import java.util.*
+import kotlin.collections.LinkedHashSet
+
+@Repository
+class AvailableChains {
+
+    private val all = LinkedHashSet<Chain>()
+    private val bus = TopicProcessor.create<Chain>()
+
+    fun add(chain: Chain) {
+        all.add(chain)
+        bus.onNext(chain)
+    }
+
+    fun observe(): Flux<Chain> {
+        return Flux.from(bus)
+    }
+
+    fun supports(chain: Chain): Boolean {
+        return all.contains(chain)
+    }
+
+    fun getAll(): Set<Chain> {
+        return Collections.unmodifiableSet(all)
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt
@@ -12,6 +12,8 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.core.env.Environment
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Repository
+import reactor.core.publisher.Flux
+import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toFlux
 import java.io.File
 import java.net.URI
@@ -21,7 +23,8 @@ import javax.annotation.PostConstruct
@Repository
 open class ConfiguredUpstreams(
        @Autowired val env: Environment,
-        @Autowired private val objectMapper: ObjectMapper
+        @Autowired private val objectMapper: ObjectMapper,
+        @Autowired private val availableChains: AvailableChains
 ) : Upstreams {

    private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
@@ -123,7 +126,8 @@ open class ConfiguredUpstreams(
                    endpoint.port ?: 443,
                    objectMapper,
                    options,
-                    up.auth
+                    up.auth,
+                    availableChains
            )
            log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
            ds.start()
@@ -145,6 +149,7 @@ open class ConfiguredUpstreams(
        if (current == null) {
            val created = ChainUpstreams(chain, ArrayList<Upstream>())
            chainMapping[chain] = created
+            availableChains.add(chain)
            return created
        }
        return current
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt
@@ -11,11 +11,13 @@ import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.*
 import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.BlockTag
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toMono
+import java.lang.Exception
 import java.math.BigInteger
 import java.time.Duration
 import java.util.concurrent.atomic.AtomicReference
@@ -74,6 +76,11 @@ open class GrpcUpstream(
                    val curr = headBlock.get()
                    curr == null || curr.totalDifficulty < block.totalDifficulty
                }
+                .flatMap {
+                    getApi()
+                            .executeAndConvert(Commands.eth().getBlock(it.hash))
+                            .timeout(Duration.ofSeconds(15))
+                }
                .doOnError { err ->
                    log.error("Head subscription error", err)
                }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt
@@ -21,7 +21,8 @@ class GrpcUpstreams(
        private val port: Int,
        private val objectMapper: ObjectMapper,
        private val options: UpstreamsConfig.Options,
-        private val auth: UpstreamsConfig.TlsAuth? = null
+        private val auth: UpstreamsConfig.TlsAuth? = null,
+        private val availableChains: AvailableChains
 ) {
    private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)

@@ -90,6 +91,7 @@ class GrpcUpstreams(
            return if (current == null) {
                val created = GrpcUpstream(chain, client!!, objectMapper, options)
                known[chain] = created
+                availableChains.add(chain)
                created.connect()
                created
            } else {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt
@@ -1,6 +1,7 @@
 package io.emeraldpay.dshackle.upstream

 import io.emeraldpay.grpc.Chain
+import reactor.core.publisher.Flux

 interface Upstreams {
    fun getOrCreateUpstream(chain: Chain): AggregatedUpstreams
--- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy
@@ -1,16 +1,22 @@
 package io.emeraldpay.dshackle.test

 import com.fasterxml.jackson.databind.ObjectMapper
+import com.google.protobuf.ByteString
+import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.upstream.EthereumApi
 import io.emeraldpay.grpc.Chain
+import io.grpc.stub.StreamObserver
 import io.infinitape.etherjar.rpc.RpcClient
 import io.infinitape.etherjar.rpc.RpcResponseError
 import io.infinitape.etherjar.rpc.json.ResponseJson
 import org.jetbrains.annotations.NotNull
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono

 class EthereumApiMock extends EthereumApi {

+    private static final Logger log = LoggerFactory.getLogger(this)
    List<PredefinedResponse> predefined = []
    private ObjectMapper objectMapper

@@ -31,11 +37,26 @@ class EthereumApiMock extends EthereumApi {
        if (predefined != null) {
            json.result = predefined.result
        } else {
+            log.error("Method ${method} with ${params} is not mocked")
            json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked")
        }
        return Mono.just(objectMapper.writeValueAsBytes(json))
    }

+    def nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
+        request.itemsList.forEach { req ->
+            def resp = execute(req.id, req.target, objectMapper.readerFor(List).readValue(req.payload.toByteArray()))
+            resp.subscribe {
+                def proto = BlockchainOuterClass.NativeCallReplyItem.newBuilder()
+                        .setId(req.id)
+                        .setSucceed(true)
+                        .setPayload(ByteString.copyFrom(resp.block()))
+                responseObserver.onNext(proto.build())
+            }
+        }
+        responseObserver.onCompleted()
+    }
+
    class PredefinedResponse {
        String method
        List params
--- a/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy
@@ -10,7 +10,15 @@ class MockServer {

    GrpcCleanupRule grpcCleanup = new GrpcCleanupRule()

-    ReactorBlockchainGrpc.ReactorBlockchainStub runServer(BlockchainGrpc.BlockchainImplBase impl){
+    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
                .forName(serverName).directExecutor().addService(impl).build().start());
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy
@@ -1,17 +1,18 @@
 package io.emeraldpay.dshackle.upstream

 import com.fasterxml.jackson.databind.ObjectMapper
-import io.emeraldpay.api.proto.BlockchainGrpc
 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.rpc.NativeCall
 import io.emeraldpay.dshackle.test.EthereumApiMock
 import io.emeraldpay.dshackle.test.MockServer
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.grpc.Chain
-import io.grpc.stub.StreamObserver
 import io.infinitape.etherjar.rpc.Batch
 import io.infinitape.etherjar.rpc.RpcCall
 import io.infinitape.etherjar.rpc.RpcClient
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 import spock.lang.Specification

 class EthereumGrpcTransportSpec extends Specification {
@@ -27,11 +28,11 @@ class EthereumGrpcTransportSpec extends Specification {
        def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
        def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)

-        def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
+        def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() {
            @Override
-            void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
-                callData["request"] = request
-                otherSideNativeCall.nativeCall(request, responseObserver)
+            Flux<BlockchainOuterClass.NativeCallReplyItem> nativeCall(Mono<BlockchainOuterClass.NativeCallRequest> request) {
+                callData["request"] = request.block()
+                return otherSideNativeCall.nativeCall(request)
            }
        })

@@ -68,12 +69,13 @@ class EthereumGrpcTransportSpec extends Specification {
        def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
        def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)

-        def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
+        def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() {
            @Override
-            void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
-                callData["request"] = request
-                otherSideNativeCall.nativeCall(request, responseObserver)
+            Flux<BlockchainOuterClass.NativeCallReplyItem> nativeCall(Mono<BlockchainOuterClass.NativeCallRequest> request) {
+                callData["request"] = request.block()
+                return otherSideNativeCall.nativeCall(request)
            }
+
        })

        EthereumGrpcTransport transport = new EthereumGrpcTransport(Chain.ETHEREUM, client, objectMapper)
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy
@@ -5,14 +5,18 @@ 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.EthereumApiMock
 import io.emeraldpay.dshackle.test.MockServer
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.grpc.Chain
 import io.grpc.stub.StreamObserver
 import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.rpc.RpcClient
+import io.infinitape.etherjar.rpc.json.BlockJson
 import org.apache.commons.codec.binary.Hex
 import spock.lang.Specification

+import java.time.Duration
 import java.util.concurrent.CompletableFuture

 class GrpcUpstreamSpec extends Specification {
@@ -23,24 +27,37 @@ class GrpcUpstreamSpec extends Specification {
    def "Subscribe to head"() {
        setup:
        def callData = [:]
-        def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
+        def chain = Chain.ETHEREUM
+        def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain)
+        def block1 = new BlockJson().with {
+            it.number = 650246
+            it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
+            it.totalDifficulty = new BigInteger("35bbde5595de6456", 16)
+            return it
+        }
+        api.answer("eth_getBlockByHash", [block1.hash.toHex(), false], block1)
+        def client = mockServer.clientForServer(new BlockchainGrpc.BlockchainImplBase() {
+            @Override
+            void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
+                api.nativeCall(request, responseObserver)
+            }
+
            @Override
            void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
                callData.chain = request.getTypeValue()
                responseObserver.onNext(
                        BlockchainOuterClass.ChainHead.newBuilder()
-                            .setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
-                            .setHeight(650246)
-                            .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456")))
-                            .build()
+                                .setBlockId(block1.hash.toHex().substring(2))
+                                .setHeight(block1.number)
+                                .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray()))
+                                .build()
                )
            }
        })
-        def chain = Chain.ETHEREUM
        def upstream = new GrpcUpstream(chain, client, objectMapper)
        when:
        upstream.connect()
-        def h = upstream.head.head.block()
+        def h = upstream.head.head.block(Duration.ofSeconds(1))
        then:
        callData.chain == Chain.ETHEREUM.id
        upstream.status == UpstreamAvailability.OK
@@ -51,32 +68,52 @@ class GrpcUpstreamSpec extends Specification {
        setup:
        def callData = [:]
        def finished = new CompletableFuture<Boolean>()
-        def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
+        def chain = Chain.ETHEREUM
+        def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain)
+        def block1 = new BlockJson().with {
+            it.number = 650246
+            it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
+            it.totalDifficulty = new BigInteger("35bbde5595de6456", 16)
+            return it
+        }
+        def block2 = new BlockJson().with {
+            it.number = 650247
+            it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")
+            it.totalDifficulty = new BigInteger("35bbde5595de6455", 16)
+            return it
+        }
+        api.answer("eth_getBlockByHash", [block1.hash.toHex(), false], block1)
+        api.answer("eth_getBlockByHash", [block2.hash.toHex(), false], block2)
+        def client = mockServer.clientForServer(new BlockchainGrpc.BlockchainImplBase() {
+            @Override
+            void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
+                api.nativeCall(request, responseObserver)
+            }
+
            @Override
            void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
                responseObserver.onNext(
                        BlockchainOuterClass.ChainHead.newBuilder()
-                                .setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
-                                .setHeight(650246)
-                                .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456")))
+                                .setBlockId(block1.hash.toHex().substring(2))
+                                .setHeight(block1.number)
+                                .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray()))
                                .build()
                )
                responseObserver.onNext(
                        BlockchainOuterClass.ChainHead.newBuilder()
-                                .setBlockId("3ec2ebf5d0ec474d0ac6bca770d8409ad750d26e119968e7919f85d5ec891521")
-                                .setHeight(650247)
-                                .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6455")))
+                                .setBlockId(block2.hash.toHex().substring(2))
+                                .setHeight(block2.number)
+                                .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray()))
                                .build()
                )
                finished.complete(true)
            }
        })
-        def chain = Chain.ETHEREUM
        def upstream = new GrpcUpstream(chain, client, objectMapper)
        when:
        upstream.connect()
        finished.get()
-        def h = upstream.head.head.block()
+        def h = upstream.head.head.block(Duration.ofSeconds(1))
        then:
        upstream.status == UpstreamAvailability.OK
        h.hash == BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
@@ -87,32 +124,52 @@ class GrpcUpstreamSpec extends Specification {
        setup:
        def callData = [:]
        def finished = new CompletableFuture<Boolean>()
-        def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
+        def chain = Chain.ETHEREUM
+        def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain)
+        def block1 = new BlockJson().with {
+            it.number = 650246
+            it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
+            it.totalDifficulty = new BigInteger("35bbde5595de6456", 16)
+            return it
+        }
+        def block2 = new BlockJson().with {
+            it.number = 650247
+            it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")
+            it.totalDifficulty = new BigInteger("35bbde5595de6457", 16)
+            return it
+        }
+        api.answer("eth_getBlockByHash", [block1.hash.toHex(), false], block1)
+        api.answer("eth_getBlockByHash", [block2.hash.toHex(), false], block2)
+        def client = mockServer.clientForServer(new BlockchainGrpc.BlockchainImplBase() {
+            @Override
+            void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
+                api.nativeCall(request, responseObserver)
+            }
+
            @Override
            void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
                responseObserver.onNext(
                        BlockchainOuterClass.ChainHead.newBuilder()
-                                .setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
-                                .setHeight(650246)
-                                .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456")))
+                                .setBlockId(block1.hash.toHex().substring(2))
+                                .setHeight(block1.number)
+                                .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray()))
                                .build()
                )
                responseObserver.onNext(
                        BlockchainOuterClass.ChainHead.newBuilder()
-                                .setBlockId("3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")
-                                .setHeight(650247)
-                                .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6457")))
+                                .setBlockId(block2.hash.toHex().substring(2))
+                                .setHeight(block2.number)
+                                .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray()))
                                .build()
                )
                finished.complete(true)
            }
        })
-        def chain = Chain.ETHEREUM
        def upstream = new GrpcUpstream(chain, client, objectMapper)
        when:
        upstream.connect()
        finished.get()
-        def h = upstream.head.head.block()
+        def h = upstream.head.head.block(Duration.ofSeconds(1))
        then:
        upstream.status == UpstreamAvailability.OK
        h.hash == BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")