From adfc35e994343e2aedf13f5e7d3704157ffc961e Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 5 May 2021 14:39:43 -0400 Subject: [PATCH] problem: Spring Reactor is too outdated --- build.gradle | 10 +++-- gradle.properties | 8 ++-- .../emeraldpay/dshackle/proxy/ProxyServer.kt | 6 +-- .../quorum/QuorumRpcReaderSpec.groovy | 8 ++++ .../rpc/TrackBitcoinAddressSpec.groovy | 10 ++--- .../dshackle/test/EthereumHeadMock.groovy | 8 ++-- .../dshackle/upstream/AbstractHeadSpec.groovy | 40 +++++++++---------- .../dshackle/upstream/FilteredApisSpec.groovy | 17 ++++---- .../upstream/HeadLagObserverSpec.groovy | 8 ++-- 9 files changed, 62 insertions(+), 53 deletions(-) diff --git a/build.gradle b/build.gradle index 2afd456e..36b91b45 100644 --- a/build.gradle +++ b/build.gradle @@ -45,6 +45,7 @@ java { repositories { mavenLocal() mavenCentral() + jcenter() maven { url "https://repo.spring.io/snapshot" } maven { url "https://repo.spring.io/milestone" } maven { url "https://dl.bintray.com/infinitape/etherjar" } @@ -61,7 +62,7 @@ configurations { } dependencies { - implementation "io.emeraldpay:emerald-api:0.9.1" + implementation "io.emeraldpay:emerald-api:0.9.2" implementation "io.grpc:grpc-protobuf:${grpcVersion}" implementation "io.grpc:grpc-stub:${grpcVersion}" @@ -83,15 +84,16 @@ dependencies { implementation "org.springframework.security:spring-security-web:$springSecurtyVersion" implementation "org.springframework.security:spring-security-config:$springSecurtyVersion" implementation "io.projectreactor:reactor-core:$reactorVersion" - implementation "io.projectreactor.netty:reactor-netty:0.9.7.RELEASE" - implementation 'io.projectreactor.addons:reactor-extra:3.3.3.RELEASE' - implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.2.RELEASE' + implementation "io.projectreactor.netty:reactor-netty:1.0.6" + implementation 'io.projectreactor.addons:reactor-extra:3.4.3' + implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.1.3' implementation "com.salesforce.servicelibs:reactor-grpc-stub:$reactiveGrpcVersion" implementation 'io.micrometer:micrometer-registry-prometheus:1.5.6' implementation 'io.lettuce:lettuce-core:5.2.2.RELEASE' implementation "io.infinitape:etherjar-domain:$etherjarVersion" implementation "io.infinitape:etherjar-hex:$etherjarVersion" + implementation "io.infinitape:etherjar-rpc-api:$etherjarVersion" implementation "io.infinitape:etherjar-rpc-http:$etherjarVersion" implementation "io.infinitape:etherjar-rpc-ws:$etherjarVersion" implementation "io.infinitape:etherjar-tx:$etherjarVersion" diff --git a/gradle.properties b/gradle.properties index 014e57ae..da138fb5 100644 --- a/gradle.properties +++ b/gradle.properties @@ -7,10 +7,10 @@ slf4jVersion=1.7.25 jacksonVersion=2.11.0 grpcVersion=1.25.0 reactiveGrpcVersion=1.0.1 -springBootVersion=2.3.5.RELEASE -springVersion=5.2.6.RELEASE -springSecurtyVersion=5.3.2.RELEASE -reactorVersion=3.3.5.RELEASE +springBootVersion=2.4.5 +springVersion=5.3.6 +springSecurtyVersion=5.4.6 +reactorVersion=3.4.5 nettyVersion=4.1.53.Final # Our Libs etherjarVersion=0.10.2 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index 6de95683..f9e37edd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -92,10 +92,8 @@ class ProxyServer( } log.info("Listening Proxy on ${config.host}:${config.port}") var serverBuilder = HttpServer.create() - .tcpConfiguration { - it.bootstrap { b -> - b.handler(errorHandler) - } + .doOnChannelInit { _, channel, _ -> + channel.pipeline().addFirst(errorHandler) } .host(config.host) .port(config.port) diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index 4849446b..6b608895 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.RpcException import reactor.core.publisher.Mono import reactor.test.StepVerifier @@ -43,6 +44,7 @@ class QuorumRpcReaderSpec extends Specification { } } def apis = new FilteredApis( + Chain.ETHEREUM, [up], Selector.empty ) def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) @@ -73,6 +75,7 @@ class QuorumRpcReaderSpec extends Specification { _ * getApi() >> api } def apis = new FilteredApis( + Chain.ETHEREUM, [up], Selector.empty ) def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) @@ -104,6 +107,7 @@ class QuorumRpcReaderSpec extends Specification { } } def apis = new FilteredApis( + Chain.ETHEREUM, [up], Selector.empty ) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) @@ -135,6 +139,7 @@ class QuorumRpcReaderSpec extends Specification { } } def apis = new FilteredApis( + Chain.ETHEREUM, [up], Selector.empty ) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) @@ -166,6 +171,7 @@ class QuorumRpcReaderSpec extends Specification { } } def apis = new FilteredApis( + Chain.ETHEREUM, [up], Selector.empty ) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) @@ -198,6 +204,7 @@ class QuorumRpcReaderSpec extends Specification { _ * getApi() >> api } def apis = new FilteredApis( + Chain.ETHEREUM, [up], Selector.empty ) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3)) @@ -227,6 +234,7 @@ class QuorumRpcReaderSpec extends Specification { } } def apis = new FilteredApis( + Chain.ETHEREUM, [up], Selector.empty ) def reader = new QuorumRpcReader(apis, new NotLaggingQuorum(1)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy index c62b8081..59a9008d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy @@ -36,7 +36,7 @@ import org.bitcoinj.params.MainNetParams import org.bitcoinj.params.TestNet3Params import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.publisher.TopicProcessor +import reactor.core.publisher.Sinks import reactor.test.StepVerifier import spock.lang.Specification @@ -267,13 +267,13 @@ class TrackBitcoinAddressSpec extends Specification { def "Get update for a balance"() { setup: - def blocks = TopicProcessor.create() + def blocks = Sinks.many().unicast().onBackpressureBuffer() Head head = Mock(Head) { 1 * getFlux() >> Flux.concat( Flux.just( new BlockContainer(0L, BlockId.from(hash1), BigInteger.ZERO, Instant.now(), false, null, null, []) ), - Flux.from(blocks) + blocks.asFlux() ) } def upstream = null @@ -312,11 +312,11 @@ class TrackBitcoinAddressSpec extends Specification { StepVerifier.create(resp) .expectNext("0") .then { - blocks.onNext(new BlockContainer(1L, BlockId.from(hash1), BigInteger.ONE, Instant.now(), false, null, null, [])) + blocks.tryEmitNext(new BlockContainer(1L, BlockId.from(hash1), BigInteger.ONE, Instant.now(), false, null, null, [])) } .expectNext("1230000") .then { - blocks.onComplete() + blocks.tryEmitComplete() } .expectComplete() .verify(Duration.ofSeconds(1)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy index 7b8f089d..d5d1d659 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy @@ -22,11 +22,11 @@ import org.jetbrains.annotations.NotNull import org.reactivestreams.Publisher import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.publisher.TopicProcessor +import reactor.core.publisher.Sinks class EthereumHeadMock implements Head { - private TopicProcessor bus = TopicProcessor.create() + private Sinks.Many bus = Sinks.many().multicast().onBackpressureBuffer() private Publisher predefined = null private BlockContainer latest private List handlers = [] @@ -38,7 +38,7 @@ class EthereumHeadMock implements Head { assert block != null println("New block: ${block.height} / ${block.hash}") latest = block - bus.onNext(block) + bus.tryEmitNext(block) } void setPredefined(Publisher predefined) { @@ -54,7 +54,7 @@ class EthereumHeadMock implements Head { if (predefined != null) { return Flux.concat(Mono.justOrEmpty(latest), Flux.from(predefined)) } else { - return Flux.concat(Mono.justOrEmpty(latest), bus).distinctUntilChanged() + return Flux.concat(Mono.justOrEmpty(latest), bus.asFlux()).distinctUntilChanged() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy index a0a022f8..b0cf0990 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy @@ -18,7 +18,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import reactor.core.publisher.Flux -import reactor.core.publisher.TopicProcessor +import reactor.core.publisher.Sinks import reactor.test.StepVerifier import spock.lang.Specification @@ -35,28 +35,28 @@ class AbstractHeadSpec extends Specification { def "Calls beforeBlock on each block"() { setup: - TopicProcessor source = TopicProcessor.create() + Sinks.Many source = Sinks.many().unicast().onBackpressureBuffer() def head = new TestHead() def called = false when: - head.follow(Flux.from(source)) + head.follow(source.asFlux()) head.onBeforeBlock { called = true } def act = head.flux - source.onNext(blocks[0]) + source.tryEmitNext(blocks[0]) then: StepVerifier.create(act) .expectNext(blocks[0]) .then { assert called called = false - source.onNext(blocks[1]) + source.tryEmitNext(blocks[1]) } .expectNext(blocks[1]) .then { assert called - source.onComplete() + source.tryEmitComplete() } .expectComplete() .verify(Duration.ofSeconds(1)) @@ -64,29 +64,29 @@ class AbstractHeadSpec extends Specification { def "Follows source"() { setup: - TopicProcessor source = TopicProcessor.create() + Sinks.Many source = Sinks.many().unicast().onBackpressureBuffer() def head = new TestHead() when: - head.follow(Flux.from(source)) + head.follow(source.asFlux()) def act = head.flux - source.onNext(blocks[0]) + source.tryEmitNext(blocks[0]) then: StepVerifier.create(act) .expectNext(blocks[0]) - .then { source.onNext(blocks[1]) } + .then { source.tryEmitNext(blocks[1]) } .expectNext(blocks[1]) - .then { source.onNext(blocks[2]) } + .then { source.tryEmitNext(blocks[2]) } .expectNext(blocks[2]) - .then { source.onNext(blocks[3]) } + .then { source.tryEmitNext(blocks[3]) } .expectNext(blocks[3]) - .then { source.onComplete() } + .then { source.tryEmitComplete() } .expectComplete() .verify(Duration.ofSeconds(1)) } def "Ignores block will less difficulty"() { setup: - TopicProcessor source = TopicProcessor.create() + Sinks.Many source = Sinks.many().unicast().onBackpressureBuffer() def head = new TestHead() def wrongblock = new BlockContainer( blocks[1].height, BlockId.from(blocks[1].hash.value.clone().tap { it[1] = 0xff as byte }), @@ -95,18 +95,18 @@ class AbstractHeadSpec extends Specification { false, null, null, [] ) when: - head.follow(Flux.from(source)) + head.follow(source.asFlux()) def act = head.flux - source.onNext(blocks[0]) + source.tryEmitNext(blocks[0]) then: StepVerifier.create(act) .expectNext(blocks[0]) - .then { source.onNext(blocks[1]) } + .then { source.tryEmitNext(blocks[1]) } .expectNext(blocks[1]) - .then { source.onNext(wrongblock) } - .then { source.onNext(blocks[3]) } + .then { source.tryEmitNext(wrongblock) } + .then { source.tryEmitNext(blocks[3]) } .expectNext(blocks[3]) - .then { source.onComplete() } + .then { source.tryEmitComplete() } .expectComplete() .verify(Duration.ofSeconds(1)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 0364daba..95b531c9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -63,7 +63,7 @@ class FilteredApisSpec extends Specification { it.setStatus(UpstreamAvailability.OK) } when: - def iter = new FilteredApis(upstreams, matcher, 0, 1, 0) + def iter = new FilteredApis(Chain.ETHEREUM, upstreams, matcher, 0, 1, 0) iter.request(10) then: StepVerifier.create(iter) @@ -74,7 +74,7 @@ class FilteredApisSpec extends Specification { .verify(Duration.ofSeconds(1)) when: - iter = new FilteredApis(upstreams, matcher, 1, 1, 0) + iter = new FilteredApis(Chain.ETHEREUM, upstreams, matcher, 1, 1, 0) iter.request(10) then: StepVerifier.create(iter) @@ -85,7 +85,7 @@ class FilteredApisSpec extends Specification { .verify(Duration.ofSeconds(1)) when: - iter = new FilteredApis(upstreams, matcher, 1, 2, 0) + iter = new FilteredApis(Chain.ETHEREUM, upstreams, matcher, 1, 2, 0) iter.request(10) then: StepVerifier.create(iter) @@ -101,7 +101,7 @@ class FilteredApisSpec extends Specification { def "Exponential backoff"() { setup: - def apis = new FilteredApis([], Selector.empty, 0, 1, 0) + def apis = new FilteredApis(Chain.ETHEREUM, [], Selector.empty, 0, 1, 0) expect: wait == apis.waitDuration(n).toMillis() as Integer where: @@ -123,7 +123,7 @@ class FilteredApisSpec extends Specification { @Retry def "Backoff uses jitter"() { setup: - def apis = new FilteredApis([], Selector.empty, 0, 1, 20) + def apis = new FilteredApis(Chain.ETHEREUM, [], Selector.empty, 0, 1, 20) when: def act = apis.waitDuration(1).toMillis() println act @@ -149,7 +149,7 @@ class FilteredApisSpec extends Specification { def up2 = TestingCommons.upstream(api2) then: StepVerifier.withVirtualTime({ - def apis = new FilteredApis([up1, up2], Selector.empty, 0, 4, 0) + def apis = new FilteredApis(Chain.ETHEREUM, [up1, up2], Selector.empty, 0, 4, 0) apis.request(10) return apis }) @@ -173,7 +173,7 @@ class FilteredApisSpec extends Specification { TestingCommons.upstream(it) } when: - def act = new FilteredApis(ups, Selector.empty, 2, 1, 0) + def act = new FilteredApis(Chain.ETHEREUM, ups, Selector.empty, 2, 1, 0) act.request(10) then: StepVerifier.create(act) @@ -225,7 +225,8 @@ class FilteredApisSpec extends Specification { } ] when: - def act = new FilteredApis([] + fallback + standard, + def act = new FilteredApis(Chain.ETHEREUM, + [] + fallback + standard, Selector.empty, 0, 3, 0) act.request(10) then: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy index 4ccf5868..34953d6f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy @@ -24,7 +24,7 @@ import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.json.BlockJson import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux -import reactor.core.publisher.TopicProcessor +import reactor.core.publisher.Sinks import reactor.test.StepVerifier import reactor.util.function.Tuples import spock.lang.Specification @@ -58,9 +58,9 @@ class HeadLagObserverSpec extends Specification { }) } - def masterBus = TopicProcessor.create() + def masterBus = Sinks.many().multicast().onBackpressureBuffer() - 1 * master.getFlux() >> Flux.from(masterBus) + 1 * master.getFlux() >> masterBus.asFlux() 1 * head1.getFlux() >> Flux.merge( Flux.just(blocks[1]), Flux.just(blocks[2]).delaySubscription(Duration.ofSeconds(1)) @@ -79,7 +79,7 @@ class HeadLagObserverSpec extends Specification { then: StepVerifier.create(act) - .then { masterBus.onNext(blocks[1]) } + .then { masterBus.tryEmitNext(blocks[1]) } .expectNextCount(3) .verifyComplete() }