diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 6d7d64df..3a9a082d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -52,7 +52,6 @@ abstract class Multistream( val chain: Chain, private val upstreams: MutableList, val caches: Caches, - val postprocessor: RequestPostprocessor ) : Upstream, Lifecycle { companion object { @@ -144,23 +143,6 @@ abstract class Multistream( return FilteredApis(chain, upstreams, matcher, i) } - /** - * Finds an API that executed directly on a remote. - */ - open fun getDirectApi(matcher: Selector.Matcher): Mono> { - val apis = getApiSource(matcher) - apis.request(1) - return Mono.from(apis) - .map(Upstream::getApi) - .map { - RequestPostprocessor.wrap( - it, - postprocessor - ) - } // TODO do it on upstream init, not each time it's called - .switchIfEmpty(Mono.error(Exception("No API available for $chain"))) - } - abstract fun getFeeEstimation(): ChainFees /** diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt deleted file mode 100644 index 69ec2693..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt +++ /dev/null @@ -1,40 +0,0 @@ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import reactor.core.publisher.Mono - -interface RequestPostprocessor { - - fun onReceive(method: String, params: List, json: ByteArray) - - class Empty : RequestPostprocessor { - override fun onReceive(method: String, params: List, json: ByteArray) {} - } - - companion object { - fun wrap( - reader: Reader, - processor: RequestPostprocessor - ): Reader { - return Wrapper(reader, processor) - } - } - - class Wrapper( - private val reader: Reader, - private val processor: RequestPostprocessor - ) : Reader { - - override fun read(key: JsonRpcRequest): Mono { - return reader.read(key) - .doOnNext { - if (it.hasResult()) { - val result = it.getResult() - processor.onReceive(key.method, key.params, result) - } - } - } - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index 3393fd82..1cfe55a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -32,8 +32,8 @@ import reactor.core.publisher.Mono open class BitcoinMultistream( chain: Chain, private val sourceUpstreams: MutableList, - caches: Caches -) : Multistream(chain, sourceUpstreams as MutableList, caches, RequestPostprocessor.Empty()), Lifecycle { + caches: Caches, +) : Multistream(chain, sourceUpstreams as MutableList, caches), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java) @@ -95,6 +95,16 @@ open class BitcoinMultistream( onHeadUpdated(head) return head } + /** + * Finds an API that executed directly on a remote. + */ + open fun getDirectApi(matcher: Selector.Matcher): Mono> { + val apis = getApiSource(matcher) + apis.request(1) + return Mono.from(apis) + .map(Upstream::getApi) + .switchIfEmpty(Mono.error(Exception("No API available for $chain"))) + } override fun getRoutedApi(matcher: Selector.Matcher): Mono> { return Mono.just(callRouter) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt deleted file mode 100644 index 83e5d142..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt +++ /dev/null @@ -1,60 +0,0 @@ -/** - * Copyright (c) 2020 EmeraldPay, Inc - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.DefaultContainer -import io.emeraldpay.dshackle.data.TxId -import io.emeraldpay.dshackle.upstream.RequestPostprocessor -import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson -import org.slf4j.LoggerFactory - -class CacheRequested( - private val caches: Caches -) : RequestPostprocessor { - - companion object { - private val log = LoggerFactory.getLogger(CacheRequested::class.java) - } - - override fun onReceive(method: String, params: List, json: ByteArray) { - try { - if (method == "eth_getTransactionReceipt") { - cacheTxReceipt(params, json) - } - } catch (e: Throwable) { - log.warn("Failed to cache result", e) - } - } - - fun cacheTxReceipt(params: List, json: ByteArray) { - if (params.size != 1) { - return - } - // note: json could be a `null` value - val parsed = Global.objectMapper.readValue(json, TransactionReceiptJson::class.java) ?: return - val value = DefaultContainer( - TxId.from(parsed.transactionHash), - BlockId.from(parsed.blockHash), - parsed.blockNumber, - json, - parsed - ) - caches.cacheReceipt(Caches.Tag.REQUESTED, value) - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index a1c984a7..67f0cf5c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -37,7 +37,7 @@ open class EthereumMultistream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)), EthereumLikeMultistream { +) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { private val log = LoggerFactory.getLogger(EthereumMultistream::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index abe8cd0a..9d797483 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -37,7 +37,7 @@ open class EthereumPosMultiStream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)), EthereumLikeMultistream { +) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { private val log = LoggerFactory.getLogger(EthereumPosMultiStream::class.java) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index fb36e45b..fc8fe7b1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -180,26 +180,6 @@ class MultistreamSpec extends Specification { !act } - def "Call postprocess after api use"() { - setup: - def request = new JsonRpcRequest("test_foo", [1], 1, null) - - def api = TestingCommons.api() - api.answer("test_foo", [1], "test") - def postprocessor = Mock(RequestPostprocessor) - def up = TestingCommons.upstream(api) - def multistream = new TestMultistream([up], postprocessor) - - when: - def rdr = multistream.getDirectApi(Selector.empty).block(Duration.ofSeconds(1)) - def act = rdr.read(request).block(Duration.ofSeconds(1)) - - then: - act != null - act.hasResult() - act.resultAsProcessedString == "test" - 1 * postprocessor.onReceive("test_foo", [1], "\"test\"".bytes) - } def "Filter upstream matching selector single"() { setup: @@ -341,48 +321,6 @@ class MultistreamSpec extends Specification { .expectComplete() .verify(Duration.ofSeconds(1)) } - - class TestMultistream extends Multistream { - - TestMultistream(List upstreams, @NotNull RequestPostprocessor postprocessor) { - super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor) - } - - @Override - Mono> getRoutedApi(@NotNull Selector.Matcher matcher) { - return null - } - - @Override - Head updateHead() { - return null - } - - @Override - void setHead(@NotNull Head head) { - - } - - @Override - Head getHead() { - return null - } - - @Override - Collection getLabels() { - return null - } - - public T cast(Class selfType) { - return this - } - - @Override - ChainFees getFeeEstimation() { - return null - } - } - class TestEthereumPosMultistream extends EthereumPosMultiStream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy deleted file mode 100644 index bf73f37d..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy +++ /dev/null @@ -1,47 +0,0 @@ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import reactor.core.publisher.Mono -import spock.lang.Specification - -import java.time.Duration - -class RequestPostprocessorSpec extends Specification { - - def "Wrappers calls onReceive for a value"() { - setup: - def request = new JsonRpcRequest("test_foo", [1], 1, null) - def processor = Mock(RequestPostprocessor) - def api = TestingCommons.api() - api.answer("test_foo", [1], "test") - def wrapped = new RequestPostprocessor.Wrapper(api, processor) - - when: - def act = wrapped.read(request).block(Duration.ofSeconds(1)) - - then: - act.hasResult() - act.resultAsProcessedString == "test" - 1 * processor.onReceive("test_foo", [1], "\"test\"".bytes) - } - - def "Wrappers doesn't call onReceive for no value"() { - setup: - def request = new JsonRpcRequest("test_foo", [1], 1, null) - def processor = Mock(RequestPostprocessor) - Reader reader = Mock(Reader) { - 1 * it.read(request) >> Mono.empty() - } - def wrapped = new RequestPostprocessor.Wrapper(reader, processor) - - when: - def act = wrapped.read(request).block(Duration.ofSeconds(1)) - - then: - act == null - 0 * processor.onReceive(_, _, _) - } -} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy deleted file mode 100644 index 2b93c5dc..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy +++ /dev/null @@ -1,58 +0,0 @@ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.data.DefaultContainer -import spock.lang.Specification - -class CacheRequestedSpec extends Specification { - - def "Do nothing if unsupported method"() { - setup: - def caches = Mock(Caches) - CacheRequested instance = new CacheRequested(caches) - when: - instance.onReceive("eth_hashrate", [], '"0x38a"'.bytes) - then: - 0 * caches._(*_) - } - - def "Caches tx receipt"() { - setup: - def caches = Mock(Caches) - CacheRequested instance = new CacheRequested(caches) - def json = '''{ - "blockHash": "0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae", - "blockNumber": "0x213b8a", - "contractAddress": null, - "cumulativeGasUsed": "0x5208", - "gasUsed": "0x5208", - "logs": [], - "transactionHash": "0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197", - "transactionIndex": "0x00" - }'''.bytes - - when: - instance.onReceive("eth_getTransactionReceipt", ["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json) - - then: - 1 * caches.cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer it -> - it.height == 0x213b8a && - it.txId.toHex() == "5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197" && - it.json == json - }) - } - - def "Ignore null"() { - setup: - def caches = Mock(Caches) - CacheRequested instance = new CacheRequested(caches) - def json = 'null'.bytes - - when: - instance.cacheTxReceipt(["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json) - - then: - 0 * caches.cacheReceipt(_, _) - } - -}