feat: add upstream-level subscription topic filtering

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -42,6 +42,7 @@ data class UpstreamsConfig(
         val labels: Labels = Labels(),
         var methods: Methods? = null,
         var methodGroups: MethodGroups? = null,
+        var subscriptions: Subscriptions? = null,
         var role: UpstreamRole = UpstreamRole.PRIMARY,
         var customHeaders: Map<String, String> = emptyMap(),
         var additionalSettings: AdditionalSettings? = null,
@@ -215,4 +216,8 @@ data class UpstreamsConfig(
         val quorum: String? = null,
         val static: String? = null,
     )
+
+    data class Subscriptions(
+        val disabled: Set<String>,
+    )
 }

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -291,6 +291,7 @@ class UpstreamsConfigReader(
         upstream.options = optionsReader.read(upNode)
         upstream.methods = tryReadMethods(upNode)
         upstream.methodGroups = tryReadMethodGroups(upNode)
+        upstream.subscriptions = tryReadSubscriptions(upNode)
         upstream.additionalSettings = readAdditionalSettings(upNode)
         getValueAsBool(upNode, "enabled")?.let {
             upstream.isEnabled = it
@@ -406,4 +407,11 @@ class UpstreamsConfigReader(
             )
         }
     }
+
+    private fun tryReadSubscriptions(upNode: MappingNode): UpstreamsConfig.Subscriptions? {
+        return getMapping(upNode, "subscriptions")?.let { snode ->
+            val disabled = getListOfString(snode, "disabled")?.toSet() ?: emptySet()
+            UpstreamsConfig.Subscriptions(disabled)
+        }
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt

@@ -69,9 +69,15 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
 
     override fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription {
         return { ms ->
-            val pendingTxes: PendingTxesSource = (ms.getAll())
+            val allUpstreams = ms.getAll()
                 .filter { it is GenericUpstream }
                 .map { it as GenericUpstream }
+
+            val disabledTopics = allUpstreams
+                .flatMap { it.getDisabledSubscriptions() }
+                .toSet()
+
+            val pendingTxes: PendingTxesSource = allUpstreams
                 .filter { it.getIngressSubscription() is EthereumIngressSubscription }
                 .mapNotNull {
                     (it.getIngressSubscription() as EthereumIngressSubscription).getPendingTxes()
@@ -84,7 +90,7 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
                         AggregatedPendingTxes(it)
                     }
                 }
-            EthereumEgressSubscription(ms, headScheduler, pendingTxes)
+            EthereumEgressSubscription(ms, headScheduler, pendingTxes, disabledTopics)
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt

@@ -44,10 +44,11 @@ data class Transaction(
     val s: String?,
 )
 
-open class EthereumEgressSubscription(
+open class EthereumEgressSubscription @JvmOverloads constructor(
     val upstream: Multistream,
     val scheduler: Scheduler,
     val pendingTxesSource: PendingTxesSource?,
+    private val disabledTopics: Set<String> = emptySet(),
 ) : EgressSubscription {
 
     companion object {
@@ -72,15 +73,19 @@ open class EthereumEgressSubscription(
         } else {
             listOf()
         }
-        return if (pendingTxesSource != null) {
+        val withPending = if (pendingTxesSource != null) {
             subs.plus(listOf(METHOD_PENDING_TXES, METHOD_DRPC_PENDING_TXES))
         } else {
             subs
         }
+        return withPending.filter { it !in disabledTopics }
     }
 
     @Suppress("UNCHECKED_CAST")
     override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher, unsubscribeMethod: String): Flux<out Any> {
+        if (topic in disabledTopics) {
+            return Flux.error(UnsupportedOperationException("subscription $topic is disabled"))
+        }
         if (topic == METHOD_NEW_HEADS) {
             return newHeads.connect(matcher)
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt

@@ -65,6 +65,7 @@ open class GenericUpstream(
     finalizationDetectorBuilder: FinalizationDetectorBuilder,
     versionRules: Supplier<CompatibleVersionsRules?>,
     private val additionalSettings: UpstreamsConfig.AdditionalSettings?,
+    private val disabledSubscriptions: Set<String> = emptySet(),
 ) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig, chain),
     Lifecycle {
     constructor(
@@ -98,6 +99,7 @@ open class GenericUpstream(
         finalizationDetectorBuilder,
         versionRules,
         config.additionalSettings,
+        config.subscriptions?.disabled ?: emptySet(),
     ) {
         rpcMethodsDetector = upstreamRpcMethodsDetectorBuilder(this, config)
         detectRpcMethods(config, buildMethods)
@@ -449,6 +451,8 @@ open class GenericUpstream(
         return additionalSettings
     }
 
+    fun getDisabledSubscriptions(): Set<String> = disabledSubscriptions
+
     fun isValid(): Boolean = isUpstreamValid.get()
 
     companion object {

src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy

@@ -80,6 +80,7 @@ class GenericUpstreamMock extends GenericUpstream {
                 io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&finalizationDetectorBuilder,
                 [get: { null }] as java.util.function.Supplier,
                 null,
+                Collections.emptySet(),
         )
         this.ethereumHeadMock = this.getHead() as EthereumHeadMock
         setLag(0)

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy

@@ -81,6 +81,7 @@ class FilteredApisSpec extends Specification {
                     cs.&finalizationDetectorBuilder,
                     [get: { null }] as java.util.function.Supplier,
                     null,
+                    Collections.emptySet(),
             )
         }
         def matcher = new Selector.LabelMatcher("test", ["foo"])