problem: Spring Reactor is too outdated

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.netty:reactor-netty:1.0.6"
-    implementation 'io.projectreactor.addons:reactor-extra:3.3.3.RELEASE'
+    implementation 'io.projectreactor.addons:reactor-extra:3.4.3'
-    implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.2.RELEASE'
+    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"

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
+springBootVersion=2.4.5
-springVersion=5.2.6.RELEASE
+springVersion=5.3.6
-springSecurtyVersion=5.3.2.RELEASE
+springSecurtyVersion=5.4.6
-reactorVersion=3.3.5.RELEASE
+reactorVersion=3.4.5
 nettyVersion=4.1.53.Final
 # Our Libs
 etherjarVersion=0.10.2

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 {
+                .doOnChannelInit { _, channel, _ ->
-                    it.bootstrap { b ->
+                    channel.pipeline().addFirst(errorHandler)
-                        b.handler(errorHandler)
-                    }
                 }
                 .host(config.host)
                 .port(config.port)

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))

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))

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<BlockContainer> bus = TopicProcessor.create()
+    private Sinks.Many<BlockContainer> bus = Sinks.many().multicast().onBackpressureBuffer()
     private Publisher<BlockContainer> predefined = null
     private BlockContainer latest
     private List<Runnable> 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<BlockContainer> 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()
         }
     }
 

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<BlockContainer> source = TopicProcessor.create()
+        Sinks.Many<BlockContainer> 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<BlockContainer> source = TopicProcessor.create()
+        Sinks.Many<BlockContainer> 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<BlockContainer> source = TopicProcessor.create()
+        Sinks.Many<BlockContainer> 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.tryEmitNext(wrongblock) }
-                .then { source.onNext(blocks[3]) }
+                .then { source.tryEmitNext(blocks[3]) }
                 .expectNext(blocks[3])
-                .then { source.onComplete() }
+                .then { source.tryEmitComplete() }
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
     }

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:

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()
     }