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 61909166..bc57c9dc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt @@ -209,7 +209,7 @@ object EthereumChainSpecific : AbstractPollChainSpecific() { } override fun makeIngressSubscription(chain: Chain, ws: WsSubscriptions): IngressSubscription { - return EthereumWsIngressSubscription(ws) + return EthereumWsIngressSubscription(chain, ws) } override fun callSelector(caches: Caches): CallSelector { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt index c25358ca..0114dd05 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum.subscribe +import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.upstream.IngressSubscription import io.emeraldpay.dshackle.upstream.SubscriptionConnect import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription @@ -22,10 +23,11 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions class EthereumWsIngressSubscription( + chain: Chain, conn: WsSubscriptions, ) : IngressSubscription, EthereumIngressSubscription { - private val pendingTxes = WebsocketPendingTxes(conn) + private val pendingTxes = WebsocketPendingTxes(chain, conn) override fun getAvailableTopics(): List { return listOf(EthereumEgressSubscription.METHOD_PENDING_TXES) 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 219d75cd..ddd71c83 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 @@ -15,6 +15,8 @@ */ package io.emeraldpay.dshackle.upstream.ethereum.subscribe +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.SubscriptionConnect import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription @@ -27,6 +29,7 @@ import reactor.core.publisher.Mono import java.time.Duration class WebsocketPendingTxes( + private val chain: Chain, private val wsSubscriptions: WsSubscriptions, ) : DefaultPendingTxesSource(), SubscriptionConnect { @@ -35,7 +38,8 @@ class WebsocketPendingTxes( } override fun createConnection(): Flux { - return wsSubscriptions.subscribe(ChainRequest("eth_subscribe", ListParams(EthereumEgressSubscription.METHOD_PENDING_TXES))) + val sub = wsSubscriptions.subscribe(ChainRequest("eth_subscribe", ListParams(EthereumEgressSubscription.METHOD_PENDING_TXES))) + return sub .data .flatMapMany { it.t2 } .timeout(Duration.ofSeconds(85), Mono.empty()) @@ -45,6 +49,17 @@ class WebsocketPendingTxes( System.arraycopy(it, 1, value, 0, value.size) TransactionId.from(String(value)) } + .doFinally { + wsSubscriptions.unsubscribe( + ChainRequest( + "eth_unsubscribe", + ListParams(Global.getSubId(sub.subId.get(), chain)), + ), + ) + .subscribe { + log.info("unsubscribed from ${sub.subId.get()}") + } + } .doOnError { t -> log.warn("Invalid pending transaction", t) } .onErrorResume { Mono.empty() } } 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 ff3434a5..68d453e0 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 @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum.subscribe +import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions @@ -38,7 +39,7 @@ class WebsocketPendingTxesSpec extends Specification { '"0xa38173981f8eab96ee70cefe42735af0f574b7ef354565f2fea32a28e5ed9bd2"', ].collect { it.bytes } def ws = Mock(WsSubscriptions) - def pending = new WebsocketPendingTxes(ws) + def pending = new WebsocketPendingTxes(Chain.ETHEREUM__MAINNET, ws) when: def txes = pending.connect(Selector.empty).take(3)