solution: initial implementation for Bitcoin upstreams

build.gradle

@@ -84,6 +84,7 @@ dependencies {
     compile "io.infinitape:etherjar-rpc-ws:$etherjarVersion"
     compile "io.infinitape:etherjar-rpc-emerald:$etherjarVersion"
     compile "io.infinitape:etherjar-tx:$etherjarVersion"
+    compile 'org.bitcoinj:bitcoinj-core:0.15.8'
 
     compile 'org.apache.httpcomponents:httpmime:4.5.8'
     compile 'org.apache.httpcomponents:httpclient:4.5.8'
@@ -108,9 +109,11 @@ dependencies {
     testCompile "org.codehaus.groovy:groovy:$groovyVersion"
     testCompile 'cglib:cglib-nodep:3.2.12'
     testCompile "org.spockframework:spock-core:$spockVersion"
+    testCompile 'de.jodamob.kotlin:kotlin-runner-spock:0.3.1'
     testCompile "io.grpc:grpc-testing:${grpcVersion}"
     testCompile "io.projectreactor:reactor-test:$reactorVersion"
     testCompile 'org.objenesis:objenesis:3.0.1'
+    testCompile 'org.mock-server:mockserver-netty:5.10'
 }
 
 compileKotlin {

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -96,6 +96,10 @@ class UpstreamsConfig {
         var ws: WsEndpoint? = null
     }
 
+    class BitcoinConnection : UpstreamConnection() {
+        var rpc: HttpEndpoint? = null
+    }
+
     class HttpEndpoint(val url: URI) {
         var basicAuth: AuthConfig.ClientBasicAuth? = null
         var tls: AuthConfig.ClientTlsAuth? = null
@@ -124,6 +128,7 @@ class UpstreamsConfig {
 
     enum class UpstreamType private constructor(vararg code: String) {
         ETHEREUM_JSON_RPC("ethereum"),
+        BITCOIN_JSON_RPC("bitcoin"),
         DSHACKLE("dshackle", "grpc"),
         UNKNOWN("unknown");
 

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -87,7 +87,7 @@ class UpstreamsConfigReader(
                 val connConfigNode = getMapping(connNode, "ethereum")!!
                 val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>()
                 readUpstreamCommon(upNode, upstream)
-                readUpstreamEthereum(upNode, upstream)
+                readUpstreamStandard(upNode, upstream)
                 if (isValid(upstream)) {
                     config.upstreams.add(upstream)
                     val connection = UpstreamsConfig.EthereumConnection()
@@ -113,6 +113,26 @@ class UpstreamsConfigReader(
                 } else {
                     log.error("Upstream at #0 has invalid configuration")
                 }
+            } else if (hasAny(connNode, "bitcoin")) {
+                val connConfigNode = getMapping(connNode, "bitcoin")!!
+                val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.BitcoinConnection>()
+                readUpstreamCommon(upNode, upstream)
+                readUpstreamStandard(upNode, upstream)
+                if (isValid(upstream)) {
+                    config.upstreams.add(upstream)
+                    val connection = UpstreamsConfig.BitcoinConnection()
+                    upstream.connection = connection
+                    getMapping(connConfigNode, "rpc")?.let { node ->
+                        getValueAsString(node, "url")?.let { url ->
+                            val http = UpstreamsConfig.HttpEndpoint(URI(url))
+                            connection.rpc = http
+                            http.basicAuth = authConfigReader.readClientBasicAuth(node)
+                            http.tls = authConfigReader.readClientTls(node)
+                        }
+                    }
+                } else {
+                    log.error("Upstream at #0 has invalid configuration")
+                }
             } else if (hasAny(connNode, "grpc")) {
                 val connConfigNode = getMapping(connNode, "grpc")!!
                 val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>()
@@ -162,13 +182,13 @@ class UpstreamsConfigReader(
         }
     }
 
-    internal fun readUpstreamEthereum(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>) {
+    internal fun readUpstreamStandard(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
         upstream.chain = getValueAsString(upNode, "chain")
         if (hasAny(upNode, "labels")) {
             getMapping(upNode, "labels")?.let { labels ->
                 labels.value.stream()
                         .filter { n -> n.keyNode is ScalarNode && n.valueNode is ScalarNode }
-                        .map { n -> Tuples.of((n.keyNode as ScalarNode).value, (n.valueNode as ScalarNode).value)}
+                        .map { n -> Tuples.of((n.keyNode as ScalarNode).value, (n.valueNode as ScalarNode).value) }
                         .map { kv -> Tuples.of(kv.t1.trim(), kv.t2.trim()) }
                         .filter { kv -> StringUtils.isNotEmpty(kv.t1) && StringUtils.isNotEmpty(kv.t2) }
                         .forEach { kv ->

src/main/kotlin/io/emeraldpay/dshackle/data/BlockId.kt

@@ -15,7 +15,9 @@
  */
 package io.emeraldpay.dshackle.data
 
+import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.rpc.json.BlockJson
+import org.bouncycastle.util.encoders.Hex
 
 class BlockId(
         value: ByteArray
@@ -23,7 +25,7 @@ class BlockId(
 
     companion object {
         @JvmStatic
-        fun from(hash: io.infinitape.etherjar.domain.BlockHash): BlockId {
+        fun from(hash: BlockHash): BlockId {
             return BlockId(hash.bytes)
         }
 
@@ -34,7 +36,13 @@ class BlockId(
 
         @JvmStatic
         fun from(id: String): BlockId {
-            return from(io.infinitape.etherjar.domain.BlockHash.from(id))
+            val clean = if (id.startsWith("0x")) {
+                id.substring(2)
+            } else {
+                id
+            }
+            val bytes = Hex.decode(clean)
+            return BlockId(bytes)
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/data/HashId.kt

@@ -28,11 +28,9 @@ open class HashId(
     }
 
     fun toHex(): String {
-        val hex = CharArray(value.size * 2 + 2)
+        val hex = CharArray(value.size * 2)
-        hex[0] = '0'
-        hex[1] = 'x'
         var i = 0
-        var j = 2
+        var j = 0
         while (i < value.size) {
             hex[j++] = HEX_DIGITS[0xF0 and value[i].toInt() ushr 4]
             hex[j++] = HEX_DIGITS[0x0F and value[i].toInt()]

src/main/kotlin/io/emeraldpay/dshackle/data/TxId.kt

@@ -17,6 +17,8 @@ package io.emeraldpay.dshackle.data
 
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.TransactionJson
+import org.bouncycastle.util.encoders.Hex
+import java.math.BigInteger
 
 class TxId(
         value: ByteArray
@@ -35,7 +37,13 @@ class TxId(
 
         @JvmStatic
         fun from(id: String): TxId {
-            return from(TransactionId.from(id))
+            val clean = if (id.startsWith("0x")) {
+                id.substring(2)
+            } else {
+                id
+            }
+            val bytes = Hex.decode(clean)
+            return TxId(bytes)
         }
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt

@@ -65,7 +65,7 @@ class StreamHead(
                 .setHeight(block.height)
                 .setTimestamp(block.timestamp!!.toEpochMilli())
                 .setWeight(ByteString.copyFrom(block.difficulty.toByteArray()))
-                .setBlockId(block.hash.toHex().substring(2))
+                .setBlockId(block.hash.toHex())
                 .build()
     }
 

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -20,6 +20,9 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.CurrentUpstreams
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcClient
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
@@ -32,6 +35,7 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Repository
 import java.net.URI
 import java.util.*
+import java.util.concurrent.atomic.AtomicInteger
 import javax.annotation.PostConstruct
 import kotlin.collections.HashMap
 
@@ -44,6 +48,7 @@ open class ConfiguredUpstreams(
 ) {
 
     private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
+    private var seq = AtomicInteger(0)
 
     private val chainNames = mapOf(
             "ethereum" to Chain.ETHEREUM,
@@ -51,7 +56,10 @@ open class ConfiguredUpstreams(
             "eth" to Chain.ETHEREUM,
             "etc" to Chain.ETHEREUM_CLASSIC,
             "morden" to Chain.TESTNET_MORDEN,
-            "kovan" to Chain.TESTNET_KOVAN
+            "kovan" to Chain.TESTNET_KOVAN,
+            "kovan-testnet" to Chain.TESTNET_KOVAN,
+            "bitcoin" to Chain.BITCOIN,
+            "bitcoin-testnet" to Chain.TESTNET_BITCOIN
     )
 
     @PostConstruct
@@ -69,13 +77,20 @@ open class ConfiguredUpstreams(
                     log.error("Chain is unknown: ${up.chain}")
                     return@forEach
                 }
-                if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) {
-                    log.error("Chain is unsupported: ${up.chain}")
-                    return@forEach
-                }
                 val options = (up.options ?: UpstreamsConfig.Options())
                         .merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults())
-                buildEthereumUpstream(up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
+                when (BlockchainType.fromBlockchain(chain)) {
+                    BlockchainType.ETHEREUM -> {
+                        buildEthereumUpstream(up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
+                    }
+                    BlockchainType.BITCOIN -> {
+                        buildBitcoinUpstream(up.cast(UpstreamsConfig.BitcoinConnection::class.java), chain, options)
+                    }
+                    else -> {
+                        log.error("Chain is unsupported: ${up.chain}")
+                        return@forEach
+                    }
+                }
             }
         }
     }
@@ -101,10 +116,30 @@ open class ConfiguredUpstreams(
         return defaultOptions
     }
 
+    private fun buildBitcoinUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.BitcoinConnection>,
+                                     chain: Chain,
+                                     options: UpstreamsConfig.Options) {
+
+        val conn = config.connection!!
+        var rpcApi: BitcoinApi? = null
+
+        conn.rpc?.let { endpoint ->
+            val rpcClient = BitcoinRpcClient(endpoint.url.toString(), endpoint.basicAuth!!)
+            rpcApi = BitcoinApi(rpcClient, objectMapper)
+        }
+        rpcApi?.let { api ->
+            val upstream = BitcoinUpstream(config.id
+                    ?: "bitcoin-${seq.getAndIncrement()}", chain, api, options, objectMapper)
+
+            upstream.start()
+            currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
+        }
+
+    }
+
     private fun buildEthereumUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
                                       chain: Chain,
-                                      options: UpstreamsConfig.Options
+                                      options: UpstreamsConfig.Options) {
-                                      ) {
         val conn = config.connection!!
         var rpcApi: DirectEthereumApi? = null
         val urls = ArrayList<URI>()

src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt

@@ -0,0 +1,52 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.data.BlockContainer
+import org.slf4j.LoggerFactory
+import reactor.core.Disposable
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.TopicProcessor
+import java.util.concurrent.atomic.AtomicReference
+
+abstract class AbstractHead : Head {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(AbstractHead::class.java)
+    }
+
+    private val head = AtomicReference<BlockContainer>(null)
+    private val stream: TopicProcessor<BlockContainer> = TopicProcessor.create()
+
+    fun follow(source: Flux<BlockContainer>): Disposable {
+        return source.distinctUntilChanged {
+            it.hash
+        }.filter { block ->
+            val curr = head.get()
+            curr == null || curr.difficulty < block.difficulty
+        }
+                .subscribe { block ->
+                    val prev = head.getAndUpdate { curr ->
+                        if (curr == null || curr.difficulty < block.difficulty) {
+                            block
+                        } else {
+                            curr
+                        }
+                    }
+                    if (prev == null || prev.hash != block.hash) {
+                        log.debug("New block ${block.height} ${block.hash}")
+                        stream.onNext(block)
+                    }
+                }
+    }
+
+    override fun getFlux(): Flux<BlockContainer> {
+        return Flux.merge(
+                Mono.justOrEmpty(head.get()),
+                Flux.from(stream)
+        ).onBackpressureLatest()
+    }
+
+    fun getCurrent(): BlockContainer? {
+        return head.get()
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt

@@ -20,7 +20,6 @@ import io.emeraldpay.dshackle.cache.*
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
@@ -89,7 +88,7 @@ abstract class AggregatedUpstream<U : UpstreamApi>(
         cacheSubscription = null
     }
 
-    fun onHeadUpdated(head: EthereumHead) {
+    fun onHeadUpdated(head: Head) {
         reconfigLock.withLock {
             cacheSubscription?.dispose()
             cacheSubscription = head.getFlux().subscribe {

src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt

@@ -18,9 +18,7 @@ package io.emeraldpay.dshackle.upstream
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.data.*
-import io.emeraldpay.dshackle.upstream.ethereum.EmptyEthereumHead
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import io.infinitape.etherjar.hex.HexQuantity
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
@@ -30,7 +28,7 @@ import java.util.function.Function
 open class CachingEthereumApi(
         private val objectMapper: ObjectMapper,
         private val caches: Caches,
-        private val head: EthereumHead
+        private val head: Head
 ): EthereumApi(objectMapper) {
 
     companion object {
@@ -41,7 +39,7 @@ open class CachingEthereumApi(
          */
         @JvmStatic
         fun empty(objectMapper: ObjectMapper): CachingEthereumApi {
-            return CachingEthereumApi(objectMapper, Caches.default(objectMapper), EmptyEthereumHead())
+            return CachingEthereumApi(objectMapper, Caches.default(objectMapper), EmptyHead())
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt

@@ -17,11 +17,14 @@ package io.emeraldpay.dshackle.upstream
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Mono
+import java.lang.IllegalStateException
+import java.time.Duration
 
 /**
  * General interface to upstream(s) to a single chain
@@ -111,5 +114,22 @@ abstract class ChainUpstreams<U : UpstreamApi>(
         return 0
     }
 
-    abstract fun printStatus()
+    fun printStatus() {
+        var height: Long? = null
+        try {
+            height = getHead().getFlux().next().block(Duration.ofSeconds(1))?.height
+        } catch (e: IllegalStateException) {
+            //timout
+        } catch (e: Exception) {
+            log.warn("Head processing error: ${e.javaClass} ${e.message}")
+        }
+        val statuses = upstreams.map { it.getStatus() }
+                .groupBy { it }
+                .map { "${it.key.name}/${it.value.size}" }
+                .joinToString(",")
+        val lag = upstreams.map { it.getLag() }
+                .joinToString(", ")
+
+        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]")
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt

@@ -16,17 +16,19 @@
 package io.emeraldpay.dshackle.upstream
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.cache.CachesFactory
 import io.emeraldpay.dshackle.startup.UpstreamChange
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
 import io.emeraldpay.grpc.Chain
-import io.infinitape.etherjar.rpc.json.BlockJson
-import io.infinitape.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.scheduling.annotation.Scheduled
@@ -34,6 +36,7 @@ import org.springframework.stereotype.Repository
 import reactor.core.publisher.Flux
 import reactor.core.publisher.TopicProcessor
 import java.util.*
+import java.util.concurrent.Callable
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
@@ -54,36 +57,60 @@ class CurrentUpstreams(
     fun update(change: UpstreamChange) {
         updateLock.withLock {
             val chain = change.chain
-            val up = change.upstream
+            when (BlockchainType.fromBlockchain(chain)) {
-                    .cast(EthereumUpstream::class.java, EthereumApi::class.java) as Upstream<EthereumApi>
+                BlockchainType.ETHEREUM -> {
-            val current = chainMapping[chain] as ChainUpstreams<EthereumApi>?
+                    val up = change.upstream
-            if (change.type == UpstreamChange.ChangeType.REMOVED) {
+                            .cast(EthereumUpstream::class.java, EthereumApi::class.java) as Upstream<EthereumApi>
-                current?.removeUpstream(up.getId())
+                    val current = chainMapping[chain] as ChainUpstreams<EthereumApi>?
-                log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed")
+                    val factory = Callable {
-            } else {
+                        EthereumChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<EthereumApi>
-                if (current == null) {
-                    val created = EthereumChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper)
-                    if (up is CachesEnabled) {
-                        up.setCaches(created.caches)
                     }
-                    created.addUpstream(up)
+                    processUpdate(change, up, current, factory)
-                    created.start()
+                }
-                    chainMapping[chain] = created
+                BlockchainType.BITCOIN -> {
-                    chainsBus.onNext(chain)
+                    val up = change.upstream
-                } else {
+                            .cast(BitcoinUpstream::class.java, BitcoinApi::class.java)
-                    if (up is CachesEnabled) {
+                    val current = chainMapping[chain] as ChainUpstreams<BitcoinApi>?
-                        up.setCaches(current.caches)
+                    val factory = Callable {
+                        BitcoinChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<BitcoinApi>
                     }
-                    current.addUpstream(up)
+                    processUpdate(change, up, current, factory)
                 }
-                if (!callTargets.containsKey(chain)) {
+                else -> {
-                    setupDefaultMethods(chain)
+                    log.error("Update for unsupported chain: $chain")
                 }
-                log.info("Upstream ${change.upstream.getId()} with chain $chain has been added")
             }
         }
     }
 
+    fun <A : UpstreamApi> processUpdate(change: UpstreamChange, up: Upstream<A>, current: ChainUpstreams<A>?, factory: Callable<ChainUpstreams<A>>) {
+        val chain = change.chain
+        if (change.type == UpstreamChange.ChangeType.REMOVED) {
+            current?.removeUpstream(up.getId())
+            log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed")
+        } else {
+            if (current == null) {
+                val created = factory.call()
+                if (up is CachesEnabled) {
+                    up.setCaches(created.caches)
+                }
+                created.addUpstream(up)
+                created.start()
+                chainMapping[chain] = created
+                chainsBus.onNext(chain)
+            } else {
+                if (up is CachesEnabled) {
+                    up.setCaches(current.caches)
+                }
+                current.addUpstream(up)
+            }
+            if (!callTargets.containsKey(chain)) {
+                setupDefaultMethods(chain)
+            }
+            log.info("Upstream ${change.upstream.getId()} with chain $chain has been added")
+        }
+    }
+
     override fun getUpstream(chain: Chain): AggregatedUpstream<*>? {
         return chainMapping[chain]
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt

@@ -15,20 +15,29 @@
  */
 package io.emeraldpay.dshackle.upstream
 
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import reactor.core.publisher.Flux
 import reactor.core.publisher.TopicProcessor
 import java.util.concurrent.atomic.AtomicReference
 
 abstract class DefaultUpstream<U : UpstreamApi>(
+        private val id: String,
         defaultLag: Long,
-        defaultAvail: UpstreamAvailability
+        defaultAvail: UpstreamAvailability,
+        private val options: UpstreamsConfig.Options,
+        private val targets: CallMethods?
 ) : Upstream<U> {
 
-    constructor() : this(Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE)
+    constructor(id: String, options: UpstreamsConfig.Options, targets: CallMethods?) : this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, targets)
 
     private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
     private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
 
+    override fun isAvailable(): Boolean {
+        return getStatus() == UpstreamAvailability.OK
+    }
+
     override fun getStatus(): UpstreamAvailability {
         return status.get().status
     }
@@ -67,5 +76,17 @@ abstract class DefaultUpstream<U : UpstreamApi>(
         return this.status.get().lag
     }
 
+    override fun getId(): String {
+        return id
+    }
+
+    override fun getOptions(): UpstreamsConfig.Options {
+        return options
+    }
+
+    override fun getMethods(): CallMethods {
+        return targets ?: throw IllegalStateException("Methods are not set")
+    }
+
     class Status(val lag: Long, val avail: UpstreamAvailability, val status: UpstreamAvailability)
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt

@@ -13,12 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.emeraldpay.dshackle.upstream.ethereum
+package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.Head
 import reactor.core.publisher.Flux
 
-class EmptyEthereumHead : EthereumHead {
+class EmptyHead : Head {
 
     override fun getFlux(): Flux<BlockContainer> {
         return Flux.empty()

src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.emeraldpay.dshackle.upstream.ethereum
+package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CachesEnabled
@@ -21,9 +21,9 @@ import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 
-class EthereumHeadMerge(
+class MergedHead(
-        private val sources: Iterable<EthereumHead>
+        private val sources: Iterable<Head>
-): DefaultEthereumHead(), Lifecycle, CachesEnabled {
+) : AbstractHead(), Lifecycle, CachesEnabled {
 
     private var subscription: Disposable? = null
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinApi.kt

@@ -0,0 +1,44 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import com.fasterxml.jackson.databind.JavaType
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.upstream.UpstreamApi
+import io.infinitape.etherjar.rpc.RpcException
+import io.infinitape.etherjar.rpc.json.FullResponseJson
+import io.infinitape.etherjar.rpc.json.RequestJson
+import io.infinitape.etherjar.rpc.json.ResponseJson
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+open class BitcoinApi(
+        val bitcoinRpcClient: BitcoinRpcClient,
+        val objectMapper: ObjectMapper
+) : UpstreamApi {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BitcoinApi::class.java)
+    }
+
+    open override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
+        //TODO optimize extraction
+        return executeAndResult(id, method, params, Object::class.java).map {
+            objectMapper.writeValueAsBytes(it)
+        }
+    }
+
+    open fun <T> executeAndResult(id: Int, method: String, params: List<Any>, resultType: Class<T>): Mono<T> {
+        val rpc = RequestJson<Int>(method, params, id)
+        return Mono.just(rpc)
+                .map(objectMapper::writeValueAsBytes)
+                .flatMap(bitcoinRpcClient::execute)
+                .flatMap { json ->
+                    val type: JavaType = objectMapper.typeFactory.constructParametricType(FullResponseJson::class.java, resultType, Int::class.java)
+                    val resp = objectMapper.readerFor(type).readValue<FullResponseJson<T, Int>>(json)
+                    if (resp.hasError()) {
+                        Mono.error(resp.error.asException())
+                    } else {
+                        Mono.just(resp.result)
+                    }
+                }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt

@@ -0,0 +1,79 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.upstream.*
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumHeadLagObserver
+import io.emeraldpay.grpc.Chain
+import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+
+class BitcoinChainUpstreams(
+        chain: Chain,
+        val upstreams: MutableList<BitcoinUpstream>,
+        caches: Caches,
+        objectMapper: ObjectMapper
+) : ChainUpstreams<BitcoinApi>(chain, upstreams as MutableList<Upstream<BitcoinApi>>, caches, objectMapper) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BitcoinChainUpstreams::class.java)
+    }
+
+    private var head: Head? = null
+
+    override fun init() {
+        if (upstreams.size > 0) {
+            head = updateHead()
+        }
+        super.init()
+    }
+
+    override fun updateHead(): Head {
+        head?.let {
+            if (it is Lifecycle) {
+                it.stop()
+            }
+        }
+        lagObserver?.stop()
+        lagObserver = null
+        val head = if (upstreams.size == 1) {
+            val upstream = upstreams.first()
+            upstream.setLag(0)
+            upstream.getHead()
+        } else {
+            val newHead = MergedHead(upstreams.map { it.getHead() }).apply {
+                this.start()
+            }
+//            val lagObserver = TODO
+//            this.lagObserver = lagObserver
+            newHead
+        }
+        onHeadUpdated(head)
+        return head
+    }
+
+    override fun setHead(head: Head) {
+        this.head = head
+    }
+
+    override fun getHead(): Head {
+        return head!!
+    }
+
+    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
+        return upstreams.flatMap { it.getLabels() }
+    }
+
+    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
+        if (!selfType.isAssignableFrom(this.javaClass)) {
+            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
+        }
+        if (!upstreamType.isAssignableFrom(BitcoinApi::class.java)) {
+            throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType")
+        }
+        return this as T
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcClient.kt

@@ -0,0 +1,54 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.config.AuthConfig
+import io.netty.buffer.Unpooled
+import io.netty.handler.codec.http.HttpHeaderNames
+import io.netty.handler.codec.http.HttpHeaders
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import reactor.netty.http.client.HttpClient
+import java.util.*
+import java.util.function.Consumer
+
+class BitcoinRpcClient(
+        private val target: String,
+        basicAuth: AuthConfig.ClientBasicAuth?
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BitcoinRpcClient::class.java)
+    }
+
+    private val httpClient: HttpClient
+
+    init {
+        var build = HttpClient.create()
+
+        build = build.headers { h ->
+            h.add(HttpHeaderNames.CONTENT_TYPE, "application/json")
+        }
+
+        basicAuth?.let { basicAuth ->
+            val authString: String = basicAuth.username + ":" + basicAuth.password
+            val authBase64 = Base64.getEncoder().encodeToString(authString.toByteArray())
+            val auth = "Basic $authBase64"
+            val headers = Consumer { h: HttpHeaders -> h.add(HttpHeaderNames.AUTHORIZATION, auth) }
+            build = build.headers(headers)
+        }
+
+        this.httpClient = build
+    }
+
+    fun execute(request: ByteArray): Mono<ByteArray> {
+        val response = httpClient
+                .post()
+                .uri(target)
+                .send(Mono.just(request).map { Unpooled.wrappedBuffer(it) })
+
+        return response.responseContent()
+                .aggregate()
+                .asByteArray()
+    }
+
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt

@@ -0,0 +1,63 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.upstream.AbstractHead
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead
+import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory
+import reactor.core.Disposable
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+import java.time.Duration
+import java.util.concurrent.Executors
+
+class BitcoinRpcHead(
+        private val api: BitcoinApi,
+        private val extractBlock: ExtractBlock,
+        private val interval: Duration = Duration.ofSeconds(15)
+) : Head, AbstractHead(), Lifecycle {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BitcoinRpcHead::class.java)
+        val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("bitcoin-rpc-head")))
+    }
+
+    private var refreshSubscription: Disposable? = null
+
+    override fun isRunning(): Boolean {
+        return refreshSubscription != null
+    }
+
+    override fun start() {
+        if (refreshSubscription != null) {
+            log.warn("Called to start when running")
+            return
+        }
+        val base = Flux.interval(interval)
+                .publishOn(scheduler)
+                .flatMap {
+                    api.executeAndResult(0, "getbestblockhash", emptyList(), String::class.java)
+                            .timeout(Defaults.timeout, Mono.error(Exception("Best block hash is not received")))
+                }
+                .distinctUntilChanged()
+                .flatMap { hash ->
+                    api.execute(0, "getblock", listOf(hash))
+                            .map(extractBlock::extract)
+                            .timeout(Defaults.timeout, Mono.error(Exception("Block data is not received")))
+                }
+                .onErrorContinue { err, _ ->
+                    log.debug("RPC error ${err.message}")
+                }
+        refreshSubscription = super.follow(base)
+    }
+
+    override fun stop() {
+        val copy = refreshSubscription
+        refreshSubscription = null
+        copy?.dispose()
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt

@@ -0,0 +1,91 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.upstream.*
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
+import io.emeraldpay.grpc.Chain
+import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
+import reactor.core.publisher.Mono
+
+class BitcoinUpstream(
+        id: String,
+        val chain: Chain,
+        private val api: BitcoinApi,
+        options: UpstreamsConfig.Options,
+        private val objectMapper: ObjectMapper
+) : DefaultUpstream<BitcoinApi>(id, options, DefaultBitcoinMethods()), Lifecycle {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java)
+    }
+
+    private val head: Head = createHead()
+    private var validatorSubscription: Disposable? = null
+
+    private fun createHead(): Head {
+        return BitcoinRpcHead(
+                api,
+                ExtractBlock(objectMapper)
+        )
+    }
+
+    override fun getHead(): Head {
+        return head
+    }
+
+    override fun getApi(matcher: Selector.Matcher): Mono<out BitcoinApi> {
+        return Mono.just(api)
+    }
+
+    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
+        return listOf(UpstreamsConfig.Labels())
+    }
+
+    override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
+        if (!selfType.isAssignableFrom(this.javaClass)) {
+            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
+        }
+        if (!upstreamType.isAssignableFrom(BitcoinApi::class.java)) {
+            throw ClassCastException("Cannot cast ${BitcoinApi::class.java} to $upstreamType")
+        }
+        return this as T
+    }
+
+    override fun isRunning(): Boolean {
+        var runningAny = validatorSubscription != null
+        if (head is Lifecycle) {
+            runningAny = runningAny || head.isRunning
+        }
+        return runningAny
+    }
+
+    override fun start() {
+        log.info("Configured for ${chain.chainName}")
+        if (head is Lifecycle) {
+            if (!head.isRunning) {
+                head.start()
+            }
+        }
+        validatorSubscription?.dispose()
+
+        if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
+            this.setLag(0)
+            this.setStatus(UpstreamAvailability.OK)
+        } else {
+            val validator = BitcoinUpstreamValidator(api, getOptions())
+            validatorSubscription = validator.start()
+                    .subscribe(this::setStatus)
+        }
+    }
+
+    override fun stop() {
+        if (head is Lifecycle) {
+            head.stop()
+        }
+        validatorSubscription?.dispose()
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt

@@ -0,0 +1,47 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.infinitape.etherjar.rpc.Commands
+import org.slf4j.LoggerFactory
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+import java.time.Duration
+import java.util.concurrent.Executors
+
+class BitcoinUpstreamValidator(
+        private val api: BitcoinApi,
+        private val options: UpstreamsConfig.Options
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BitcoinUpstreamValidator::class.java)
+        val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("bitcoin-validator")))
+    }
+
+    fun validate(): Mono<UpstreamAvailability> {
+        return api.executeAndResult(0, "getconnectioncount", emptyList(), Int::class.java)
+                .map { count ->
+                    val minPeers = options.minPeers ?: 1
+                    if (count < minPeers) {
+                        UpstreamAvailability.IMMATURE
+                    } else {
+                        UpstreamAvailability.OK
+                    }
+                }
+                .onErrorReturn(UpstreamAvailability.UNAVAILABLE)
+    }
+
+    fun start(): Flux<UpstreamAvailability> {
+        return Flux.interval(Duration.ofSeconds(15))
+                .subscribeOn(scheduler)
+                .flatMap {
+                    validate()
+                }
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/DefaultBitcoinMethods.kt

@@ -0,0 +1,10 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
+
+class DefaultBitcoinMethods : DirectCallMethods(
+        listOf("getbestblockhash", "getblock", "gettransaction")
+) {
+
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlock.kt

@@ -0,0 +1,40 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.TxId
+import org.apache.commons.codec.binary.Hex
+import org.slf4j.LoggerFactory
+import java.math.BigInteger
+import java.time.Instant
+
+class ExtractBlock(
+        private val objectMapper: ObjectMapper
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ExtractBlock::class.java)
+    }
+
+    fun extract(json: ByteArray): BlockContainer {
+        val data = objectMapper.readValue(json, Map::class.java) as Map<String, Any>
+
+        val height = data["height"] as Number? ?: throw IllegalArgumentException("Block JSON has no height")
+        val time = data["time"] as Number? ?: throw IllegalArgumentException("Block JSON has no time")
+        val hash = data["hash"] as String? ?: throw IllegalArgumentException("Block JSON has no hash")
+        val chainwork = data["chainwork"] as String? ?: throw IllegalArgumentException("Block JSON has no chainwork")
+        val transactions = (data["tx"] as List<String>?)?.map(TxId.Companion::from) ?: emptyList()
+
+        return BlockContainer(
+                height.toLong(),
+                BlockId.from(hash),
+                BigInteger(1, Hex.decodeHex(chainwork)),
+                Instant.ofEpochMilli(time.toLong() * 1000),
+                false,
+                json,
+                transactions
+        )
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DirectCallMethods.kt

@@ -22,10 +22,10 @@ import io.emeraldpay.dshackle.quorum.CallQuorum
  * Configuration that uses [AlwaysQuorum] for all available methods. The methods list itself
  * is provided in constructor (or empty otherwise)
  */
-class DirectCallMethods(private val methods: Set<String>) : CallMethods {
+open class DirectCallMethods(private val methods: Set<String>) : CallMethods {
 
-    constructor(): this(emptySet())
+    constructor() : this(emptySet())
-    constructor(methods: Collection<String>): this(methods.toSet())
+    constructor(methods: Collection<String>) : this(methods.toSet())
 
     override fun getQuorumFor(method: String): CallQuorum {
         return AlwaysQuorum()

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

@@ -1,49 +1,8 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.AbstractHead
-import org.slf4j.LoggerFactory
+import io.emeraldpay.dshackle.upstream.Head
-import reactor.core.Disposable
-import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import reactor.core.publisher.TopicProcessor
-import java.util.concurrent.atomic.AtomicReference
 
-open class DefaultEthereumHead: EthereumHead {
+open class DefaultEthereumHead : Head, AbstractHead() {
 
-    private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
-    private val head = AtomicReference<BlockContainer>(null)
-    private val stream: TopicProcessor<BlockContainer> = TopicProcessor.create()
-
-    fun follow(source: Flux<BlockContainer>): Disposable {
-        return source.distinctUntilChanged {
-            it.hash
-        }.filter { block ->
-            val curr = head.get()
-            curr == null || curr.difficulty < block.difficulty
-        }
-                .subscribe { block ->
-                    val prev = head.getAndUpdate { curr ->
-                        if (curr == null || curr.difficulty < block.difficulty) {
-                            block
-                        } else {
-                            curr
-                        }
-            }
-            if (prev == null || prev.hash != block.hash) {
-                log.debug("New block ${block.height} ${block.hash}")
-                stream.onNext(block)
-            }
-        }
-    }
-
-    override fun getFlux(): Flux<BlockContainer> {
-        return Flux.merge(
-                Mono.justOrEmpty(head.get()),
-                Flux.from(stream)
-        ).onBackpressureLatest()
-    }
-
-    fun getCurrent(): BlockContainer? {
-        return head.get()
-    }
 }

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

@@ -20,13 +20,8 @@ import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.grpc.Chain
-import io.infinitape.etherjar.rpc.json.BlockJson
-import io.infinitape.etherjar.rpc.json.TransactionJson
-import io.infinitape.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
-import java.lang.IllegalStateException
-import java.time.Duration
 
 class EthereumChainUpstreams(
         chain: Chain,
@@ -39,7 +34,7 @@ class EthereumChainUpstreams(
         private val log = LoggerFactory.getLogger(EthereumChainUpstreams::class.java)
     }
 
-    private var head: EthereumHead? = null
+    private var head: Head? = null
 
     init {
         this.init()
@@ -52,15 +47,15 @@ class EthereumChainUpstreams(
         super.init()
     }
 
-    override fun getHead(): EthereumHead {
+    override fun getHead(): Head {
         return head!!
     }
 
     override fun setHead(head: Head) {
-        this.head = head as EthereumHead
+        this.head = head
     }
 
-    override fun updateHead(): EthereumHead {
+    override fun updateHead(): Head {
         head?.let {
             if (it is Lifecycle) {
                 it.stop()
@@ -73,7 +68,7 @@ class EthereumChainUpstreams(
             upstream.setLag(0)
             upstream.getHead()
         } else {
-            val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }).apply {
+            val newHead = MergedHead(upstreams.map { it.getHead() }).apply {
                 this.start()
             }
             val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream<EthereumApi>>).apply {
@@ -90,25 +85,6 @@ class EthereumChainUpstreams(
         return upstreams.flatMap { it.getLabels() }
     }
 
-    override fun printStatus() {
-        var height: Long? = null
-        try {
-            height = getHead().getFlux().next().block(Duration.ofSeconds(1))?.height
-        } catch (e: IllegalStateException) {
-            //timout
-        } catch (e: Exception) {
-            log.warn("Head processing error: ${e.javaClass} ${e.message}")
-        }
-        val statuses = upstreams.map { it.getStatus() }
-                .groupBy { it }
-                .map { "${it.key.name}/${it.value.size}" }
-                .joinToString(",")
-        val lag = upstreams.map { it.getLag() }
-                .joinToString(", ")
-
-        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]")
-    }
-
     @SuppressWarnings("unchecked")
     override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
         if (!selfType.isAssignableFrom(this.javaClass)) {

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

@@ -1,24 +0,0 @@
-/**
- * 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.upstream.Head
-import io.infinitape.etherjar.domain.TransactionId
-import io.infinitape.etherjar.rpc.json.BlockJson
-import io.infinitape.etherjar.rpc.json.TransactionRefJson
-
-interface EthereumHead : Head {
-}

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

@@ -18,12 +18,13 @@ package io.emeraldpay.dshackle.upstream.ethereum
 import io.emeraldpay.dshackle.upstream.HeadLagObserver
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.Head
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import java.time.Duration
 
 class EthereumHeadLagObserver(
-        master: EthereumHead,
+        master: Head,
         followers: Collection<Upstream<EthereumApi>>
 ) : HeadLagObserver<EthereumApi>(master, followers) {
 

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

@@ -24,8 +24,6 @@ import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
 import io.emeraldpay.grpc.Chain
-import io.infinitape.etherjar.rpc.json.BlockJson
-import io.infinitape.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
@@ -33,15 +31,15 @@ import reactor.core.publisher.Mono
 import java.time.Duration
 
 open class EthereumUpstream(
-        private val id: String,
+        id: String,
         val chain: Chain,
         private val api: DirectEthereumApi,
         private val ethereumWs: EthereumWs? = null,
-        private val options: UpstreamsConfig.Options,
+        options: UpstreamsConfig.Options,
         val node: QuorumForLabels.QuorumItem,
-        private val targets: CallMethods,
+        targets: CallMethods,
         private val objectMapper: ObjectMapper
-) : DefaultUpstream<EthereumApi>(), Upstream<EthereumApi>, CachesEnabled, Lifecycle {
+) : DefaultUpstream<EthereumApi>(id, options, targets), Upstream<EthereumApi>, CachesEnabled, Lifecycle {
 
     constructor(id: String, chain: Chain, api: DirectEthereumApi, objectMapper: ObjectMapper) : this(id, chain, api, null,
             UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
@@ -50,7 +48,7 @@ open class EthereumUpstream(
 
     private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
 
-    private val head: EthereumHead = this.createHead()
+    private val head: Head = this.createHead()
     private var validatorSubscription: Disposable? = null
 
     init {
@@ -64,18 +62,14 @@ open class EthereumUpstream(
         }
     }
 
-    override fun getId(): String {
-        return id
-    }
-
     override fun start() {
         log.info("Configured for ${chain.chainName}")
 
-        if (options.disableValidation != null && options.disableValidation!!) {
+        if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
             this.setLag(0)
             this.setStatus(UpstreamAvailability.OK)
         } else {
-            val validator = UpstreamValidator(this, options)
+            val validator = EthereumUpstreamValidator(this, getOptions())
             validatorSubscription = validator.start()
                     .subscribe(this::setStatus)
         }
@@ -93,7 +87,7 @@ open class EthereumUpstream(
         }
     }
 
-    open fun createHead(): EthereumHead {
+    open fun createHead(): Head {
         return if (ethereumWs != null) {
             val ws = EthereumWsHead(ethereumWs).apply {
                 this.start()
@@ -102,22 +96,18 @@ open class EthereumUpstream(
             val rpc = EthereumRpcHead(api, objectMapper, Duration.ofSeconds(30)).apply {
                 this.start()
             }
-            EthereumHeadMerge(listOf(rpc, ws)).apply {
+            MergedHead(listOf(rpc, ws)).apply {
                 this.start()
             }
         } else {
-            log.warn("Setting up upstream $id with RPC-only access, less effective than WS+RPC")
+            log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC")
             EthereumRpcHead(api, objectMapper).apply {
                 this.start()
             }
         }
     }
 
-    override fun isAvailable(): Boolean {
+    override fun getHead(): Head {
-        return getStatus() == UpstreamAvailability.OK
-    }
-
-    override fun getHead(): EthereumHead {
         return head
     }
 
@@ -125,18 +115,10 @@ open class EthereumUpstream(
         return Mono.just(api)
     }
 
-    override fun getOptions(): UpstreamsConfig.Options {
-        return options
-    }
-
     override fun getLabels(): Collection<UpstreamsConfig.Labels> {
         return listOf(node.labels)
     }
 
-    override fun getMethods(): CallMethods {
-        return targets
-    }
-
     @Suppress("unchecked")
     override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, upstreamType: Class<TA>): T {
         if (!selfType.isAssignableFrom(this.javaClass)) {

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

@@ -13,11 +13,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.emeraldpay.dshackle.upstream
+package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.infinitape.etherjar.rpc.*
 import org.slf4j.LoggerFactory
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory
@@ -27,13 +28,13 @@ import reactor.core.scheduler.Schedulers
 import java.time.Duration
 import java.util.concurrent.Executors
 
-class UpstreamValidator(
+class EthereumUpstreamValidator(
         private val ethereumUpstream: EthereumUpstream,
         private val options: UpstreamsConfig.Options
 ) {
     companion object {
-        private val log = LoggerFactory.getLogger(UpstreamValidator::class.java)
+        private val log = LoggerFactory.getLogger(EthereumUpstreamValidator::class.java)
-        val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("validator")))
+        val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("ethereum-validator")))
     }
 
     fun validate(): Mono<UpstreamAvailability> {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt

@@ -33,7 +33,6 @@ import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead
 import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.rpc.*
@@ -59,13 +58,16 @@ open class EthereumGrpcUpstream(
         private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub,
         private val objectMapper: ObjectMapper,
         private val rpcClient: ReactorEmeraldClient
-) : DefaultUpstream<EthereumApi>(), CachesEnabled, Lifecycle {
+) : DefaultUpstream<EthereumApi>(
+        "$parentId/${chain.chainCode}",
+        UpstreamsConfig.Options.getDefaults(),
+        null
+), CachesEnabled, Lifecycle {
 
     private var allLabels: Collection<UpstreamsConfig.Labels> = ArrayList<UpstreamsConfig.Labels>()
     private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
     private var caches: Caches? = null
 
-    private val options = UpstreamsConfig.Options.getDefaults()
     private val nodes = AtomicReference<QuorumForLabels>(QuorumForLabels())
     private val head = DefaultEthereumHead()
     private var targets: CallMethods? = null
@@ -84,10 +86,6 @@ open class EthereumGrpcUpstream(
         }
     }
 
-    override fun getId(): String {
-        return "$parentId/${chain.chainCode}"
-    }
-
     override fun start() {
         if (this.isRunning) return
         val chainRef = Common.Chain.newBuilder()
@@ -200,12 +198,12 @@ open class EthereumGrpcUpstream(
     }
 
     override fun isAvailable(): Boolean {
-        return getStatus() == UpstreamAvailability.OK && head.getCurrent() != null && nodes.get().getAll().any {
+        return super.isAvailable() && head.getCurrent() != null && nodes.get().getAll().any {
             it.quorum > 0
         }
     }
 
-    override fun getHead(): EthereumHead {
+    override fun getHead(): Head {
         return head
     }
 
@@ -213,10 +211,6 @@ open class EthereumGrpcUpstream(
         return Mono.just(createApi(matcher))
     }
 
-    override fun getOptions(): UpstreamsConfig.Options {
-        return options
-    }
-
     override fun setCaches(caches: Caches) {
         this.caches = caches
     }
@@ -231,5 +225,4 @@ open class EthereumGrpcUpstream(
         }
         return this as T
     }
-
 }

src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy

@@ -77,9 +77,9 @@ class BlocksMemCacheSpec extends Specification {
         def act3 = cache.read(BlockId.from(hash3)).block()
         def act4 = cache.read(BlockId.from(hash4)).block()
         then:
-        act2.hash.toHex() == hash2
+        act2.hash.toHex() == hash2.substring(2)
-        act3.hash.toHex() == hash3
+        act3.hash.toHex() == hash3.substring(2)
-        act4.hash.toHex() == hash4
+        act4.hash.toHex() == hash4.substring(2)
         act1 == null
     }
 

src/test/groovy/io/emeraldpay/dshackle/cache/HeightCacheSpec.groovy

@@ -38,10 +38,10 @@ class HeightCacheSpec extends Specification {
         def act3 = cache.read(102).block()
         def act4 = cache.read(103).block()
         then:
-        act1.toHex() == hash1
+        act1.toHex() == hash1.substring(2)
-        act2.toHex() == hash2
+        act2.toHex() == hash2.substring(2)
-        act3.toHex() == hash3
+        act3.toHex() == hash3.substring(2)
-        act4.toHex() == hash4
+        act4.toHex() == hash4.substring(2)
     }
 
     def "Keeps only configured amount"() {
@@ -65,8 +65,8 @@ class HeightCacheSpec extends Specification {
         def act4 = cache.read(103).block()
         then:
         act1 == null
-        act2.toHex() == hash2
+        act2.toHex() == hash2.substring(2)
-        act3.toHex() == hash3
+        act3.toHex() == hash3.substring(2)
-        act4.toHex() == hash4
+        act4.toHex() == hash4.substring(2)
     }
 }

src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy

@@ -57,9 +57,9 @@ class TxMemCacheSpec extends Specification {
         def act3 = cache.read(TxId.from(hash3)).block()
         def act4 = cache.read(TxId.from(hash4)).block()
         then:
-        act2.hash.toHex() == hash2
+        act2.hash.toHex() == hash2.substring(2)
-        act3.hash.toHex() == hash3
+        act3.hash.toHex() == hash3.substring(2)
-        act4.hash.toHex() == hash4
+        act4.hash.toHex() == hash4.substring(2)
         act1 == null
     }
 
@@ -93,8 +93,8 @@ class TxMemCacheSpec extends Specification {
         then:
         act1 == null
         act2 == null
-        act3.hash.toHex() == hash3
+        act3.hash.toHex() == hash3.substring(2)
-        act4.hash.toHex() == hash4
+        act4.hash.toHex() == hash4.substring(2)
     }
 
     def "Evict all by block data"() {
@@ -137,7 +137,7 @@ class TxMemCacheSpec extends Specification {
         then:
         act1 == null
         act2 == null
-        act3.hash.toHex() == hash3
+        act3.hash.toHex() == hash3.substring(2)
-        act4.hash.toHex() == hash4
+        act4.hash.toHex() == hash4.substring(2)
     }
 }

src/test/groovy/io/emeraldpay/dshackle/data/BlockIdSpec.groovy

@@ -0,0 +1,30 @@
+package io.emeraldpay.dshackle.data
+
+import io.infinitape.etherjar.domain.BlockHash
+import spock.lang.Specification
+
+class BlockIdSpec extends Specification {
+
+    def "Create from string"() {
+        expect:
+        BlockId.from(source).toHex() == exp
+        where:
+        exp                                                                | source
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "A0E65CBC1B52A8CA60562112C6060552D882F16F34A9DBA2CCDC05C0A6A27100"
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0xA0E65CBC1B52A8CA60562112C6060552D882F16F34A9DBA2CCDC05C0A6A27100"
+        "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0x00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+    }
+
+    def "Create from block hash"() {
+        expect:
+        BlockId.from(BlockHash.from(source)).toHex() == exp
+        where:
+        exp                                                                | source
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0x00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/data/TxIdSpec.groovy

@@ -0,0 +1,29 @@
+package io.emeraldpay.dshackle.data
+
+import io.infinitape.etherjar.domain.TransactionId
+import spock.lang.Specification
+
+class TxIdSpec extends Specification {
+
+    def "Create from string"() {
+        expect:
+        TxId.from(source).toHex() == exp
+        where:
+        exp                                                                | source
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "A0E65CBC1B52A8CA60562112C6060552D882F16F34A9DBA2CCDC05C0A6A27100"
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0xA0E65CBC1B52A8CA60562112C6060552D882F16F34A9DBA2CCDC05C0A6A27100"
+        "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0x00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+    }
+
+    def "Create from block hash"() {
+        expect:
+        TxId.from(TransactionId.from(source)).toHex() == exp
+        where:
+        exp                                                                | source
+        "a0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+        "00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100" | "0x00e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27100"
+    }
+}

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

@@ -16,12 +16,12 @@
 package io.emeraldpay.dshackle.test
 
 import io.emeraldpay.dshackle.data.BlockContainer
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
+import io.emeraldpay.dshackle.upstream.Head
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.TopicProcessor
 
-class EthereumHeadMock implements EthereumHead {
+class EthereumHeadMock implements Head {
 
     private TopicProcessor<BlockContainer> bus = TopicProcessor.create()
     private BlockContainer latest

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

@@ -17,11 +17,11 @@ package io.emeraldpay.dshackle.test
 
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.grpc.Chain
@@ -58,12 +58,12 @@ class EthereumUpstreamMock extends EthereumUpstream {
     }
 
     @Override
-    EthereumHead createHead() {
+    Head createHead() {
         return ethereumHeadMock
     }
 
     @Override
-    EthereumHead getHead() {
+    Head getHead() {
         return ethereumHeadMock
     }
 }

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

@@ -21,15 +21,15 @@ import io.grpc.inprocess.InProcessChannelBuilder
 import io.grpc.inprocess.InProcessServerBuilder
 import io.grpc.testing.GrpcCleanupRule
 
-class MockServer {
+class MockGrpcServer {
 
     GrpcCleanupRule grpcCleanup = new GrpcCleanupRule()
 
-    ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(ReactorBlockchainGrpc.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())
+        def channel = grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build())
         return ReactorBlockchainGrpc.newReactorStub(channel)
     }
 

src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy

@@ -10,7 +10,6 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.test.TestingCommons
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
@@ -30,7 +29,7 @@ class CachingEthereumApiSpec extends Specification {
 
     def "Get blockNumber from head"() {
         setup:
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.default(objectMapper),
@@ -58,7 +57,7 @@ class CachingEthereumApiSpec extends Specification {
 
     def "Return empty if block is not cached"() {
         setup:
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.default(objectMapper),
@@ -76,7 +75,7 @@ class CachingEthereumApiSpec extends Specification {
     def "Return block by hash when cached"() {
         setup:
         def cache = new BlocksMemCache();
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.newBuilder().setObjectMapper(objectMapper).setBlockByHash(cache).build(),
@@ -106,7 +105,7 @@ class CachingEthereumApiSpec extends Specification {
         setup:
         def blocksCache = new BlocksMemCache()
         def heightCache = new HeightCache()
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.newBuilder().setObjectMapper(objectMapper).setBlockByHash(blocksCache).setBlockByHeight(heightCache).build(),
@@ -135,7 +134,7 @@ class CachingEthereumApiSpec extends Specification {
         setup:
         def blocksCache = Mock(BlocksMemCache)
         def txCache = Mock(TxMemCache)
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.newBuilder().setObjectMapper(objectMapper).setBlockByHash(blocksCache).setTxByHash(txCache).build(),
@@ -161,7 +160,7 @@ class CachingEthereumApiSpec extends Specification {
         setup:
         def blocksCache = Mock(BlocksMemCache)
         def txCache = Mock(TxMemCache)
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.newBuilder().setObjectMapper(objectMapper).setBlockByHash(blocksCache).setTxByHash(txCache).build(),
@@ -191,7 +190,7 @@ class CachingEthereumApiSpec extends Specification {
         def blocksCache = Mock(BlocksMemCache)
         def txCache = Mock(TxMemCache)
         def heightCache = Mock(HeightCache)
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.newBuilder().setObjectMapper(objectMapper).setBlockByHash(blocksCache).setTxByHash(txCache).setBlockByHeight(heightCache).build(),
@@ -219,7 +218,7 @@ class CachingEthereumApiSpec extends Specification {
         def blocksCache = Mock(BlocksMemCache)
         def txCache = Mock(TxMemCache)
         def heightCache = Mock(HeightCache)
-        def head = Mock(EthereumHead.class)
+        def head = Mock(Head.class)
         def api = new CachingEthereumApi(
                 objectMapper,
                 Caches.newBuilder().setObjectMapper(objectMapper).setBlockByHash(blocksCache).setTxByHash(txCache).setBlockByHeight(heightCache).build(),

src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinApiSpec.groovy

@@ -0,0 +1,108 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.infinitape.etherjar.rpc.RpcException
+import org.mockserver.integration.ClientAndServer
+import org.mockserver.model.HttpRequest
+import org.mockserver.model.HttpResponse
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+
+class BitcoinApiSpec extends Specification {
+
+    ClientAndServer mockServer
+    BitcoinApi api
+
+    def setup() {
+        mockServer = ClientAndServer.startClientAndServer(18332);
+        api = new BitcoinApi(
+                new BitcoinRpcClient("localhost:18332", null),
+                TestingCommons.objectMapper()
+        )
+    }
+
+    def cleanup() {
+        mockServer.stop()
+    }
+
+    def "Request simple"() {
+        setup:
+        def resp = '{' +
+                '  "result": "0000000000000000000889c2e52ca5e1cecac60bce9a3754201a7a9a67791e90",' +
+                '  "error": null,' +
+                '  "id": 15' +
+                '}'
+        mockServer.when(
+                HttpRequest.request()
+        ).respond(
+                HttpResponse.response(resp)
+        )
+        when:
+        def act = api.executeAndResult(15, "getbestblockhash", [], String)
+        then:
+        StepVerifier.create(act)
+                .expectNext("0000000000000000000889c2e52ca5e1cecac60bce9a3754201a7a9a67791e90")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+        mockServer.verify(
+                HttpRequest.request()
+                        .withMethod("POST")
+                        .withBody('{"jsonrpc":"2.0","method":"getbestblockhash","params":[],"id":15}')
+        )
+    }
+
+    def "Request with params"() {
+        setup:
+        def resp = '{' +
+                '  "result": "something",' +
+                '  "id": 1' +
+                '}'
+        mockServer.when(
+                HttpRequest.request()
+        ).respond(
+                HttpResponse.response(resp)
+        )
+        when:
+        def act = api.executeAndResult(1, "getsomething", ["something", false], String)
+        then:
+        StepVerifier.create(act)
+                .expectNext("something")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+        mockServer.verify(
+                HttpRequest.request()
+                        .withMethod("POST")
+                        .withBody('{"jsonrpc":"2.0","method":"getsomething","params":["something",false],"id":1}')
+        )
+    }
+
+    def "Returns error"() {
+        setup:
+        def resp = '{' +
+                '  "result": null,' +
+                '  "error": {' +
+                '    "code": -32601,' +
+                '    "message": "Method not found"' +
+                '  },' +
+                '  "id": 1' +
+                '}'
+        mockServer.when(
+                HttpRequest.request()
+        ).respond(
+                HttpResponse.response(resp)
+        )
+        when:
+        def act = api.executeAndResult(1, "geterror", [], String)
+        then:
+        StepVerifier.create(act)
+                .expectError(RpcException)
+                .verify(Duration.ofSeconds(1))
+        mockServer.verify(
+                HttpRequest.request()
+                        .withMethod("POST")
+                        .withBody('{"jsonrpc":"2.0","method":"geterror","params":[],"id":1}')
+        )
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcClientSpec.groovy

@@ -0,0 +1,86 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.config.AuthConfig
+import org.mockserver.integration.ClientAndServer
+import org.mockserver.matchers.Times
+import org.mockserver.model.HttpRequest
+import org.mockserver.model.HttpResponse
+import org.mockserver.model.MediaType
+import org.mockserver.verify.VerificationTimes
+import reactor.test.StepVerifier
+import spock.lang.Shared
+import spock.lang.Specification
+
+import java.time.Duration
+
+class BitcoinRpcClientSpec extends Specification {
+
+    ClientAndServer mockServer
+
+    def setup() {
+        mockServer = ClientAndServer.startClientAndServer(18332);
+    }
+
+    def cleanup() {
+        mockServer.stop()
+    }
+
+    def "Make request"() {
+        setup:
+        def client = new BitcoinRpcClient("localhost:18332", null)
+
+        mockServer.when(
+                HttpRequest.request()
+                        .withMethod("POST")
+                        .withBody("ping"),
+                Times.exactly(1)
+        ).respond(
+                HttpResponse.response()
+                        .withBody("pong")
+        )
+        when:
+        def act = client.execute("ping".bytes).map { new String(it) }
+        then:
+        StepVerifier.create(act)
+                .expectNext("pong")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+        mockServer.verify(
+                HttpRequest.request()
+                        .withMethod("POST")
+                        .withBody("ping")
+                        .withContentType(MediaType.APPLICATION_JSON)
+        )
+
+    }
+
+    def "Make request with basic auth"() {
+        setup:
+        def auth = new AuthConfig.ClientBasicAuth("user", "passwd")
+        def client = new BitcoinRpcClient("localhost:18332", auth)
+
+        mockServer.when(
+                HttpRequest.request()
+                        .withMethod("POST")
+                        .withBody("ping")
+        ).respond(
+                HttpResponse.response()
+                        .withBody("pong")
+        )
+        when:
+        def act = client.execute("ping".bytes).map { new String(it) }
+        then:
+        StepVerifier.create(act)
+                .expectNext("pong")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+        mockServer.verify(
+                HttpRequest.request()
+                        .withMethod("POST")
+                        .withBody("ping")
+                        .withContentType(MediaType.APPLICATION_JSON)
+                        .withHeader("authorization", "Basic dXNlcjpwYXNzd2Q=")
+        )
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy

@@ -0,0 +1,95 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import de.jodamob.kotlin.testrunner.OpenedClasses
+import de.jodamob.kotlin.testrunner.SpotlinTestRunner
+import io.emeraldpay.dshackle.test.TestingCommons
+import org.junit.runner.RunWith
+import org.mockserver.integration.ClientAndServer
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+
+@RunWith(SpotlinTestRunner)
+@OpenedClasses(BitcoinApi)
+class BitcoinRpcHeadSpec extends Specification {
+
+    ClientAndServer mockServer
+
+    def "Follow 2 blocks created over 3 requests"() {
+        setup:
+        String hash1 = "0000000000000000000cf5a5d4dfc4347c0c1a863ec5fdb429b02b2162e50001"
+        String hash2 = "0000000000000000000cf5a5d4dfc4347c0c1a863ec5fdb429b02b2162e50002"
+
+        def block1 = """
+            {
+                "hash": "${hash1}",
+                "confirmations": 2,
+                "strippedsize": 800464,
+                "size": 1591897,
+                "weight": 3993289,
+                "height": 626472,
+                "version": 549453824,
+                "versionHex": "20c00000",
+                "merkleroot": "7835a26b6c9316232f8194c6b64b0a366e26a9a14e6a7e7409c34594d98a83c2",
+                "tx": [],
+                "time": 1587171526,
+                "mediantime": 1587168930,
+                "nonce": 4111305375,
+                "bits": "171320bc",
+                "difficulty": 14715214060656.53,
+                "chainwork": "00000000000000000000000000000000000000000e9baec5f63190a2b0bbcf6b",
+                "nTx": 0,
+                "previousblockhash": "000000000000000000106b8cb739dd3c412613aa5d459a739794ec2572a74c10"
+            }        
+        """
+        def block2 = """
+            {
+                "hash": "${hash2}",
+                "confirmations": 2,
+                "strippedsize": 800464,
+                "size": 1591897,
+                "weight": 3993289,
+                "height": 626473,
+                "version": 549453824,
+                "versionHex": "20c00000",
+                "merkleroot": "7835a26b6c9316232f8194c6b64b0a366e26a9a14e6a7e7409c34594d98a83c2",
+                "tx": [],
+                "time": 1587172526,
+                "mediantime": 1587168930,
+                "nonce": 4111305375,
+                "bits": "171320bc",
+                "difficulty": 14715214060656.53,
+                "chainwork": "00000000000000000000000000000000000000000e9baec5f63190a2b0bbcf6c",
+                "nTx": 0,
+                "previousblockhash": "${hash1}"
+            }        
+        """
+
+        BitcoinApi api = Mock(BitcoinApi) {
+            _ * executeAndResult(_, "getbestblockhash", _, String) >>> [
+                    Mono.just(hash1), Mono.just(hash1), Mono.just(hash2)
+            ]
+            _ * execute(_, "getblock", [hash1]) >> Mono.just(block1.bytes)
+            _ * execute(_, "getblock", [hash2]) >> Mono.just(block2.bytes)
+        }
+        BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(TestingCommons.objectMapper()), Duration.ofMillis(200))
+
+        when:
+        def act = head.flux.take(2)
+        head.start()
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { block ->
+                    block.hash.toHex() == hash1
+                }.as("Block 1")
+                .expectNextMatches { block ->
+                    block.hash.toHex() == hash2
+                }.as("Block 2")
+                .expectComplete()
+                .verify(Duration.ofSeconds(3))
+
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlockSpec.groovy

@@ -0,0 +1,24 @@
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import io.emeraldpay.dshackle.test.TestingCommons
+import spock.lang.Specification
+
+class ExtractBlockSpec extends Specification {
+
+    ExtractBlock extractBlock = new ExtractBlock(TestingCommons.objectMapper())
+
+    def "Extract standard block"() {
+        setup:
+        def json = this.class.getClassLoader().getResourceAsStream("bitcoin/block-626472.json").bytes
+        when:
+        def act = extractBlock.extract(json)
+        then:
+        act.hash.toHex() == "0000000000000000000889c2e52ca5e1cecac60bce9a3754201a7a9a67791e90"
+        act.height == 626472
+        act.timestamp.toString() == "2020-04-18T00:58:46Z"
+        act.difficulty.toString(16) == "e9baec5f63190a2b0bbcf6b"
+        !act.full
+        act.transactions.size() == 1487
+        act.json == json
+    }
+}

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

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.HeadLagObserver
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.infinitape.etherjar.domain.BlockHash
@@ -37,10 +38,10 @@ class EthereumHeadLagObserverSpec extends Specification {
 
     def "Updates lag distance"() {
         setup:
-        EthereumHead master = Mock()
+        Head master = Mock()
 
-        EthereumHead head1 = Mock()
+        Head head1 = Mock()
-        EthereumHead head2 = Mock()
+        Head head2 = Mock()
 
         Upstream up1 = Mock {
             _ * getHead() >> head1
@@ -89,7 +90,7 @@ class EthereumHeadLagObserverSpec extends Specification {
 
     def "Probes until there is no difference"() {
         setup:
-        EthereumHead master = Mock()
+        Head master = Mock()
         HeadLagObserver observer = new EthereumHeadLagObserver(master, [])
         Upstream up = Mock()
 
@@ -118,7 +119,7 @@ class EthereumHeadLagObserverSpec extends Specification {
 
     def "Correct distance"() {
         setup:
-        EthereumHead master = Mock()
+        Head master = Mock()
         HeadLagObserver observer = new EthereumHeadLagObserver(master, [])
         expect:
         def top = new BlockJson().with {

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy

@@ -21,7 +21,7 @@ import io.emeraldpay.api.proto.BlockchainGrpc
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.data.BlockId
-import io.emeraldpay.dshackle.test.MockServer
+import io.emeraldpay.dshackle.test.MockGrpcServer
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.grpc.Chain
@@ -38,7 +38,7 @@ import java.util.concurrent.CompletableFuture
 
 class EthereumGrpcUpstreamSpec extends Specification {
 
-    MockServer mockServer = new MockServer()
+    MockGrpcServer mockServer = new MockGrpcServer()
     ObjectMapper objectMapper = TestingCommons.objectMapper()
 
     def "Subscribe to head"() {