Fix unsubscribe pending requests (#722)

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 {

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<String> {
         return listOf(EthereumEgressSubscription.METHOD_PENDING_TXES)

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<TransactionId> {
 
@@ -35,7 +38,8 @@ class WebsocketPendingTxes(
     }
 
     override fun createConnection(): Flux<TransactionId> {
-        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() }
     }

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)