From 0ed77f1cfd35c35b029d9cab95f4b55fc53b45bf Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Wed, 10 Sep 2025 15:18:54 +0400 Subject: [PATCH] Fix unsubscribe requests (#721) --- .../kotlin/io/emeraldpay/dshackle/Global.kt | 15 ++++++++ .../beaconchain/BeaconChainSpecific.kt | 2 +- .../upstream/cosmos/CosmosChainSpecific.kt | 2 +- .../ethereum/EthereumChainSpecific.kt | 4 +-- .../upstream/ethereum/GenericWsHead.kt | 34 +++++++++++++++---- .../upstream/ethereum/WsSubscriptions.kt | 3 +- .../upstream/ethereum/WsSubscriptionsImpl.kt | 9 ++--- .../subscribe/WebsocketPendingTxes.kt | 1 + .../upstream/generic/AbstractChainSpecific.kt | 2 +- .../upstream/generic/ChainSpecific.kt | 4 +-- .../generic/GenericIngressSubscription.kt | 27 +++++++++++---- .../connectors/GenericConnectorFactory.kt | 1 + .../generic/connectors/GenericRpcConnector.kt | 2 +- .../generic/connectors/GenericWsConnector.kt | 4 ++- .../upstream/kadena/KadenaChainSpecific.kt | 2 +- .../upstream/near/NearChainSpecific.kt | 2 +- .../polkadot/PolkadotChainSpecific.kt | 6 ++-- .../upstream/ripple/RippleChainSpecific.kt | 2 +- .../upstream/solana/SolanaChainSpecific.kt | 6 ++-- .../starknet/StarknetChainSpecific.kt | 2 +- .../dshackle/upstream/ton/TonHttpSpecific.kt | 2 +- .../ethereum/GenericWsHeadSpec.groovy | 23 +++++++------ .../ethereum/WsSubscriptionsImplSpec.groovy | 4 +-- .../subscribe/WebsocketPendingTxesSpec.groovy | 6 ++-- .../upstream/ethereum/GenericWsHeadTest.kt | 15 ++++---- .../generic/GenericSubscriptionConnectTest.kt | 7 ++-- 26 files changed, 125 insertions(+), 62 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index ec7da933..8d5ee45f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt @@ -35,6 +35,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.domain.TransactionId import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.TransactionIdSerializer import io.emeraldpay.dshackle.upstream.ton.TonMasterchainInfo import io.emeraldpay.dshackle.upstream.ton.TonMasterchainInfoDeserializer +import java.math.BigInteger import java.text.SimpleDateFormat import java.util.Locale import java.util.TimeZone @@ -64,6 +65,20 @@ class Global { } ?: Chain.UNSPECIFIED } + fun getSubId(subId: String, chain: Chain): Any { + return if (isSolana(chain)) { + runCatching { + BigInteger(subId) as Any + }.getOrElse { subId } + } else { + subId + } + } + + private fun isSolana(chain: Chain): Boolean { + return chain == Chain.SOLANA__MAINNET || chain == Chain.SOLANA__DEVNET || chain == Chain.SOLANA__TESTNET + } + @JvmStatic val objectMapper: ObjectMapper = createObjectMapper() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt index 25d3c8a8..b7f33524 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt @@ -37,7 +37,7 @@ object BeaconChainSpecific : AbstractPollChainSpecific() { throw NotImplementedError() } - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest { throw NotImplementedError() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/cosmos/CosmosChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/cosmos/CosmosChainSpecific.kt index 6054f337..ae6a8c8d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/cosmos/CosmosChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/cosmos/CosmosChainSpecific.kt @@ -74,7 +74,7 @@ object CosmosChainSpecific : AbstractPollChainSpecific() { override fun listenNewHeadsRequest() = throw NotImplementedError() // ChainRequest("subscribe", ListParams("tm.event = 'NewBlockHeader'")) - override fun unsubscribeNewHeadsRequest(subId: String) = throw NotImplementedError() + override fun unsubscribeNewHeadsRequest(subId: Any) = throw NotImplementedError() // ChainRequest("unsubscribe", ListParams("tm.event = 'NewBlockHeader'")) override fun upstreamValidators( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt index c19c56be..61909166 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt @@ -57,7 +57,7 @@ object EthereumChainSpecific : AbstractPollChainSpecific() { ChainRequest("eth_getBlockByNumber", ListParams("latest", false)) override fun listenNewHeadsRequest(): ChainRequest = ChainRequest("eth_subscribe", ListParams("newHeads")) - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest = + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest = ChainRequest("eth_unsubscribe", ListParams(subId)) override fun localReaderBuilder( @@ -208,7 +208,7 @@ object EthereumChainSpecific : AbstractPollChainSpecific() { return EthereumUpstreamSettingsDetector(upstream, chain) } - override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { + override fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription { return EthereumWsIngressSubscription(ws) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt index c4e83c48..8523a12e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt @@ -16,6 +16,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.upstream.BlockValidator @@ -48,7 +49,7 @@ class GenericWsHead( private val wsSubscriptions: WsSubscriptions, private val wsConnectionResubscribeScheduler: Scheduler, headScheduler: Scheduler, - upstream: DefaultUpstream, + private val upstream: DefaultUpstream, private val chainSpecific: ChainSpecific, jsonRpcWsClient: JsonRpcWsClient, timeout: Duration, @@ -122,19 +123,30 @@ class GenericWsHead( .flatMap { data -> chainSpecific.getFromHeader(data, "unknown", api) } - .timeout(wsHeadTimeout, Mono.error(RuntimeException("No response from subscribe to newHeads"))) + .timeout( + wsHeadTimeout, + Mono.error(RuntimeException("No response from subscribe to newHeads")), + ) .onErrorResume { err -> log.error("Error getting heads for {}, message {}", upstreamId, err.message) unsubscribe() } } + UPSTREAM_SETTINGS_ERROR -> { - log.warn("Couldn't check chain settings via ws connection for {}, ws sub will be recreated", upstreamId) + log.warn( + "Couldn't check chain settings via ws connection for {}, ws sub will be recreated", + upstreamId, + ) subscribed.set(false) Mono.empty() } + UPSTREAM_FATAL_SETTINGS_ERROR -> { - log.error("Chain settings check hasn't been passed via ws connection, upstream {} will be removed", upstreamId) + log.error( + "Chain settings check hasn't been passed via ws connection, upstream {} will be removed", + upstreamId, + ) headLivenessSink.emitNext(HeadLivenessState.FATAL_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } Mono.empty() } @@ -154,7 +166,13 @@ class GenericWsHead( private fun unsubscribe(): Mono { subscribed.set(false) - return wsSubscriptions.unsubscribe(chainSpecific.unsubscribeNewHeadsRequest(subscriptionId.get()).copy(id = ids.getAndIncrement())) + return wsSubscriptions.unsubscribe( + chainSpecific.unsubscribeNewHeadsRequest( + Global.getSubId(subscriptionId.get(), upstream.getChain()), + ).copy( + id = ids.getAndIncrement(), + ), + ) .flatMap { it.requireResult() } .doOnNext { log.warn("{} has just unsubscribed from newHeads", upstreamId) } .onErrorResume { @@ -171,11 +189,13 @@ class GenericWsHead( wsSubscriptions.subscribe(chainSpecific.listenNewHeadsRequest().copy(id = ids.getAndIncrement())) .also { connectionId.set(it.connectionId) - subscriptionId.set(it.subId.get()) if (!connected.get()) { connected.set(true) } - }.data + }.data.flatMapMany { + subscriptionId.set(it.t1) + it.t2 + } } catch (e: Exception) { Flux.error(e) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptions.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptions.kt index 74e72891..f77f572e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptions.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptions.kt @@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.util.function.Tuple2 import java.util.concurrent.atomic.AtomicReference /** @@ -48,7 +49,7 @@ interface WsSubscriptions { fun unsubscribe(request: ChainRequest): Mono data class SubscribeData( - val data: Flux, + val data: Mono>>, val connectionId: String, val subId: AtomicReference, ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt index fc562340..6b7be8c2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.util.function.Tuples import java.util.concurrent.atomic.AtomicReference class WsSubscriptionsImpl( @@ -40,8 +41,8 @@ class WsSubscriptionsImpl( .filter { it.result != null } // should never happen .map { it.result!! } - val messageFlux = conn.callRpc(request) - .flatMapMany { + val message = conn.callRpc(request) + .flatMap { if (it.hasError()) { log.warn("Failed to establish subscription: ${it.error?.message}") Mono.error(ChainException(it.id, it.error!!)) @@ -52,11 +53,11 @@ class WsSubscriptionsImpl( it.getResultAsProcessedString() } subscriptionId.set(id) - messages + Mono.just(Tuples.of(subscriptionId.get(), messages)) } } - return WsSubscriptions.SubscribeData(messageFlux, conn.connectionId(), subscriptionId) + return WsSubscriptions.SubscribeData(message, conn.connectionId(), subscriptionId) } override fun unsubscribe(request: ChainRequest): Mono { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt index 4ff3168f..219d75cd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt @@ -37,6 +37,7 @@ class WebsocketPendingTxes( override fun createConnection(): Flux { return wsSubscriptions.subscribe(ChainRequest("eth_subscribe", ListParams(EthereumEgressSubscription.METHOD_PENDING_TXES))) .data + .flatMapMany { it.t2 } .timeout(Duration.ofSeconds(85), Mono.empty()) .map { // comes as a JS string, i.e., within quotes diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt index 559c42c6..12e3d753 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt @@ -72,7 +72,7 @@ abstract class AbstractChainSpecific : ChainSpecific { config: UpstreamsConfig.Upstream<*>?, ): UpstreamRpcMethodsDetector? = null - override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { + override fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription { return NoIngressSubscription() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt index cdd4e19f..fed279b8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -65,7 +65,7 @@ interface ChainSpecific { fun listenNewHeadsRequest(): ChainRequest - fun unsubscribeNewHeadsRequest(subId: String): ChainRequest + fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest fun finalizationDetectorBuilder(): FinalizationDetector @@ -96,7 +96,7 @@ interface ChainSpecific { config: UpstreamsConfig.Upstream<*>?, ): UpstreamRpcMethodsDetector? - fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription + fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription fun callSelector(caches: Caches): CallSelector? diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt index f9858cb3..a4848584 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt @@ -1,5 +1,7 @@ package io.emeraldpay.dshackle.upstream.generic +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.IngressSubscription import io.emeraldpay.dshackle.upstream.SubscriptionConnect @@ -11,9 +13,12 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.time.Duration import java.util.concurrent.ConcurrentHashMap -import kotlin.math.log -class GenericIngressSubscription(val conn: WsSubscriptions, val methods: List) : IngressSubscription { +class GenericIngressSubscription( + val chain: Chain, + val conn: WsSubscriptions, + val methods: List, +) : IngressSubscription { override fun getAvailableTopics(): List { return methods } @@ -24,6 +29,7 @@ class GenericIngressSubscription(val conn: WsSubscriptions, val methods: List get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect { return holders.computeIfAbsent(topic to params) { key -> GenericSubscriptionConnect( + chain, conn, key.first, key.second, @@ -34,6 +40,7 @@ class GenericIngressSubscription(val conn: WsSubscriptions, val methods: List { val sub = conn.subscribe(ChainRequest(topic, ListParams(getParams(params) as List))) return sub.data + .flatMapMany { it.t2 } .timeout( Duration.ofSeconds(85), Mono.empty().doOnEach { @@ -57,11 +65,18 @@ class GenericSubscriptionConnect( .onErrorResume { log.error("Error during subscription to $topic", it) Mono.empty() - }.doFinally { + } + .doFinally { if (unsubscribeMethod != "") { - conn.unsubscribe(ChainRequest(unsubscribeMethod, ListParams(sub.subId.get()))).subscribe { - log.info("unsubscribed from ${sub.subId.get()}") - } + conn.unsubscribe( + ChainRequest( + unsubscribeMethod, + ListParams(Global.getSubId(sub.subId.get(), chain)), + ), + ) + .subscribe { + log.info("unsubscribed from ${sub.subId.get()}") + } } } as Flux } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactory.kt index 907649f7..0db2e2a5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactory.kt @@ -53,6 +53,7 @@ open class GenericConnectorFactory( val specific = ChainSpecificRegistry.resolve(chain) if (wsFactory != null && connectorType == WS_ONLY) { return GenericWsConnector( + chain, wsFactory, upstream, forkChoice, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt index d4c639dc..7e0a60b1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt @@ -71,7 +71,7 @@ class GenericRpcConnector( pool = wsFactory?.create(upstream) wsSubs = pool?.let { WsSubscriptionsImpl(it) } jsonRpcWsClient = pool?.let { JsonRpcWsClient(pool) } - ingressSubscription = wsSubs?.let { chainSpecific.makeIngressSubscription(it) } + ingressSubscription = wsSubs?.let { chainSpecific.makeIngressSubscription(chain, it) } head = when (connectorType) { RPC_ONLY -> { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt index 2a847f7b..98794195 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.upstream.generic.connectors +import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.DefaultUpstream @@ -19,6 +20,7 @@ import reactor.core.scheduler.Scheduler import java.time.Duration class GenericWsConnector( + chain: Chain, wsFactory: WsConnectionPoolFactory, upstream: DefaultUpstream, forkChoice: ForkChoice, @@ -51,7 +53,7 @@ class GenericWsConnector( expectedBlockTime, ) liveness = HeadLivenessValidatorImpl(head, expectedBlockTime, headLivenessScheduler, upstream.getId()) - subscriptions = chainSpecific.makeIngressSubscription(wsSubscriptions) + subscriptions = chainSpecific.makeIngressSubscription(chain, wsSubscriptions) } override fun headLivenessEvents(): Flux { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/kadena/KadenaChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/kadena/KadenaChainSpecific.kt index b4895b69..d9ee996f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/kadena/KadenaChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/kadena/KadenaChainSpecific.kt @@ -56,7 +56,7 @@ object KadenaChainSpecific : AbstractPollChainSpecific() { throw NotImplementedError() } - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest { throw NotImplementedError() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt index 9bb1a7fd..2caeebe3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt @@ -53,7 +53,7 @@ object NearChainSpecific : AbstractPollChainSpecific() { throw NotImplementedError() } - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest { throw NotImplementedError() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt index ae8b3b79..17a9c870 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt @@ -72,7 +72,7 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() { override fun listenNewHeadsRequest(): ChainRequest = ChainRequest("chain_subscribeNewHeads", ListParams()) - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest = + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest = ChainRequest("chain_unsubscribeNewHeads", ListParams(subId)) override fun localReaderBuilder( @@ -146,8 +146,8 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() { } } - override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { - return GenericIngressSubscription(ws, DefaultPolkadotMethods.subs.map { it.first }) + override fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription { + return GenericIngressSubscription(chain, ws, DefaultPolkadotMethods.subs.map { it.first }) } override fun upstreamRpcMethodsDetector( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt index a1b0a48d..b0bcc125 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt @@ -61,7 +61,7 @@ object RippleChainSpecific : AbstractPollChainSpecific() { throw NotImplementedError() } - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest { throw NotImplementedError() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt index 388d76c1..77b0a9aa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt @@ -111,7 +111,7 @@ object SolanaChainSpecific : AbstractChainSpecific() { ) } - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest { return ChainRequest("blockUnsubscribe", ListParams(subId)) } @@ -154,8 +154,8 @@ object SolanaChainSpecific : AbstractChainSpecific() { return SolanaUpstreamSettingsDetector(upstream) } - override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { - return GenericIngressSubscription(ws, DefaultSolanaMethods.subs.map { it.first }) + override fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription { + return GenericIngressSubscription(chain, ws, DefaultSolanaMethods.subs.map { it.first }) } override fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt index 88fbefcc..5e508e9b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt @@ -55,7 +55,7 @@ object StarknetChainSpecific : AbstractPollChainSpecific() { throw NotImplementedError() } - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest { throw NotImplementedError() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ton/TonHttpSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ton/TonHttpSpecific.kt index 6ed79ed3..6e7b99d4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ton/TonHttpSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ton/TonHttpSpecific.kt @@ -40,7 +40,7 @@ object TonHttpSpecific : AbstractPollChainSpecific() { throw NotImplementedError() } - override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest { throw NotImplementedError() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy index e28c2d6e..622b6270 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy @@ -37,6 +37,7 @@ import reactor.core.publisher.Mono import reactor.core.publisher.Sinks import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier +import reactor.util.function.Tuples import spock.lang.Specification import java.time.Duration @@ -77,7 +78,7 @@ class GenericWsHeadSpec extends Specification { def client = new JsonRpcWsClient(pool) 1 * ws.subscribe(_) >> new WsSubscriptions.SubscribeData( - Flux.fromIterable([headBlock]), "id", new AtomicReference("") + Mono.just(Tuples.of("", Flux.fromIterable([headBlock]))), "id", new AtomicReference("") ) def head = new GenericWsHead( @@ -130,8 +131,8 @@ class GenericWsHeadSpec extends Specification { def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux() 2 * subscribe(_) >>> [ - new WsSubscriptions.SubscribeData(Flux.error(new RuntimeException()), "id", new AtomicReference("")), - new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference("")) + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.error(new RuntimeException()))), "id", new AtomicReference("")), + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference("")) ] 1 * it.unsubscribe(new ChainRequest("eth_unsubscribe", new ListParams(""), 2, null, null, false, Selector.UpstreamFilter.default)) >> Mono.just(new ChainResponse("".bytes, null)) @@ -202,8 +203,8 @@ class GenericWsHeadSpec extends Specification { def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux() 2 * subscribe(_) >>> [ - new WsSubscriptions.SubscribeData(Flux.fromIterable([firstHeadBlock]), "id", new AtomicReference("")), - new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference("")) + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference("")), + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference("")) ] } @@ -259,7 +260,7 @@ class GenericWsHeadSpec extends Specification { def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux() 1 * subscribe(_) >>> [ - new WsSubscriptions.SubscribeData(Flux.fromIterable([firstHeadBlock]), "id", new AtomicReference("")), + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference("")), ] } @@ -314,7 +315,7 @@ class GenericWsHeadSpec extends Specification { def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux() 1 * subscribe(_) >>> [ - new WsSubscriptions.SubscribeData(Flux.fromIterable([firstHeadBlock]), "id", new AtomicReference("")), + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference("")), ] } @@ -382,8 +383,8 @@ class GenericWsHeadSpec extends Specification { def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux() 2 * subscribe(_) >>> [ - new WsSubscriptions.SubscribeData(Flux.fromIterable([firstHeadBlock]), "id", new AtomicReference("")), - new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference("")), + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference("")), + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference("")), ] } @@ -451,7 +452,7 @@ class GenericWsHeadSpec extends Specification { def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> Flux.empty() 1 * it.subscribe(_) >> new WsSubscriptions.SubscribeData( - Flux.error(new RuntimeException()), "id", new AtomicReference(subId) + Mono.just(Tuples.of(subId, Flux.error(new RuntimeException()))), "id", new AtomicReference(subId) ) 1 * it.unsubscribe(new ChainRequest("eth_unsubscribe", new ListParams(subId), 2, null, null, false, Selector.UpstreamFilter.default)) >> Mono.just(new ChainResponse("".bytes, null)) @@ -506,7 +507,7 @@ class GenericWsHeadSpec extends Specification { def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux() 1 * subscribe(_) >>> [ - new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference("")) + new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference("")) ] } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy index f7de40f2..a6d3285a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy @@ -47,7 +47,7 @@ class WsSubscriptionsImplSpec extends Specification { when: def act = ws.subscribe(new ChainRequest("eth_subscribe", new ListParams(["foo_bar"]))) - .data + .data.flatMapMany { it.getT2() } .map { new String(it) } .take(3) .collectList().block(Duration.ofSeconds(1)) @@ -85,7 +85,7 @@ class WsSubscriptionsImplSpec extends Specification { when: def act = ws.subscribe(new ChainRequest("eth_subscribe", new ListParams(["foo_bar"]))) - .data + .data.flatMapMany { it.getT2() } .map { new String(it) } .take(3) .collectList().block(Duration.ofSeconds(1)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxesSpec.groovy index 7f079c26..ff3434a5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxesSpec.groovy @@ -16,10 +16,12 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.util.function.Tuples import spock.lang.Specification import java.time.Duration @@ -44,7 +46,7 @@ class WebsocketPendingTxesSpec extends Specification { then: 1 * ws.subscribe(new ChainRequest("eth_subscribe", new ListParams(["newPendingTransactions"]))) >> new WsSubscriptions.SubscribeData( - Flux.fromIterable(responses), "id", new AtomicReference("") + Mono.just(Tuples.of("", Flux.fromIterable(responses))), "id", new AtomicReference("") ) txes.collect {it.toHex() } == [ "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c", diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadTest.kt index b81724c4..03e1b659 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadTest.kt @@ -26,6 +26,7 @@ import reactor.core.publisher.Mono import reactor.core.publisher.Sinks import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier +import reactor.util.function.Tuples import java.math.BigInteger import java.time.Duration import java.time.Instant @@ -43,7 +44,7 @@ class GenericWsHeadTest { val wsSub = mock { on { connectionInfoFlux() } doReturn Flux.empty() on { subscribe(ChainRequest("eth_subscribe", ListParams("newHeads"))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(Global.objectMapper.writeValueAsBytes(block)), "id", AtomicReference("subId")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("subId", Flux.just(Global.objectMapper.writeValueAsBytes(block)))), "id", AtomicReference("subId")) } val connection = mock { on { isConnected } doReturn true @@ -141,7 +142,7 @@ class GenericWsHeadTest { val wsSub = mock { on { connectionInfoFlux() } doReturn Flux.empty() on { subscribe(ChainRequest("eth_subscribe", ListParams("newHeads"))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)), "id", AtomicReference("subId")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("subId", Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)))), "id", AtomicReference("subId")) } val connection = mock() val wsPool = mock { @@ -189,7 +190,7 @@ class GenericWsHeadTest { val wsSub = mock { on { connectionInfoFlux() } doReturn connectionInfoSink.asFlux() on { subscribe(ChainRequest("eth_subscribe", ListParams("newHeads"))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(Global.objectMapper.writeValueAsBytes(block)), "id", AtomicReference("subId")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("subId", Flux.just(Global.objectMapper.writeValueAsBytes(block)))), "id", AtomicReference("subId")) } val connection = mock { on { isConnected } doReturn true @@ -247,7 +248,7 @@ class GenericWsHeadTest { val wsSub = mock { on { connectionInfoFlux() } doReturn Flux.empty() on { subscribe(ChainRequest("eth_subscribe", ListParams("newHeads"))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)), "id", AtomicReference("subId")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("sudId", Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)))), "id", AtomicReference("subId")) } val connection = mock { on { isConnected } doReturn true @@ -299,7 +300,7 @@ class GenericWsHeadTest { val wsSub = mock { on { connectionInfoFlux() } doReturn Flux.empty() on { subscribe(ChainRequest("eth_subscribe", ListParams("newHeads"))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)), "id", AtomicReference("subId")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("subId", Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)))), "id", AtomicReference("subId")) } val connection = mock { on { isConnected } doReturn true @@ -351,7 +352,7 @@ class GenericWsHeadTest { val wsSub = mock { on { connectionInfoFlux() } doReturn Flux.empty() on { subscribe(ChainRequest("eth_subscribe", ListParams("newHeads"))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)), "id", AtomicReference("subId")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("subId", Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)))), "id", AtomicReference("subId")) } val connection = mock { on { isConnected } doReturn true @@ -405,7 +406,7 @@ class GenericWsHeadTest { val wsSub = mock { on { connectionInfoFlux() } doReturn Flux.empty() on { subscribe(ChainRequest("eth_subscribe", ListParams("newHeads"))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)), "id", AtomicReference("subId")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("sudId", Flux.just(Global.objectMapper.writeValueAsBytes(block1), Global.objectMapper.writeValueAsBytes(block2)))), "id", AtomicReference("subId")) } val connection = mock { on { isConnected } doReturn true diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt index 4d030a60..f8b04816 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.upstream.generic +import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions import io.emeraldpay.dshackle.upstream.rpcclient.ListParams @@ -8,7 +9,9 @@ import org.mockito.Mockito.verify import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import reactor.test.StepVerifier +import reactor.util.function.Tuples import java.time.Duration import java.util.concurrent.atomic.AtomicReference @@ -21,10 +24,10 @@ class GenericSubscriptionConnectTest { val response = "hello".toByteArray() val ws = mock { on { subscribe(ChainRequest(topic, ListParams(param))) } doReturn - WsSubscriptions.SubscribeData(Flux.just(response), "", AtomicReference("")) + WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.just(response))), "", AtomicReference("")) } - val genericSubscriptionConnect = GenericSubscriptionConnect(ws, topic, param, "") + val genericSubscriptionConnect = GenericSubscriptionConnect(Chain.ETHEREUM__MAINNET, ws, topic, param, "") StepVerifier.create(genericSubscriptionConnect.createConnection()) .expectNext(response)