Fix unsubscribe requests (#721)

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

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

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(

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

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<BlockContainer> {
         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)
         }

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<ChainResponse>
 
     data class SubscribeData(
-        val data: Flux<ByteArray>,
+        val data: Mono<Tuple2<String, Flux<ByteArray>>>,
         val connectionId: String,
         val subId: AtomicReference<String>,
     )

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)
+        val message = conn.callRpc(request)
-            .flatMapMany {
+            .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<ChainResponse> {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt

@@ -37,6 +37,7 @@ class WebsocketPendingTxes(
     override fun createConnection(): Flux<TransactionId> {
         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

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

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?
 

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<String>) : IngressSubscription {
+class GenericIngressSubscription(
+    val chain: Chain,
+    val conn: WsSubscriptions,
+    val methods: List<String>,
+) : IngressSubscription {
     override fun getAvailableTopics(): List<String> {
         return methods
     }
@@ -24,6 +29,7 @@ class GenericIngressSubscription(val conn: WsSubscriptions, val methods: List<St
     override fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T> {
         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<St
 }
 
 class GenericSubscriptionConnect(
+    val chain: Chain,
     val conn: WsSubscriptions,
     val topic: String,
     val params: Any?,
@@ -48,6 +55,7 @@ class GenericSubscriptionConnect(
     override fun createConnection(): Flux<Any> {
         val sub = conn.subscribe(ChainRequest(topic, ListParams(getParams(params) as List<Any>)))
         return sub.data
+            .flatMapMany { it.t2 }
             .timeout(
                 Duration.ofSeconds(85),
                 Mono.empty<ByteArray?>().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 {
+                    conn.unsubscribe(
-                        log.info("unsubscribed from ${sub.subId.get()}")
+                        ChainRequest(
-                    }
+                            unsubscribeMethod,
+                            ListParams(Global.getSubId(sub.subId.get(), chain)),
+                        ),
+                    )
+                        .subscribe {
+                            log.info("unsubscribed from ${sub.subId.get()}")
+                        }
                 }
             } as Flux<Any>
     }

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,

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

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<HeadLivenessState> {

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

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

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 {
+    override fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription {
-        return GenericIngressSubscription(ws, DefaultPolkadotMethods.subs.map { it.first })
+        return GenericIngressSubscription(chain, ws, DefaultPolkadotMethods.subs.map { it.first })
     }
 
     override fun upstreamRpcMethodsDetector(

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

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 {
+    override fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription {
-        return GenericIngressSubscription(ws, DefaultSolanaMethods.subs.map { it.first })
+        return GenericIngressSubscription(chain, ws, DefaultSolanaMethods.subs.map { it.first })
     }
 
     override fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription {

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

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

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<String>("")
+                Mono.just(Tuples.of("", Flux.fromIterable([headBlock]))), "id", new AtomicReference<String>("")
         )
 
         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<String>("")),
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.error(new RuntimeException()))), "id", new AtomicReference<String>("")),
-                    new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference<String>(""))
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference<String>(""))
             ]
             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<String>("")),
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference<String>("")),
-                    new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference<String>(""))
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference<String>(""))
             ]
         }
 
@@ -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<String>("")),
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference<String>("")),
             ]
         }
 
@@ -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<String>("")),
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference<String>("")),
             ]
         }
 
@@ -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<String>("")),
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([firstHeadBlock]))), "id", new AtomicReference<String>("")),
-                    new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference<String>("")),
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference<String>("")),
             ]
         }
 
@@ -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<String>(subId)
+                    Mono.just(Tuples.of(subId, Flux.error(new RuntimeException()))), "id", new AtomicReference<String>(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<String>(""))
+                    new WsSubscriptions.SubscribeData(Mono.just(Tuples.of("", Flux.fromIterable([secondHeadBlock]))), "id", new AtomicReference<String>(""))
             ]
         }
 

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

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<String>("")
+                Mono.just(Tuples.of("", Flux.fromIterable(responses))), "id", new AtomicReference<String>("")
         )
         txes.collect {it.toHex() } == [
                 "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",

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<WsSubscriptions> {
             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<WsConnection> {
             on { isConnected } doReturn true
@@ -141,7 +142,7 @@ class GenericWsHeadTest {
         val wsSub = mock<WsSubscriptions> {
             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<WsConnection>()
         val wsPool = mock<WsConnectionPool> {
@@ -189,7 +190,7 @@ class GenericWsHeadTest {
         val wsSub = mock<WsSubscriptions> {
             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<WsConnection> {
             on { isConnected } doReturn true
@@ -247,7 +248,7 @@ class GenericWsHeadTest {
         val wsSub = mock<WsSubscriptions> {
             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<WsConnection> {
             on { isConnected } doReturn true
@@ -299,7 +300,7 @@ class GenericWsHeadTest {
         val wsSub = mock<WsSubscriptions> {
             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<WsConnection> {
             on { isConnected } doReturn true
@@ -351,7 +352,7 @@ class GenericWsHeadTest {
         val wsSub = mock<WsSubscriptions> {
             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<WsConnection> {
             on { isConnected } doReturn true
@@ -405,7 +406,7 @@ class GenericWsHeadTest {
         val wsSub = mock<WsSubscriptions> {
             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<WsConnection> {
             on { isConnected } doReturn true

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