ws subscription refactoring

added support of pending transactions subscription

src/main/kotlin/io/emeraldpay/dshackle/Global.kt

@@ -25,8 +25,10 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
 import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspentDeserializer
 import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
 import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspentDeserializer
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.TransactionIdSerializer
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.domain.TransactionId
 import java.text.SimpleDateFormat
 import java.util.Locale
 import java.util.TimeZone
@@ -84,6 +86,7 @@ class Global {
         private fun createObjectMapper(): ObjectMapper {
             val module = SimpleModule("EmeraldDshackle", Version(1, 0, 0, null, null, null))
             module.addSerializer(JsonRpcResponse::class.java, JsonRpcResponse.ResponseJsonSerializer())
+            module.addSerializer(TransactionId::class.java, TransactionIdSerializer())
 
             module.addDeserializer(EsploraUnspent::class.java, EsploraUnspentDeserializer())
             module.addDeserializer(RpcUnspent::class.java, RpcUnspentDeserializer())

src/main/kotlin/io/emeraldpay/dshackle/commons/DurableFlux.kt

@@ -0,0 +1,101 @@
+package io.emeraldpay.dshackle.commons
+
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import org.springframework.util.backoff.BackOff
+import org.springframework.util.backoff.BackOffExecution
+import org.springframework.util.backoff.ExponentialBackOff
+import org.springframework.util.backoff.FixedBackOff
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.time.Duration
+
+/**
+ * A flux holder that reconnects to it on failure taking into account a back off strategy
+ */
+class DurableFlux<T>(
+    private val provider: () -> Flux<T>,
+    private val errorBackOff: BackOff,
+    private val log: Logger,
+) {
+
+    companion object {
+        private val defaultLog = LoggerFactory.getLogger(DurableFlux::class.java)
+
+        @JvmStatic
+        fun newBuilder(): Builder<*> {
+            return Builder<Any>()
+        }
+    }
+
+    private var messagesSinceStart = 0
+    private var errorBackOffExecution = errorBackOff.start()
+
+    fun connect(): Flux<T> {
+        return provider.invoke()
+            .doOnNext {
+                if (messagesSinceStart == 0) {
+                    errorBackOffExecution = errorBackOff.start()
+                }
+                messagesSinceStart++
+            }
+            .doOnSubscribe {
+                messagesSinceStart = 0
+            }
+            .onErrorResume { t ->
+                val backoff = errorBackOffExecution.nextBackOff()
+                if (backoff != BackOffExecution.STOP) {
+                    log.warn("Connection closed with ${t.message}. Reconnecting in ${backoff}ms")
+                    connect().delaySubscription(Duration.ofMillis(backoff))
+                } else {
+                    log.warn("Connection closed with ${t.message}. Not reconnecting")
+                    Mono.error(t)
+                }
+            }
+    }
+
+    class Builder<T> {
+
+        private var provider: (() -> Flux<T>)? = null
+
+        protected var errorBackOff: BackOff = FixedBackOff(1_000, Long.MAX_VALUE)
+        protected var log: Logger = DurableFlux.defaultLog
+
+        @Suppress("UNCHECKED_CAST")
+        fun <X> using(provider: () -> Flux<X>): Builder<X> {
+            this.provider = provider as () -> Flux<T>
+            return this as Builder<X>
+        }
+
+        fun backoffOnError(time: Duration): Builder<T> {
+            errorBackOff = FixedBackOff(time.toMillis(), Long.MAX_VALUE)
+            return this
+        }
+
+        fun backoffOnError(time: Duration, multiplier: Double, max: Duration? = null): Builder<T> {
+            errorBackOff = ExponentialBackOff(time.toMillis(), multiplier).also {
+                if (max != null) {
+                    it.maxInterval = max.toMillis()
+                }
+            }
+            return this
+        }
+
+        fun backoffOnError(backOff: BackOff): Builder<T> {
+            errorBackOff = backOff
+            return this
+        }
+
+        fun logTo(log: Logger): Builder<T> {
+            this.log = log
+            return this
+        }
+
+        fun build(): DurableFlux<T> {
+            if (provider == null) {
+                throw IllegalStateException("No provider for original Flux")
+            }
+            return DurableFlux(provider!!, errorBackOff, log)
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/commons/ExpiringSet.kt

@@ -0,0 +1,152 @@
+package io.emeraldpay.dshackle.commons
+
+import org.apache.commons.collections4.iterators.UnmodifiableIterator
+import org.slf4j.LoggerFactory
+import java.time.Duration
+import java.util.LinkedList
+import java.util.TreeSet
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+/**
+ * A naive implementation of a Set with a limit for elements and an expiration time. Supposed to be used a filter for uniqueness.
+ * Internally it uses a TreeSet and a journal of added elements, which is used ot remove elements when they expire or the list grows too large.
+ * It's tread safe, but may be suboptimal to use in multithreaded scenario because of internal locks.
+ */
+class ExpiringSet<T>(
+    ttl: Duration,
+    comparator: Comparator<T>,
+    val limit: Int,
+) : MutableSet<T> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ExpiringSet::class.java)
+    }
+
+    private val tree = TreeSet<T>(comparator)
+    private val lock = ReentrantLock()
+    private val journal = LinkedList<JournalItem<T>>()
+    private var count = 0
+
+    private val ttl = ttl.toMillis()
+
+    data class JournalItem<T>(
+        val since: Long = System.currentTimeMillis(),
+        val value: T
+    ) {
+        fun isExpired(ttl: Long): Boolean {
+            return System.currentTimeMillis() > since + ttl
+        }
+    }
+
+    override val size: Int
+        get() = count
+
+    override fun clear() {
+        lock.withLock {
+            tree.clear()
+            journal.clear()
+            count = 0
+        }
+    }
+
+    override fun addAll(elements: Collection<T>): Boolean {
+        var changed = false
+        elements.forEach {
+            changed = changed || add(it)
+        }
+        return changed
+    }
+
+    override fun add(element: T): Boolean {
+        lock.withLock {
+            val added = tree.add(element)
+            if (added) {
+                journal.offer(JournalItem(value = element))
+                count++
+                shrink()
+            }
+            return added
+        }
+    }
+
+    override fun isEmpty(): Boolean {
+        return count == 0
+    }
+
+    override fun iterator(): MutableIterator<T> {
+        // not mutable
+        return UnmodifiableIterator.unmodifiableIterator(tree.iterator())
+    }
+
+    override fun retainAll(elements: Collection<T>): Boolean {
+        lock.withLock {
+            var changed = false
+            val iter = tree.iterator()
+            while (iter.hasNext()) {
+                val next = iter.next()
+                if (!elements.contains(next)) {
+                    changed = true
+                    iter.remove()
+                    count--
+                }
+            }
+            return changed
+        }
+    }
+
+    override fun removeAll(elements: Collection<T>): Boolean {
+        var changed = false
+        elements.forEach {
+            changed = changed || remove(it)
+        }
+        return changed
+    }
+
+    override fun remove(element: T): Boolean {
+        lock.withLock {
+            val changed = tree.remove(element)
+            if (changed) {
+                count--
+            }
+            return changed
+        }
+    }
+
+    override fun containsAll(elements: Collection<T>): Boolean {
+        return elements.all { contains(it) }
+    }
+
+    override fun contains(element: T): Boolean {
+        lock.withLock {
+            return tree.contains(element)
+        }
+    }
+
+    fun shrink() {
+        lock.withLock {
+            val iter = journal.iterator()
+            val removeAtLeast = (count - limit).coerceAtLeast(0)
+            var removed = 0
+            var stop = false
+            while (!stop && iter.hasNext()) {
+                val next = iter.next()
+                val overflow = removeAtLeast > removed
+                val expired = next.isExpired(ttl)
+                if (overflow || expired) {
+                    iter.remove()
+                    if (tree.remove(next.value)) {
+                        removed++
+                    }
+                }
+                // we always delete expired elements so don't stop on that
+                if (!expired) {
+                    // but if we already deleted all non-expired element (i.e., started because it grew too large)
+                    // then we stop as soon as we don't have any overflow
+                    stop = !overflow
+                }
+            }
+            count -= removed
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/commons/SharedFluxHolder.kt

@@ -0,0 +1,67 @@
+package io.emeraldpay.dshackle.commons
+
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.locks.ReentrantReadWriteLock
+import kotlin.concurrent.read
+import kotlin.concurrent.write
+
+/**
+ * A flux holder that that creates it only if requested. Keeps it for the following calls, so all the following calls will
+ * reuse it. Forgets as soon as it completes/cancelled, so it will be recreated again if needed.
+ */
+class SharedFluxHolder<T>(
+    /**
+     * Provider for the flux. Note that it can be called multiple times but only one is used at the same time.
+     * I.e., if there is a few calls because of a thread-race only one is kept.
+     * But once it's completed a new one may be created if requested.
+     */
+    private val provider: () -> Flux<T>
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(SharedFluxHolder::class.java)
+    }
+
+    private val ids = AtomicLong()
+    private val lock = ReentrantReadWriteLock()
+    private var current: Holder<T>? = null
+
+    fun get(): Flux<T> {
+        lock.read {
+            if (current != null) {
+                return current!!.flux
+            }
+        }
+        // The following doesn't consume resources because it's just create a Flux without actual subscription
+        // So even for the case of a thread race it's okay to create many. B/c only one is going to be kept as `current` and subscribed
+        val id = ids.incrementAndGet()
+        val created = Holder(
+            provider.invoke()
+                .share()
+                .doFinally { onClose(id) },
+            id
+        )
+        lock.write {
+            if (current != null) {
+                return current!!.flux
+            }
+            current = created
+        }
+        return created.flux
+    }
+
+    private fun onClose(id: Long) {
+        lock.write {
+            if (current?.id == id) {
+                current = null
+            }
+        }
+    }
+
+    data class Holder<T>(
+        val flux: Flux<T>,
+        val id: Long,
+    )
+}

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt

@@ -69,7 +69,7 @@ open class NativeSubscribe(
          * If not possible - performs subscription logic on the current instance
          * @see EthereumLikeMultistream.tryProxy
          */
-        val publisher = getUpstream(chain)?.tryProxy(matcher, request) ?: run {
+        val publisher = getUpstream(chain).tryProxy(matcher, request) ?: run {
             val method = request.method
             val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
                 objectMapper.readValue(it.newInput(), Map::class.java)
@@ -97,12 +97,10 @@ open class NativeSubscribe(
     }
 
     open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> =
-        getUpstream(chain)?.getSubscribe()?.subscribe(method, params, matcher)
+        getUpstream(chain).getSubscriptionApi().subscribe(method, params, matcher)
-            ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
 
-    private fun getUpstream(chain: Chain): EthereumLikeMultistream? =
+    private fun getUpstream(chain: Chain): EthereumLikeMultistream =
-        multistreamHolder.getUpstream(chain)
+        multistreamHolder.getUpstream(chain).let { it as EthereumLikeMultistream }
-            ?.let { it as EthereumLikeMultistream }
 
     fun convertToProto(holder: ResponseHolder): NativeSubscribeReplyItem {
         if (holder.response is NativeSubscribeReplyItem) {

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt

@@ -88,12 +88,11 @@ class TrackERC20Address(
         val asset = request.asset.code.lowercase(Locale.getDefault())
         val tokenDefinition = tokens[TokenId(chain, asset)] ?: return Flux.empty()
         val logs = getUpstream(chain)
-            .getSubscribe().logs
+            .getSubscriptionApi().logs
-            .start(
+            .create(
                 listOf(tokenDefinition.token.contract),
                 listOf(EventId.fromSignature("Transfer", "address", "address", "uint256")),
-                Selector.empty
+            ).connect(Selector.empty)
-            )
 
         return ethereumAddresses.extract(request.address)
             .map { TrackedAddress(chain, it, tokenDefinition.token, tokenDefinition.name) }

src/main/kotlin/io/emeraldpay/dshackle/upstream/BlockValidator.kt

@@ -11,6 +11,7 @@ interface BlockValidator {
     }
 
     companion object {
+        @JvmField
         val ALWAYS_VALID = AlwaysValid()
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/NoUpstreamSubscriptions.kt

@@ -0,0 +1,27 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+open class NoUpstreamSubscriptions : UpstreamSubscriptions {
+
+    companion object {
+        val DEFAULT = NoUpstreamSubscriptions()
+    }
+
+    override fun <T> get(method: String): SubscriptionConnect<T>? {
+        return null
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/SubscriptionConnect.kt

@@ -0,0 +1,26 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+import reactor.core.publisher.Flux
+
+/**
+ * Note that T is supposed to be serializable as JSON
+ */
+interface SubscriptionConnect<T> {
+
+    fun connect(matcher: Selector.Matcher): Flux<T>
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSubscriptions.kt

@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+/**
+ * Subscriptions available on the current upstream
+ */
+interface UpstreamSubscriptions {
+
+    fun <T> get(method: String): SubscriptionConnect<T>?
+}

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

@@ -29,7 +29,7 @@ import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 
 open class DefaultEthereumHead(
-    private val upstreamId: String,
+    protected val upstreamId: String,
     forkChoice: ForkChoice,
     blockValidator: BlockValidator
 ) : Head, AbstractHead(forkChoice, blockValidator, 60_000, upstreamId) {

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

@@ -8,7 +8,7 @@ import reactor.core.publisher.Flux
 
 interface EthereumLikeMultistream : Upstream {
     fun getReader(): EthereumCachingReader
-    fun getSubscribe(): EthereumSubscribe
+    fun getSubscriptionApi(): EthereumSubscriptionApi
 
     fun getHead(mather: Selector.Matcher): Head
 

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

@@ -23,6 +23,9 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.Lifecycle
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -48,7 +51,7 @@ open class EthereumMultistream(
         ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
 
     private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory())
-    private val subscribe = EthereumSubscribe(this)
+    private var subscribe = EthereumSubscriptionApi(this, NoPendingTxes())
 
     private val supportsEIP1559 = when (chain) {
         Chain.ETHEREUM, Chain.TESTNET_ROPSTEN, Chain.TESTNET_GOERLI, Chain.TESTNET_RINKEBY -> true
@@ -68,6 +71,24 @@ open class EthereumMultistream(
         super.init()
     }
 
+    override fun onUpstreamsUpdated() {
+        super.onUpstreamsUpdated()
+
+        val pendingTxes: PendingTxesSource = upstreams
+            .mapNotNull {
+                it.getUpstreamSubscriptions().getPendingTxes()
+            }.let {
+                if (it.isEmpty()) {
+                    NoPendingTxes()
+                } else if (it.size == 1) {
+                    it.first()
+                } else {
+                    AggregatedPendingTxes(it)
+                }
+            }
+        subscribe = EthereumSubscriptionApi(this, pendingTxes)
+    }
+
     override fun start() {
         super.start()
         reader.start()
@@ -158,7 +179,7 @@ open class EthereumMultistream(
         return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
     }
 
-    override fun getSubscribe(): EthereumSubscribe {
+    override fun getSubscriptionApi(): EthereumSubscriptionApi {
         return subscribe
     }
 

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

@@ -69,6 +69,11 @@ open class EthereumRpcUpstream(
                 .subscribe(this::setStatus)
         }
     }
+
+    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+        return connector.getUpstreamSubscriptions()
+    }
+
     override fun getHead(): Head {
         return connector.getHead()
     }

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

@@ -4,17 +4,24 @@ import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectNewHeads
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectSyncing
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.hex.Hex32
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 
-open class EthereumSubscribe(
+open class EthereumSubscriptionApi(
-    val upstream: EthereumLikeMultistream
+    val upstream: EthereumLikeMultistream,
+    val pendingTxesSource: PendingTxesSource
 ) {
 
     companion object {
-        private val log = LoggerFactory.getLogger(EthereumSubscribe::class.java)
+        private val log = LoggerFactory.getLogger(EthereumSubscriptionApi::class.java)
+
+        const val METHOD_NEW_HEADS = "newHeads"
+        const val METHOD_LOGS = "logs"
+        const val METHOD_SYNCING = "syncing"
+        const val METHOD_PENDING_TXES = "newPendingTransactions"
     }
 
     private val newHeads = ConnectNewHeads(upstream)
@@ -23,10 +30,10 @@ open class EthereumSubscribe(
 
     @Suppress("UNCHECKED_CAST")
     open fun subscribe(method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
-        if (method == "newHeads") {
+        if (method == METHOD_NEW_HEADS) {
             return newHeads.connect(matcher)
         }
-        if (method == "logs") {
+        if (method == METHOD_LOGS) {
             val paramsMap = try {
                 if (params != null && Map::class.java.isAssignableFrom(params.javaClass)) {
                     readLogsRequest(params as Map<String, Any?>)
@@ -36,10 +43,13 @@ open class EthereumSubscribe(
             } catch (t: Throwable) {
                 return Flux.error(UnsupportedOperationException("Invalid parameter for $method. Error: ${t.message}"))
             }
-            return logs.start(paramsMap.address, paramsMap.topics, matcher)
+            return logs.create(paramsMap.address, paramsMap.topics).connect(matcher)
         }
-        if (method == "syncing") {
+        if (method == METHOD_SYNCING) {
-            return syncing.connect()
+            return syncing.connect(matcher)
+        }
+        if (method == METHOD_PENDING_TXES) {
+            return pendingTxesSource.connect(matcher)
         }
         return Flux.error(UnsupportedOperationException("Method $method is not supported"))
     }

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

@@ -44,4 +44,6 @@ abstract class EthereumUpstream(
     override fun getLabels(): Collection<UpstreamsConfig.Labels> {
         return node?.let { listOf(it.labels) } ?: emptyList()
     }
+
+    abstract fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions
 }

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

@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.upstream.UpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
+
+interface EthereumUpstreamSubscriptions : UpstreamSubscriptions {
+
+    fun getPendingTxes(): PendingTxesSource?
+}

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

@@ -58,11 +58,11 @@ class EthereumWsFactory(
         )
     }
 
-    fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?): WsConnection {
+    fun create(upstream: DefaultUpstream?): WsConnectionImpl {
         require(upstream == null || upstream.getId() == id) {
             "Creating instance for different upstream. ${upstream?.getId()} != id"
         }
-        return WsConnection(id, uri, origin, basicAuth, metrics, upstream, validator).also { ws ->
+        return WsConnectionImpl(id, uri, origin, basicAuth, metrics, upstream).also { ws ->
             config?.frameSize?.let {
                 ws.frameSize = it
             }

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

@@ -16,19 +16,32 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.SilentException
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.rpc.json.BlockJson
+import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+import reactor.retry.Repeat
+import java.time.Duration
 
 class EthereumWsHead(
-    private val ws: WsConnection,
     upstreamId: String,
     forkChoice: ForkChoice,
-    blockValidator: BlockValidator
+    blockValidator: BlockValidator,
+    private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val wsSubscriptions: WsSubscriptions,
 ) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)
@@ -44,12 +57,53 @@ class EthereumWsHead(
         this.subscription?.dispose()
         val heads = Flux.merge(
             // get the current block, not just wait for the next update
-            getLatestBlock(JsonRpcWsClient(ws)),
+            getLatestBlock(api),
-            ws.getBlocksFlux()
+            listenNewHeads()
         )
         this.subscription = super.follow(heads)
     }
 
+    fun listenNewHeads(): Flux<BlockContainer> {
+        return wsSubscriptions.subscribe("newHeads")
+            .map {
+                Global.objectMapper.readValue(it, BlockJson::class.java) as BlockJson<TransactionRefJson>
+            }
+            .flatMap { block ->
+                // newHeads returns incomplete blocks, i.e. without some fields and without transaction hashes,
+                // so we need to fetch the full block data
+                if (block.difficulty == null || block.transactions == null) {
+                    // TODO do we really need this ?
+                    enhanceRealBlock(block)
+                } else {
+                    Mono.just(BlockContainer.from(block))
+                }
+            }
+    }
+
+    fun enhanceRealBlock(block: BlockJson<TransactionRefJson>): Mono<BlockContainer> {
+        return Mono.just(block.hash)
+            .flatMap { hash ->
+                api.read(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false)))
+                    .flatMap { resp ->
+                        if (resp.isNull()) {
+                            Mono.error(SilentException("Received null for block $hash"))
+                        } else {
+                            Mono.just(resp)
+                        }
+                    }
+                    .flatMap(JsonRpcResponse::requireResult)
+                    .map { BlockContainer.fromEthereumJson(it, upstreamId) }
+                    .subscribeOn(Schedulers.boundedElastic())
+                    .timeout(Defaults.timeoutInternal, Mono.empty())
+            }.repeatWhenEmpty { n ->
+                Repeat.times<Any>(5)
+                    .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500))
+                    .apply(n)
+            }
+            .timeout(Defaults.timeout, Mono.empty())
+            .onErrorResume { Mono.empty() }
+    }
+
     override fun stop() {
         super.stop()
         subscription?.dispose()

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

@@ -0,0 +1,31 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.upstream.NoUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
+
+class NoEthereumUpstreamSubscriptions : NoUpstreamSubscriptions(), EthereumUpstreamSubscriptions {
+
+    companion object {
+        @JvmStatic
+        val DEFAULT = NoEthereumUpstreamSubscriptions()
+    }
+
+    override fun getPendingTxes(): PendingTxesSource? {
+        return null
+    }
+}

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

@@ -17,20 +17,17 @@ package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.Global
-import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.config.AuthConfig
-import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage
 import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.emeraldpay.etherjar.rpc.RpcResponseError
-import io.emeraldpay.etherjar.rpc.json.BlockJson
-import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import io.netty.buffer.ByteBuf
 import io.netty.buffer.ByteBufInputStream
 import io.netty.buffer.Unpooled
@@ -51,7 +48,6 @@ import reactor.netty.http.client.HttpClient
 import reactor.netty.http.client.WebsocketClientSpec
 import reactor.netty.http.websocket.WebsocketInbound
 import reactor.netty.http.websocket.WebsocketOutbound
-import reactor.retry.Repeat
 import reactor.util.function.Tuples
 import java.net.URI
 import java.time.Duration
@@ -60,26 +56,22 @@ import java.util.Base64
 import java.util.concurrent.Executors
 import java.util.concurrent.ScheduledFuture
 import java.util.concurrent.TimeUnit
-import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 
-open class WsConnection(
+open class WsConnectionImpl(
     private val id: String,
     private val uri: URI,
     private val origin: URI,
     private val basicAuth: AuthConfig.ClientBasicAuth?,
     private val rpcMetrics: RpcMetrics?,
     private val upstream: DefaultUpstream?,
-    private val validator: EthereumUpstreamValidator?
 ) : AutoCloseable {
 
     companion object {
-        private val log = LoggerFactory.getLogger(WsConnection::class.java)
+        private val log = LoggerFactory.getLogger(WsConnectionImpl::class.java)
 
         private const val IDS_START = 100
-        private const val START_REQUEST =
-            "{\"jsonrpc\":\"2.0\", \"method\":\"eth_subscribe\", \"id\":\"blocks\", \"params\":[\"newHeads\"]}"
 
         // WebSocket Frame limit.
         // Default is 65_536, but Geth responds with larger frames,
@@ -108,26 +100,22 @@ open class WsConnection(
     private val resetBackoffExecutor = Executors.newScheduledThreadPool(2)
     private var resetBackoffTask: ScheduledFuture<Unit>? = null
 
-    private val blocks = Sinks
+    private val messages = Sinks
         .many()
         .multicast()
-        .directBestEffort<BlockContainer>()
+        .directBestEffort<ResponseWSParser.WsResponse>()
+
     private var rpcSend = Sinks
         .many()
         .unicast()
         .onBackpressureBuffer<JsonRpcRequest>()
-    private val rpcReceive = Sinks
+
-        .many()
-        .multicast()
-        .directBestEffort<JsonRpcResponse>()
     private val disconnects = Sinks
         .many()
         .multicast()
         .directBestEffort<Instant>()
     private val sendIdSeq = AtomicInteger(IDS_START)
-    private val sendExecutor = Executors.newFixedThreadPool(
+    private val sendExecutor = Executors.newSingleThreadExecutor()
-        1.coerceAtLeast(Runtime.getRuntime().availableProcessors() / 2)
-    )
     private var keepConnection = true
     private var connection: Disposable? = null
     private val reconnecting = AtomicBoolean(false)
@@ -183,7 +171,6 @@ open class WsConnection(
 
     private fun connectInternal() {
         log.info("Connecting to WebSocket: $uri")
-        log.info("Available processors: ${Runtime.getRuntime().availableProcessors()}")
         connection?.dispose()
         connection = HttpClient.create()
             .resolver(DefaultAddressResolverGroup.INSTANCE)
@@ -236,6 +223,7 @@ open class WsConnection(
     }
 
     fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> {
+        var read = false
         val consumer = inbound
             .aggregateFrames(msgSizeLimit)
             .receiveFrames()
@@ -244,11 +232,17 @@ open class WsConnection(
             .flatMap {
                 try {
                     val msg = parser.parse(it)
-                    if (msg.type == ResponseWSParser.Type.SUBSCRIPTION) {
+                    if (!read) {
-                        onSubscription(msg)
+                        if (msg.error != null) {
-                    } else {
+                            log.warn("Received error ${msg.error.code} from $uri: ${msg.error.message}")
-                        onRpc(msg)
+                        } else {
+                            // restart backoff only after a successful read from the connection,
+                            // otherwise it may restart it even if the connection is faulty or responds only with error
+                            currentBackOff = reconnectBackoff.start()
+                            read = true
+                        }
                     }
+                    onMessage(msg)
                 } catch (t: Throwable) {
                     log.warn("Failed to process WS message. ${t.javaClass}: ${t.message}")
                     Mono.empty()
@@ -265,117 +259,57 @@ open class WsConnection(
         val calls = rpcSend
             .asFlux()
             .map {
-                Unpooled.wrappedBuffer(it.toJson())
+                Unpooled.wrappedBuffer(Global.objectMapper.writeValueAsBytes(it))
             }
 
         return outbound.send(
             Flux.merge(
-                startWhenValidated(),
                 calls.subscribeOn(Schedulers.boundedElastic()),
                 consumer.then(Mono.empty<ByteBuf>()).subscribeOn(Schedulers.boundedElastic())
             )
         )
     }
 
-    /**
+    fun onMessage(msg: ResponseWSParser.WsResponse): Mono<Void> {
-     * Starts subscriptions ('newHeads') when the upstream is fully validated. If upstream is invalid it breaks flow with an Error.
-     * I.e., the first requests are made from a Validator and when it returns OK the Connection continues with other stuff.
-     */
-    fun startWhenValidated(): Publisher<ByteBuf> {
-        val start = Mono.just(START_REQUEST).map {
-            Unpooled.wrappedBuffer(it.toByteArray())
-        }
-
-        return if (validator != null) {
-            validator.validate()
-                .timeout(
-                    Defaults.timeoutInternal,
-                    Mono.fromCallable { log.warn("Not received a validation result from $uri") }.then(Mono.error(TimeoutException()))
-                )
-                .flatMap {
-                    if (it == UpstreamAvailability.OK) {
-                        start
-                    } else {
-                        tryReconnectLater()
-                        Mono.error(IllegalStateException("Upstream $uri is not ready"))
-                    }
-                }
-        } else {
-            start
-        }
-    }
-
-    fun onRpc(msg: ResponseWSParser.WsResponse): Mono<Void> {
-        return if (msg.id.isNumber()) {
-            val resp = JsonRpcResponse(
-                msg.value,
-                msg.error,
-                msg.id,
-                null
-            )
-            Mono.fromCallable {
-                val status = rpcReceive.tryEmitNext(resp)
-                if (status.isFailure) {
-                    if (status == Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) {
-                        log.debug("No subscribers to WS response")
-                    } else {
-                        log.warn("Failed to proceed with a RPC message: $status")
-                    }
-                }
-            }.then()
-        } else {
-            // it's a response to the newHeads subscription, just ignore it
-            Mono.empty<Void>()
-        }
-    }
-
-    fun onSubscription(msg: ResponseWSParser.WsResponse): Mono<Void> {
-        if (msg.error != null) {
-            return Mono.error(IllegalStateException("Received error from WS upstream: ${msg.error.message}"))
-        }
-        // we always expect an answer to the `newHeads`, since we are not initiating any other subscriptions
         return Mono.fromCallable {
-            Global.objectMapper.readValue(msg.value, BlockJson::class.java) as BlockJson<TransactionRefJson>
+            val status = messages.tryEmitNext(msg)
-        }.flatMap { onNewHeads(it) }.then()
+            if (status.isFailure) {
+                if (status == Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) {
+                    log.debug("No subscribers to WS response")
+                } else {
+                    log.warn("Failed to proceed with a WS message: $status")
+                }
+            }
+        }.then()
     }
 
-    fun onNewHeads(block: BlockJson<TransactionRefJson>): Mono<Void> {
+    open fun getRpcResponses(): Flux<JsonRpcResponse> {
-        // newHeads returns incomplete blocks, i.e. without some fields and without transaction hashes,
+        return Flux.from(messages.asFlux())
-        // so we need to fetch the full block data
+            .publishOn(Schedulers.boundedElastic())
-        return if (block.difficulty == null || block.transactions == null) {
+            .filter {
-            Mono.just(block.hash)
+                it.type == ResponseWSParser.Type.RPC
-                .flatMap { hash ->
+            }
-                    call(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false)))
+            .map { msg ->
-                        .flatMap { resp ->
+                JsonRpcResponse(
-                            if (resp.isNull()) {
+                    msg.value, msg.error, msg.id, null
-                                Mono.error(SilentException("Received null for block $hash"))
+                )
-                            } else {
+            }
-                                Mono.just(resp)
-                            }
-                        }
-                        .flatMap(JsonRpcResponse::requireResult)
-                        .map { BlockContainer.fromEthereumJson(it, id) }
-                        .subscribeOn(Schedulers.boundedElastic())
-                        .timeout(Defaults.timeoutInternal, Mono.empty())
-                }.repeatWhenEmpty { n ->
-                    Repeat.times<Any>(5)
-                        .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500))
-                        .apply(n)
-                }
-                .timeout(Defaults.timeout, Mono.empty())
-                .onErrorResume { Mono.empty() }
-                .doOnNext {
-                    blocks.tryEmitNext(it)
-                }
-                .then()
-        } else {
-            Mono.fromCallable {
-                blocks.tryEmitNext(BlockContainer.from(block, id))
-            }.then()
-        }
     }
 
-    fun call(originalRequest: JsonRpcRequest): Mono<JsonRpcResponse> {
+    open fun getSubscribeResponses(): Flux<JsonRpcWsMessage> {
+        return Flux.from(messages.asFlux())
+            .publishOn(Schedulers.boundedElastic())
+            .filter {
+                it.type == ResponseWSParser.Type.SUBSCRIPTION
+            }
+            .map { msg ->
+                JsonRpcWsMessage(
+                    msg.value, msg.error, msg.id.asString(),
+                )
+            }
+    }
+
+    open fun callRpc(originalRequest: JsonRpcRequest): Mono<JsonRpcResponse> {
         return Mono.fromCallable {
             val startTime = System.nanoTime()
             // use an internal id sequence, to avoid id conflicts with user calls
@@ -387,10 +321,13 @@ open class WsConnection(
         }
     }
 
-    fun sendRpc(request: JsonRpcRequest) {
+    private fun sendRpc(request: JsonRpcRequest) {
         // submit to upstream in a separate thread, to free current thread (needs for subscription, etc)
         sendExecutor.execute {
-            rpcSend.emitNext(request) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+            val result = rpcSend.tryEmitNext(request)
+            if (result.isFailure) {
+                log.warn("Failed to send RPC request: $result")
+            }
         }
     }
 
@@ -405,7 +342,7 @@ open class WsConnection(
             false
         )
 
-        val response = Flux.from(rpcReceive.asFlux())
+        val response = Flux.from(getRpcResponses())
             .doOnSubscribe { sendRpc(request) }
             .filter { resp -> resp.id.asNumber() == expectedId }
             .take(Defaults.timeout)
@@ -432,10 +369,6 @@ open class WsConnection(
             .switchIfEmpty(Mono.error(noResponse))
     }
 
-    fun getBlocksFlux(): Flux<BlockContainer> {
-        return this.blocks.asFlux()
-    }
-
     override fun close() {
         log.info("Closing connection to WebSocket $uri")
         keepConnection = false

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

@@ -0,0 +1,41 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import reactor.core.publisher.Flux
+
+/**
+ * A JSON-RPC Subscription client.
+ * In general, it's a Websocket extension for JSON RPC that allows multiple responses to the same method.
+ *
+ * Example:
+ *
+ * <pre><code>
+ * > {"id": 1, "method": "eth_subscribe", "params": ["newPendingTransactions"]}
+ * > {"jsonrpc":"2.0","id":1,"result":"0xcff45d00e77e8e050d919daf284516c8"}
+ * > {"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xcff45d00e77e8e050d919daf284516c8","result":"0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c"}}
+ * > {"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xcff45d00e77e8e050d919daf284516c8","result":"0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e"}}
+ * > {"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xcff45d00e77e8e050d919daf284516c8","result":"0x67f22a3b441ea312306f97694ca8159f8d6faaccf0f5ce6442c84b13991f1d23"}}
+ * </code></pre>
+ *
+ * */
+interface WsSubscriptions {
+
+    /**
+     * Subscribe on remote
+     */
+    fun subscribe(method: String): Flux<ByteArray>
+}

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

@@ -0,0 +1,54 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.atomic.AtomicReference
+
+class WsSubscriptionsImpl(
+    val conn: WsConnectionImpl,
+) : WsSubscriptions {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(WsSubscriptionsImpl::class.java)
+    }
+
+    private val ids = AtomicLong(1)
+
+    override fun subscribe(method: String): Flux<ByteArray> {
+        val subscriptionId = AtomicReference("")
+        val messages = conn.getSubscribeResponses()
+            .filter { it.subscriptionId == subscriptionId.get() }
+            .filter { it.result != null } // should never happen
+            .map { it.result!! }
+
+        return conn.callRpc(JsonRpcRequest("eth_subscribe", listOf(method), ids.incrementAndGet()))
+            .flatMapMany {
+                if (it.hasError()) {
+                    log.warn("Failed to establish ETH Subscription: ${it.error?.message}")
+                    Mono.error(JsonRpcException(it.id, it.error!!))
+                } else {
+                    subscriptionId.set(it.getResultAsProcessedString())
+                    messages
+                }
+            }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.connectors
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 
@@ -10,4 +11,6 @@ interface EthereumConnector : Lifecycle {
     fun getHead(): Head
 
     fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
+
+    fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt

@@ -29,7 +29,7 @@ open class EthereumConnectorFactory(
 
     override fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector {
         if (wsFactory != null && !preferHttp) {
-            return EthereumWsConnector(wsFactory, upstream, validator, forkChoice, blockValidator)
+            return EthereumWsConnector(wsFactory, upstream, forkChoice, blockValidator)
         }
         if (httpFactory == null) {
             throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set")

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt

@@ -7,10 +7,7 @@ import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.MergedHead
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead
+import io.emeraldpay.dshackle.upstream.ethereum.*
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
 import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -25,7 +22,7 @@ class EthereumRpcConnector(
     forkChoice: ForkChoice,
     blockValidator: BlockValidator
 ) : EthereumConnector, CachesEnabled {
-    private val conn: WsConnection?
+    private val conn: WsConnectionImpl?
     private val head: Head
 
     companion object {
@@ -35,15 +32,16 @@ class EthereumRpcConnector(
     init {
         if (wsFactory != null) {
             // do not set upstream to the WS, since it doesn't control the RPC upstream
-            conn = wsFactory.create(null, null)
+            conn = wsFactory.create(null)
-            val wsHead = EthereumWsHead(conn, id, AlwaysForkChoice(), blockValidator)
+            val subscriptions = WsSubscriptionsImpl(conn)
-            // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
+            val wsHead = EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getApi(), subscriptions)
-            val rpcHead = EthereumRpcHead(directReader, AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
+            // receive all new blocks through WebSockets, but also periodically verify with RPC in case if WS failed
+            val rpcHead = EthereumRpcHead(getApi(), AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
             head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id")
         } else {
             conn = null
             log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC")
-            head = EthereumRpcHead(directReader, forkChoice, id, blockValidator)
+            head = EthereumRpcHead(getApi(), forkChoice, id, blockValidator)
         }
     }
 
@@ -78,6 +76,10 @@ class EthereumRpcConnector(
         return directReader
     }
 
+    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+        return NoEthereumUpstreamSubscriptions.DEFAULT
+    }
+
     override fun getHead(): Head {
         return head
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt

@@ -4,10 +4,8 @@ import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
+import io.emeraldpay.dshackle.upstream.ethereum.*
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsSubscriptions
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -16,18 +14,20 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 class EthereumWsConnector(
     wsFactory: EthereumWsFactory,
     upstream: DefaultUpstream,
-    validator: EthereumUpstreamValidator,
     forkChoice: ForkChoice,
     blockValidator: BlockValidator
 ) : EthereumConnector {
-    private val conn: WsConnection
+    private val conn: WsConnectionImpl
     private val api: Reader<JsonRpcRequest, JsonRpcResponse>
     private val head: EthereumWsHead
+    private val subscriptions: EthereumUpstreamSubscriptions
 
     init {
-        conn = wsFactory.create(upstream, validator)
+        conn = wsFactory.create(upstream)
-        head = EthereumWsHead(conn, upstream.getId(), forkChoice, blockValidator)
         api = JsonRpcWsClient(conn)
+        val wsSubscriptions = WsSubscriptionsImpl(conn)
+        head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, api, wsSubscriptions)
+        subscriptions = EthereumWsSubscriptions(wsSubscriptions)
     }
 
     override fun start() {
@@ -48,6 +48,10 @@ class EthereumWsConnector(
         return api
     }
 
+    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+        return subscriptions
+    }
+
     override fun getHead(): Head {
         return head
     }

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

@@ -0,0 +1,45 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.commons.ExpiringSet
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.etherjar.domain.TransactionId
+import io.emeraldpay.etherjar.hex.HexDataComparator
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import java.time.Duration
+
+class AggregatedPendingTxes(
+    private val sources: List<PendingTxesSource>
+) : PendingTxesSource {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(AggregatedPendingTxes::class.java)
+    }
+
+    private val track = ExpiringSet<TransactionId>(
+        Duration.ofSeconds(30),
+        HexDataComparator() as Comparator<TransactionId>,
+        10_000
+    )
+
+    override fun connect(matcher: Selector.Matcher): Flux<TransactionId> {
+        return Flux.merge(
+            sources.map { it.connect(matcher) } // todo check
+        ).filter(track::add)
+    }
+}

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

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
@@ -33,7 +34,7 @@ import kotlin.concurrent.write
 
 class ConnectBlockUpdates(
     private val upstream: EthereumLikeMultistream
-) {
+) : SubscriptionConnect<ConnectBlockUpdates.Update> {
 
     companion object {
         private val log = LoggerFactory.getLogger(ConnectBlockUpdates::class.java)
@@ -49,7 +50,7 @@ class ConnectBlockUpdates(
     private val connected: MutableMap<String, Flux<Update>> = ConcurrentHashMap()
 
     fun connect() = connect(Selector.empty)
-    fun connect(matcher: Selector.Matcher): Flux<Update> {
+    override fun connect(matcher: Selector.Matcher): Flux<Update> {
         return connected.computeIfAbsent(matcher.describeInternal()) { key ->
             extract(upstream.getHead(matcher))
                 .publishOn(Schedulers.boundedElastic())

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

@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
 import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.etherjar.domain.Address
@@ -45,14 +46,18 @@ open class ConnectLogs(
         return produceLogs.produce(connectBlockUpdates.connect(matcher))
     }
 
-    open fun start(addresses: List<Address>, topics: List<Hex32>, matcher: Selector.Matcher): Flux<LogMessage> {
+    open fun create(addresses: List<Address>, topics: List<Hex32>): SubscriptionConnect<LogMessage> {
-        // shortcut to the whole output if we don't have any filters
+        return object : SubscriptionConnect<LogMessage> {
-        if (addresses.isEmpty() && topics.isEmpty()) {
+            override fun connect(matcher: Selector.Matcher): Flux<LogMessage> {
-            return start(matcher)
+                // shortcut to the whole output if we don't have any filters
+                if (addresses.isEmpty() && topics.isEmpty()) {
+                    return start(matcher)
+                }
+                // filtered output
+                return start(matcher)
+                    .transform(filtered(addresses, topics))
+            }
         }
-        // filtered output
-        return start(matcher)
-            .transform(filtered(addresses, topics))
     }
 
     fun filtered(addresses: List<Address>, topics: List<Hex32>): Function<Flux<LogMessage>, Flux<LogMessage>> {

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

@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
 import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage
 import org.slf4j.LoggerFactory
@@ -29,7 +30,7 @@ import java.util.concurrent.ConcurrentHashMap
  */
 class ConnectNewHeads(
     private val upstream: EthereumLikeMultistream
-) {
+) : SubscriptionConnect<NewHeadMessage> {
 
     companion object {
         private val log = LoggerFactory.getLogger(ConnectNewHeads::class.java)
@@ -37,7 +38,7 @@ class ConnectNewHeads(
 
     private val connected: MutableMap<String, Flux<NewHeadMessage>> = ConcurrentHashMap()
 
-    fun connect(matcher: Selector.Matcher): Flux<NewHeadMessage> =
+    override fun connect(matcher: Selector.Matcher): Flux<NewHeadMessage> =
         connected.computeIfAbsent(matcher.describeInternal()) { key ->
             ProduceNewHeads(upstream.getHead(matcher))
                 .start()

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

@@ -15,6 +15,8 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import org.slf4j.LoggerFactory
@@ -25,7 +27,7 @@ import kotlin.concurrent.withLock
 
 class ConnectSyncing(
     private val upstream: EthereumLikeMultistream
-) {
+) : SubscriptionConnect<Boolean> {
 
     companion object {
         private val log = LoggerFactory.getLogger(ConnectSyncing::class.java)
@@ -34,7 +36,7 @@ class ConnectSyncing(
     private var connected: Flux<Boolean>? = null
     private val connectLock = ReentrantLock()
 
-    fun connect(): Flux<Boolean> {
+    override fun connect(matcher: Selector.Matcher): Flux<Boolean> {
         val current = connected
         if (current != null) {
             return current

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

@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.commons.DurableFlux
+import io.emeraldpay.dshackle.commons.SharedFluxHolder
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
+import io.emeraldpay.etherjar.domain.TransactionId
+import reactor.core.publisher.Flux
+import java.time.Duration
+
+abstract class DefaultPendingTxesSource : SubscriptionConnect<TransactionId>, PendingTxesSource {
+
+    private val connectionSource = DurableFlux
+        .newBuilder()
+        .using(::createConnection)
+        .backoffOnError(Duration.ofMillis(100), 1.5, Duration.ofSeconds(60))
+        .build()
+    private val holder = SharedFluxHolder<TransactionId>(
+        connectionSource::connect
+    )
+
+    override fun connect(matcher: Selector.Matcher): Flux<TransactionId> {
+        return holder.get()
+    }
+
+    abstract fun createConnection(): Flux<TransactionId>
+}

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

@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeReplyItem
+import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeRequest
+import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.etherjar.domain.TransactionId
+import reactor.core.publisher.Flux
+
+class DshacklePendingTxesSource(
+    private val blockchain: Chain,
+    private val conn: ReactorBlockchainGrpc.ReactorBlockchainStub,
+) : PendingTxesSource, DefaultPendingTxesSource() {
+
+    private val request = NativeSubscribeRequest.newBuilder()
+        .setChainValue(blockchain.id)
+        .setMethod(EthereumSubscriptionApi.METHOD_PENDING_TXES)
+        .build()
+
+    var available = false
+
+    override fun createConnection(): Flux<TransactionId> {
+        if (!available) {
+            return Flux.empty()
+        }
+        return conn
+            .nativeSubscribe(request)
+            .map(::readResponse)
+            .map(TransactionId::from)
+    }
+
+    fun readResponse(resp: NativeSubscribeReplyItem): String {
+        // comes as a string, so cut off the quotes
+        return resp.payload.substring(1, resp.payload.size() - 1).toStringUtf8()
+    }
+
+    fun update(conf: BlockchainOuterClass.DescribeChain) {
+        available = conf.supportedMethodsList.any {
+            it == EthereumSubscriptionApi.METHOD_PENDING_TXES
+        }
+    }
+}

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

@@ -0,0 +1,47 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
+import io.emeraldpay.dshackle.upstream.UpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+
+class EthereumDshackleSubscriptions(
+    blockchain: Chain,
+    conn: ReactorBlockchainGrpc.ReactorBlockchainStub,
+) : UpstreamSubscriptions, EthereumUpstreamSubscriptions {
+
+    private val pendingTxes = DshacklePendingTxesSource(blockchain, conn)
+
+    override fun <T> get(method: String): SubscriptionConnect<T>? {
+        if (method == EthereumSubscriptionApi.METHOD_PENDING_TXES) {
+            return pendingTxes as SubscriptionConnect<T>
+        }
+        return null
+    }
+
+    fun update(conf: BlockchainOuterClass.DescribeChain) {
+        pendingTxes.update(conf)
+    }
+
+    override fun getPendingTxes(): PendingTxesSource? {
+        return pendingTxes
+    }
+}

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

@@ -0,0 +1,45 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
+import io.emeraldpay.dshackle.upstream.UpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
+import org.slf4j.LoggerFactory
+
+class EthereumWsSubscriptions(
+    private val conn: WsSubscriptions
+) : UpstreamSubscriptions, EthereumUpstreamSubscriptions {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EthereumWsSubscriptions::class.java)
+    }
+
+    private val pendingTxes = WebsocketPendingTxes(conn)
+
+    override fun <T> get(method: String): SubscriptionConnect<T>? {
+        if (method == EthereumSubscriptionApi.METHOD_PENDING_TXES) {
+            return pendingTxes as SubscriptionConnect<T>
+        }
+        return null
+    }
+
+    override fun getPendingTxes(): PendingTxesSource? {
+        return pendingTxes
+    }
+}

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

@@ -0,0 +1,34 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.etherjar.domain.TransactionId
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+
+class NoPendingTxes : PendingTxesSource {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(NoPendingTxes::class.java)
+
+        val DEFAULT = NoPendingTxes()
+    }
+
+    override fun connect(matched: Selector.Matcher): Flux<TransactionId> {
+        return Flux.empty()
+    }
+}

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

@@ -0,0 +1,25 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
+import io.emeraldpay.etherjar.domain.TransactionId
+
+/**
+ * A source to subscribe to newPendingTransactions.
+ * When using a Websocket RPC on a node, it produces hashes of new transactions received.
+ */
+interface PendingTxesSource : SubscriptionConnect<TransactionId>

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

@@ -0,0 +1,47 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
+import io.emeraldpay.etherjar.domain.TransactionId
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.time.Duration
+
+class WebsocketPendingTxes(
+    private val wsSubscriptions: WsSubscriptions
+) : DefaultPendingTxesSource(), SubscriptionConnect<TransactionId> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(WebsocketPendingTxes::class.java)
+    }
+
+    override fun createConnection(): Flux<TransactionId> {
+        return wsSubscriptions.subscribe(EthereumSubscriptionApi.METHOD_PENDING_TXES)
+            .timeout(Duration.ofSeconds(60), Mono.empty())
+            .map {
+                // comes as a JS string, i.e., within quotes
+                val value = ByteArray(it.size - 2)
+                System.arraycopy(it, 1, value, 0, value.size)
+                TransactionId.from(String(value))
+            }
+            .doOnError { t -> log.warn("Invalid pending transaction", t) }
+            .onErrorResume { Mono.empty() }
+    }
+}

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

@@ -0,0 +1,37 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json
+
+import com.fasterxml.jackson.core.JsonGenerator
+import com.fasterxml.jackson.databind.JsonSerializer
+import com.fasterxml.jackson.databind.SerializerProvider
+import io.emeraldpay.etherjar.domain.TransactionId
+import org.slf4j.LoggerFactory
+
+class TransactionIdSerializer : JsonSerializer<TransactionId>() {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(TransactionIdSerializer::class.java)
+    }
+
+    override fun serialize(value: TransactionId?, gen: JsonGenerator, serializers: SerializerProvider) {
+        if (value == null) {
+            gen.writeNull()
+            return
+        }
+        gen.writeString(value.toHex())
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -29,6 +29,9 @@ import io.emeraldpay.dshackle.upstream.MergedHead
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
 import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -52,8 +55,8 @@ open class EthereumPosMultiStream(
     private var head: Head? = null
 
     private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory())
+    private var subscribe = EthereumSubscriptionApi(this, NoPendingTxes())
     private val feeEstimation = EthereumPriorityFees(this, reader, 256)
-    private val subscribe = EthereumSubscribe(this)
     private val filteredHeads: MutableMap<String, Head> =
         ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
 
@@ -157,7 +160,7 @@ open class EthereumPosMultiStream(
         return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
     }
 
-    override fun getSubscribe(): EthereumSubscribe {
+    override fun getSubscriptionApi(): EthereumSubscriptionApi {
         return subscribe
     }
 
@@ -182,4 +185,22 @@ open class EthereumPosMultiStream(
     override fun getFeeEstimation(): ChainFees {
         return feeEstimation
     }
+
+    override fun onUpstreamsUpdated() {
+        super.onUpstreamsUpdated()
+
+        val pendingTxes: PendingTxesSource = upstreams
+            .mapNotNull {
+                it.getUpstreamSubscriptions().getPendingTxes()
+            }.let {
+                if (it.isEmpty()) {
+                    NoPendingTxes()
+                } else if (it.size == 1) {
+                    it.first()
+                } else {
+                    AggregatedPendingTxes(it)
+                }
+            }
+        subscribe = EthereumSubscriptionApi(this, pendingTxes)
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt

@@ -98,4 +98,8 @@ open class EthereumPosRpcUpstream(
         }
         return this as T
     }
+
+    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+        return connector.getUpstreamSubscriptions()
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt

@@ -44,4 +44,6 @@ abstract class EthereumPosUpstream(
     override fun getLabels(): Collection<UpstreamsConfig.Labels> {
         return node?.let { listOf(it.labels) } ?: emptyList()
     }
+
+    abstract fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt

@@ -17,7 +17,7 @@
 package io.emeraldpay.dshackle.upstream.grpc
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
-import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.api.proto.ReactorBlockchainGrpc.ReactorBlockchainStub
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
@@ -32,6 +32,8 @@ import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleSubscriptions
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -53,7 +55,7 @@ open class EthereumGrpcUpstream(
     hash: Byte,
     role: UpstreamsConfig.UpstreamRole,
     private val chain: Chain,
-    private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
+    private val remote: ReactorBlockchainStub,
     private val client: JsonRpcGrpcClient,
     overrideLabels: UpstreamsConfig.Labels?
 ) : EthereumUpstream(
@@ -107,8 +109,9 @@ open class EthereumGrpcUpstream(
 
     private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
     var timeout = Defaults.timeout
+    private val ethereumSubscriptions = EthereumDshackleSubscriptions(chain, remote)
 
-    override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub {
+    override fun getBlockchainApi(): ReactorBlockchainStub {
         return remote
     }
 
@@ -141,6 +144,10 @@ open class EthereumGrpcUpstream(
         return upstreamStatus.getLabels()
     }
 
+    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+        return ethereumSubscriptions
+    }
+
     override fun getMethods(): CallMethods {
         return upstreamStatus.getCallMethods()
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt

@@ -32,6 +32,8 @@ import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleSubscriptions
 import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -107,6 +109,7 @@ open class EthereumPosGrpcUpstream(
 
     private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
     var timeout = Defaults.timeout
+    private val ethereumSubscriptions = EthereumDshackleSubscriptions(chain, remote)
 
     override fun start() {
     }
@@ -174,4 +177,8 @@ open class EthereumPosGrpcUpstream(
     override fun isGrpc(): Boolean {
         return true
     }
+
+    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+        return ethereumSubscriptions
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt

@@ -16,12 +16,12 @@
 package io.emeraldpay.dshackle.upstream.rpcclient
 
 import io.emeraldpay.dshackle.reader.Reader
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
+import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionImpl
 import io.emeraldpay.etherjar.rpc.RpcResponseError
 import reactor.core.publisher.Mono
 
 class JsonRpcWsClient(
-    private val ws: WsConnection
+    private val ws: WsConnectionImpl
 ) : Reader<JsonRpcRequest, JsonRpcResponse> {
 
     override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
@@ -36,6 +36,6 @@ class JsonRpcWsClient(
                 )
             )
         }
-        return ws.call(key)
+        return ws.callRpc(key)
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsMessage.kt

@@ -0,0 +1,22 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+class JsonRpcWsMessage(
+    val result: ByteArray?,
+    val error: JsonRpcError?,
+    val subscriptionId: String,
+)

src/test/groovy/io/emeraldpay/dshackle/commons/DurableFluxSpec.groovy

@@ -0,0 +1,135 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.commons
+
+import org.springframework.util.backoff.ExponentialBackOff
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Retry
+import spock.lang.Specification
+
+import java.time.Duration
+
+class DurableFluxSpec extends Specification {
+
+    private static CLOCK_ALLOW_ERROR_MS = 10
+
+    def "Normal subscribe works"() {
+        when:
+        def flux = DurableFlux.newBuilder()
+                .using({
+                    Flux.fromIterable([1, 2, 3])
+                })
+                .build()
+        def values = flux.connect().collectList().block(Duration.ofSeconds(1))
+
+        then:
+        values == [1, 2, 3]
+    }
+
+    def "Reconnects when broken"() {
+        when:
+        def connects = 0
+        def flux = DurableFlux.newBuilder()
+                .using({
+                    connects++
+                    def id = connects
+                    Flux.fromIterable([100+id, 200+id, 300+id])
+                            .concatWith(Mono.error(new RuntimeException("[TEST Reached the end of $id]")))
+                })
+                .backoffOnError(Duration.ofMillis(50))
+                .build()
+        def values = flux.connect()
+                .take(5)
+                .collectList().block(Duration.ofSeconds(1))
+
+        then:
+        values == [101, 201, 301, 102, 202]
+    }
+
+    @Retry // sometimes it goes too fast or too slow
+    def "No continuous backoff if restored"() {
+        when:
+        def connects = 0
+        def flux = DurableFlux.newBuilder()
+                .using({
+                    connects++
+                    def id = connects
+                    Flux.fromIterable([100+id, 200+id])
+                            .concatWith(Mono.error(new RuntimeException("[TEST ERROR $id]")))
+                })
+                .backoffOnError(new ExponentialBackOff(100, 2))
+                .build()
+        def values = flux.connect()
+                .take(7)
+
+        def verifier = StepVerifier.create(values)
+                .expectNext(101, 201)
+                .expectNoEvent(Duration.ofMillis(100 - CLOCK_ALLOW_ERROR_MS))
+                .expectNext(102, 202)
+                .expectNoEvent(Duration.ofMillis(100 - CLOCK_ALLOW_ERROR_MS))
+                .expectNext(103, 203)
+                .expectNoEvent(Duration.ofMillis(100 - CLOCK_ALLOW_ERROR_MS))
+                .expectNext(104)
+                .expectComplete()
+                .verifyLater()
+
+        then:
+        verifier.verify(Duration.ofSeconds(3))
+    }
+
+    @Retry // sometimes it goes too fast or too slow
+    def "Continue backoff if not restored immediately"() {
+        when:
+        def connects = 0
+        def flux = DurableFlux.newBuilder()
+                .using({
+                    connects++
+                    def id = connects
+                    if (id in [2, 4,5,6 ]) {
+                        Flux.error(new RuntimeException("[TEST ERROR $id]"))
+                    }  else {
+                        Flux.fromIterable([100+id, 200+id])
+                                .concatWith(Mono.error(new RuntimeException("[TEST ERROR $id]")))
+                    }
+                })
+                .backoffOnError(new ExponentialBackOff(100, 2))
+                .build()
+        def values = flux.connect()
+                .take(7)
+
+        def verifier = StepVerifier.create(values)
+                .expectNext(101, 201)
+                .as("first batch")
+                .expectNoEvent(Duration.ofMillis(100 + 200 - CLOCK_ALLOW_ERROR_MS))
+                .as("immediate fail on #2")
+                .expectNext(103, 203)
+                .as("second batch")
+                .expectNoEvent(Duration.ofMillis(100 + 200 + 400 + 800 - CLOCK_ALLOW_ERROR_MS))
+                .as("three fails in row (after the original) as #4, #5, #6")
+                .expectNext(107, 207)
+                .as("third batch")
+                .expectNoEvent(Duration.ofMillis(100 - CLOCK_ALLOW_ERROR_MS))
+                .expectNext(108)
+                .expectComplete()
+                .verifyLater()
+
+        then:
+        verifier.verify(Duration.ofSeconds(3))
+
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/commons/ExpiringSetSpec.groovy

@@ -0,0 +1,104 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.commons
+
+import io.emeraldpay.etherjar.domain.TransactionId
+import io.emeraldpay.etherjar.hex.HexDataComparator
+import io.emeraldpay.etherjar.tx.Transaction
+import spock.lang.Specification
+
+import java.time.Duration
+
+class ExpiringSetSpec extends Specification {
+
+    def "Add and check item"() {
+        setup:
+        def set = new ExpiringSet(Duration.ofSeconds(60), new HexDataComparator() as Comparator<TransactionId>, 100)
+
+        when:
+        def firstAdded = set.add(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c"))
+        def firstExists = set.contains(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c"))
+        def secondAdded = set.add(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d"))
+        def secondExists = set.contains(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d"))
+        def thirdExists = set.contains(TransactionId.from("0xe218f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0"))
+
+        then:
+        firstAdded
+        secondAdded
+        firstExists
+        secondExists
+        !thirdExists
+
+        set.size == 2
+    }
+
+    def "Doesn't grow after limit"() {
+        setup:
+        def set = new ExpiringSet(Duration.ofSeconds(60), new HexDataComparator() as Comparator<TransactionId>, 3)
+
+        when:
+        set.add(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c"))
+        set.add(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d"))
+        set.add(TransactionId.from("0xe218f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0"))
+        set.add(TransactionId.from("0xd3082daa344a64369c8aace137f22beb5085351bf111859202bac66b70b28bdd"))
+
+        then:
+        set.size == 3
+
+        when:
+        def firstExists = set.contains(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c"))
+        def secondExists = set.contains(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d"))
+        def thirdExists = set.contains(TransactionId.from("0xe218f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0"))
+
+        then:
+        !firstExists
+        secondExists
+        thirdExists
+    }
+
+    def "Remove expired"() {
+        setup:
+        def set = new ExpiringSet(Duration.ofMillis(100), new HexDataComparator() as Comparator<TransactionId>, 100)
+
+        when:
+        set.add(TransactionId.from("0x1118b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c"))
+        set.add(TransactionId.from("0x222e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d"))
+
+        then:
+        set.size == 2
+
+        when:
+        Thread.sleep(60)
+        set.add(TransactionId.from("0x3338f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0"))
+        set.add(TransactionId.from("0x44482daa344a64369c8aace137f22beb5085351bf111859202bac66b70b28bdd"))
+
+        then:
+        set.size == 4
+
+        when:
+        Thread.sleep(60)
+        set.add(TransactionId.from("0x55531d466acf4ef72f7e0fbc60a5c2c9d2845b90a3d27b6d7581575cb119cac9"))
+        set.add(TransactionId.from("0x6665e1f32cb21aee5f27d804cfc65781d5c140b002776bc073f9405479e8b1e5"))
+
+        then:
+        set.size == 4
+        set.contains(TransactionId.from("0x3338f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0"))
+        set.contains(TransactionId.from("0x44482daa344a64369c8aace137f22beb5085351bf111859202bac66b70b28bdd"))
+        set.contains(TransactionId.from("0x55531d466acf4ef72f7e0fbc60a5c2c9d2845b90a3d27b6d7581575cb119cac9"))
+        set.contains(TransactionId.from("0x6665e1f32cb21aee5f27d804cfc65781d5c140b002776bc073f9405479e8b1e5"))
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/commons/SharedFluxHolderSpec.groovy

@@ -0,0 +1,79 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.commons
+
+import reactor.core.publisher.Flux
+import spock.lang.Specification
+
+import java.time.Duration
+
+class SharedFluxHolderSpec extends Specification {
+
+    def "Keeps one flux"() {
+        when:
+        def called = 0
+        def shared = new SharedFluxHolder<Integer>({
+            Flux.fromIterable([100+called, 200+called, 300+called, 400+called, 500+called])
+                    .delayElements(Duration.ofMillis(100))
+                    .doOnSubscribe {called++ }
+        })
+        List<Integer> values1 = []
+        List<Integer> values2 = []
+        new Thread({
+            values1 = shared.get().collectList().block(Duration.ofSeconds(1))
+        }).start()
+        new Thread({
+            values2 = shared.get().collectList().block(Duration.ofSeconds(1))
+        }).start()
+        Thread.sleep(1000)
+
+        then:
+        values1 == [100, 200, 300, 400, 500]
+        values2 == [100, 200, 300, 400, 500]
+        called == 1
+    }
+
+    def "Create a new flux if existing completes"() {
+        when:
+        def called = 0
+        def shared = new SharedFluxHolder<Integer>({
+            Flux.fromIterable([100+called, 200+called, 300+called, 400+called, 500+called])
+                    .delayElements(Duration.ofMillis(100))
+                    .doOnSubscribe {called++ }
+        })
+        List<Integer> values1 = []
+        List<Integer> values2 = []
+        List<Integer> values3 = []
+        new Thread({
+            values1 = shared.get().collectList().block(Duration.ofSeconds(1))
+        }).start()
+        new Thread({
+            values2 = shared.get().collectList().block(Duration.ofSeconds(1))
+        }).start()
+        Thread.sleep(1000)
+
+        new Thread({
+            values3 = shared.get().collectList().block(Duration.ofSeconds(1))
+        }).start()
+        Thread.sleep(1000)
+
+        then:
+        values1 == [100, 200, 300, 400, 500]
+        values2 == [100, 200, 300, 400, 500]
+        values3 == [101, 201, 301, 401, 501]
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy

@@ -20,7 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.test.MultistreamHolderMock
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
 import io.emeraldpay.dshackle.upstream.signature.NoSigner
 import io.emeraldpay.dshackle.Chain
 import reactor.core.publisher.Flux
@@ -39,12 +39,12 @@ class NativeSubscribeSpec extends Specification {
                 .setMethod("newHeads")
                 .build()
 
-        def subscribe = Mock(EthereumSubscribe) {
+        def subscribe = Mock(EthereumSubscriptionApi) {
             1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
         }
         def up = Mock(EthereumPosMultiStream) {
             1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null
-            1 * it.getSubscribe() >> subscribe
+            1 * it.getSubscriptionApi() >> subscribe
         }
 
         def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)
@@ -70,7 +70,7 @@ class NativeSubscribeSpec extends Specification {
                 ))
                 .build()
 
-        def subscribe = Mock(EthereumSubscribe) {
+        def subscribe = Mock(EthereumSubscriptionApi) {
             1 * it.subscribe("logs", { params ->
                 println("params: $params")
                 def ok = params instanceof Map &&
@@ -83,7 +83,7 @@ class NativeSubscribeSpec extends Specification {
         }
         def up = Mock(EthereumPosMultiStream) {
             1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null
-            1 * it.getSubscribe() >> subscribe
+            1 * it.getSubscriptionApi() >> subscribe
         }
 
         def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)
@@ -106,7 +106,7 @@ class NativeSubscribeSpec extends Specification {
                 .build()
         def up = Mock(EthereumPosMultiStream) {
             1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> Flux.just("{}")
-            0 * it.getSubscribe()
+            0 * it.getSubscriptionApi()
         }
 
         def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy

@@ -5,10 +5,10 @@ import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.config.TokensConfig
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
 import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.etherjar.domain.BlockHash
@@ -169,21 +169,26 @@ class TrackERC20AddressSpec extends Specification {
                         "unknown"
                 )
         ]
-        def logs = Mock(ConnectLogs) {
+
-            1 * start(
+        def connect = Mock(SubscriptionConnect) {
-                    [Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")],
+            1 * connect(Selector.empty) >> {
-                    [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
-                    Selector.empty
-            ) >> { args ->
-                println("ConnectLogs.start $args")
                 Flux.fromIterable(events)
             }
         }
-        def sub = Mock(EthereumSubscribe) {
+        def logs = Mock(ConnectLogs) {
+            1 * create(
+                    [Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")],
+                    [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
+            ) >> { args ->
+                println("ConnectLogs.start $args")
+                connect
+            }
+        }
+        def sub = Mock(EthereumSubscriptionApi) {
             1 * getLogs() >> logs
         }
         def up = Mock(EthereumPosMultiStream) {
-            1 * getSubscribe() >> sub
+            1 * getSubscriptionApi() >> sub
             _ * cast(EthereumPosMultiStream) >> { args ->
                 it
             }

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

@@ -2,6 +2,8 @@ package io.emeraldpay.dshackle.test
 
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumUpstreamSubscriptions
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -34,4 +36,9 @@ class EthereumConnectorMock implements EthereumConnector {
     boolean isRunning() {
         return true
     }
+
+    @Override
+    EthereumUpstreamSubscriptions getUpstreamSubscriptions() {
+        return NoEthereumUpstreamSubscriptions.DEFAULT
+    }
 }

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

@@ -48,7 +48,7 @@ class FilteredApisSpec extends Specification {
             def httpFactory = Mock(HttpFactory) {
                 create(_, _) >> TestingCommons.api().tap { it.id = "${i++}" }
             }
-            def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
+            def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID)
             new EthereumRpcUpstream(
                     "test",
                     (byte)123,

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy

@@ -32,7 +32,7 @@ import java.time.Instant
 
 class DefaultEthereumHeadSpec extends Specification {
 
-    DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
+    DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID)
     ObjectMapper objectMapper = Global.objectMapper
 
     def blocks = (10L..20L).collect { i ->

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscriptionApiSpec.groovy

@@ -16,15 +16,16 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.hex.Hex32
 import spock.lang.Specification
 
-class EthereumSubscribeSpec extends Specification {
+class EthereumSubscriptionApiSpec extends Specification {
 
     def "read empty logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([:])
 
@@ -35,7 +36,7 @@ class EthereumSubscribeSpec extends Specification {
 
     def "read single address logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "0x829bd824b016326a401d083b33d092293333a830"
@@ -60,7 +61,7 @@ class EthereumSubscribeSpec extends Specification {
 
     def "ignores invalid address for logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "829bd824b016326a401d083b33d092293333a830"
@@ -73,7 +74,7 @@ class EthereumSubscribeSpec extends Specification {
 
     def "read multi address logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: ["0x829bd824b016326a401d083b33d092293333a830", "0x401d083b33d092293333a83829bd824b016326a0"]
@@ -89,7 +90,7 @@ class EthereumSubscribeSpec extends Specification {
 
     def "read single topic logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
@@ -114,7 +115,7 @@ class EthereumSubscribeSpec extends Specification {
 
     def "read invalid topic for request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: [
@@ -132,7 +133,7 @@ class EthereumSubscribeSpec extends Specification {
 
     def "read multi topic logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: [
@@ -151,7 +152,7 @@ class EthereumSubscribeSpec extends Specification {
 
     def "read full logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumPosMultiStream)
+        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy

@@ -0,0 +1,74 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.BlockValidator
+import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice
+import io.emeraldpay.etherjar.domain.BlockHash
+import io.emeraldpay.etherjar.domain.TransactionId
+import io.emeraldpay.etherjar.rpc.json.BlockJson
+import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
+import reactor.core.publisher.Flux
+import spock.lang.Specification
+
+import java.time.Instant
+import java.time.temporal.ChronoUnit
+
+class EthereumWsHeadSpec extends Specification {
+
+    def "Fetch block"() {
+        setup:
+        def block = new BlockJson<TransactionRefJson>()
+        block.number = 100
+        block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
+        block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
+        block.transactions = [
+                new TransactionRefJson(TransactionId.from("0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8")),
+                new TransactionRefJson(TransactionId.from("0xebe8f22a55a9e26892a8545b93cbb2bfa4fd81c3184e50e5cf6276025bb42b93"))
+        ]
+        block.uncles = []
+        block.totalDifficulty = BigInteger.ONE
+
+        def headBlock = block.copy().tap {
+            it.transactions = null
+        }.with {
+            Global.objectMapper.writeValueAsBytes(it)
+        }
+
+        def apiMock = TestingCommons.api()
+        apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block)
+
+        def ws = Mock(WsSubscriptions)
+
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws)
+
+        when:
+        def act = head.listenNewHeads().blockFirst()
+
+        then:
+        act == BlockContainer.from(block)
+        act.transactions.size() == 2
+        act.transactions[0].toHexWithPrefix() == "0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8"
+        act.transactions[1].toHexWithPrefix() == "0xebe8f22a55a9e26892a8545b93cbb2bfa4fd81c3184e50e5cf6276025bb42b93"
+
+        1 * ws.subscribe("newHeads") >> Flux.fromIterable([
+                headBlock
+        ])
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy

@@ -11,7 +11,7 @@ import spock.lang.Specification
 
 import java.time.Duration
 
-class WsConnectionRealSpec extends Specification {
+class WsConnectionImplRealSpec extends Specification {
 
     static SLEEP = 500
 
@@ -19,7 +19,7 @@ class WsConnectionRealSpec extends Specification {
     @Shared
     MockWSServer server
     @Shared
-    WsConnection conn
+    WsConnectionImpl conn
 
     def setup() {
         if (System.getenv("CI") == "true") {
@@ -31,7 +31,7 @@ class WsConnectionRealSpec extends Specification {
         server = new MockWSServer(port)
         server.start()
         Thread.sleep(SLEEP)
-        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null)
+        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null)
     }
 
     def cleanup() {
@@ -39,22 +39,10 @@ class WsConnectionRealSpec extends Specification {
         server.stop()
     }
 
-    def "Connects to server"() {
+    def "Can make a RPC request"() {
         when:
         conn.connect()
-        Thread.sleep(SLEEP)
+        def resp = conn.callRpc(new JsonRpcRequest("foo_bar", []))
-        println("verify....")
-        def act = server.received
-        then:
-        act.size() > 0
-        act[0].value.contains("\"method\":\"eth_subscribe\"")
-        act[0].value.contains("\"params\":[\"newHeads\"]")
-    }
-
-    def "Makes RPC request"() {
-        when:
-        conn.connect()
-        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
         then:
         StepVerifier.create(resp)
                 .then {
@@ -70,8 +58,8 @@ class WsConnectionRealSpec extends Specification {
         Thread.sleep(SLEEP)
         def act = server.received
         then:
-        act.size() == 2
+        act.size() == 1
-        act[1].value.contains("\"method\":\"foo_bar\"")
+        act[0].value.contains("\"method\":\"foo_bar\"")
     }
 
     def "Reconnects after server disconnect"() {
@@ -83,15 +71,15 @@ class WsConnectionRealSpec extends Specification {
         Thread.sleep(SLEEP)
         server = new MockWSServer(port)
         server.start()
-        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+        server.onNextReply('{"jsonrpc":"2.0","id":100,"result":1}')
         // reconnects in 2 seconds, give 1 extra
         Thread.sleep(3_000)
+        def resp = conn.callRpc(new JsonRpcRequest("foo_bar", [])).block(Duration.ofSeconds(1))
         def act = server.received
 
         then:
-        act.size() > 0
+        act.size() == 1
-        act[0].value.contains("\"method\":\"eth_subscribe\"")
+        act[0].value.contains("\"method\":\"foo_bar\"")
-        act[0].value.contains("\"params\":[\"newHeads\"]")
     }
 
     def "Error on request when server disconnects"() {
@@ -99,7 +87,7 @@ class WsConnectionRealSpec extends Specification {
         conn.connect()
         conn.reconnectIntervalSeconds = 2
 
-        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+        def resp = conn.callRpc(new JsonRpcRequest("foo_bar", []))
 
         then:
         StepVerifier.create(resp)
@@ -113,7 +101,7 @@ class WsConnectionRealSpec extends Specification {
         def up = Mock(DefaultUpstream) {
             _ * getId() >> "test"
         }
-        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null)
+        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up)
         when:
         conn.connect()
         conn.reconnectIntervalSeconds = 10
@@ -125,18 +113,6 @@ class WsConnectionRealSpec extends Specification {
         1 * up.setStatus(UpstreamAvailability.UNAVAILABLE)
     }
 
-    def "Validates after connect"() {
-        setup:
-        def validator = Mock(EthereumUpstreamValidator)
-        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator)
-        when:
-        conn.connect()
-        Thread.sleep(100)
-
-        then:
-        1 * validator.validate()
-    }
-
     def "Try to connects to server until it's available"() {
         when:
         server.stop()
@@ -146,12 +122,14 @@ class WsConnectionRealSpec extends Specification {
         Thread.sleep(3_000)
         server = new MockWSServer(port)
         server.start()
-        Thread.sleep(2_000)
+        server.onNextReply('{"jsonrpc":"2.0","id":100,"result":1}')
+        Thread.sleep(3_000)
+
+        def resp = conn.callRpc(new JsonRpcRequest("foo_bar", [])).block(Duration.ofSeconds(1))
         def act = server.received
         then:
-        act.size() > 0
+        act.size() == 1
-        act[0].value.contains("\"method\":\"eth_subscribe\"")
+        act[0].value.contains("\"method\":\"foo_bar\"")
-        act[0].value.contains("\"params\":[\"newHeads\"]")
     }
 
     def "Call after reconnect"() {
@@ -166,7 +144,7 @@ class WsConnectionRealSpec extends Specification {
         // reconnects in 2 seconds, give 1 extra
         Thread.sleep(3_000)
 
-        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+        def resp = conn.callRpc(new JsonRpcRequest("foo_bar", []))
         then:
         StepVerifier.create(resp)
                 .then {

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy

@@ -34,48 +34,14 @@ import java.time.Duration
 import java.time.Instant
 import java.time.temporal.ChronoUnit
 
-class WsConnectionSpec extends Specification {
+class WsConnectionImplSpec extends Specification {
-
-    def "Fetch block"() {
-        setup:
-        def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
-
-        def block = new BlockJson<TransactionRefJson>()
-        block.number = 100
-        block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
-        block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
-        block.transactions = []
-        block.uncles = []
-        block.totalDifficulty = BigInteger.ONE
-
-        def headBlock = block.copy().tap {
-            it.transactions = null
-        }
-
-        def apiMock = TestingCommons.api()
-        def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null)
-
-        apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block)
-
-        when:
-        Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
-        def act = Flux.from(ws.getBlocksFlux())
-
-        then:
-        StepVerifier.create(act)
-                .then { ws.onNewHeads(headBlock).subscribe() }
-                .expectNext(BlockContainer.from(block))
-                .thenCancel()
-                .verify(Duration.ofSeconds(5))
-    }
 
     def "Makes a RPC call"() {
         setup:
         def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null)
+        def ws = wsf.create(null)
 
         def tx = new TransactionJson().tap {
             hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
@@ -84,7 +50,7 @@ class WsConnectionSpec extends Specification {
 
         when:
         Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
-        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
+        def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
 
         then:
         StepVerifier.create(act)
@@ -100,13 +66,13 @@ class WsConnectionSpec extends Specification {
         def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null)
+        def ws = wsf.create(null)
 
         apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null)
 
         when:
         Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
-        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
+        def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
 
         then:
         StepVerifier.create(act)
@@ -123,14 +89,14 @@ class WsConnectionSpec extends Specification {
         def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null)
+        def ws = wsf.create(null)
 
         apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"],
                 new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test"))
 
         when:
         Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
-        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
+        def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
 
         then:
         StepVerifier.create(act)

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy

@@ -0,0 +1,90 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.Sinks
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+
+class WsSubscriptionsImplSpec extends Specification {
+
+    def "Makes a subscription"() {
+        setup:
+        def answers = Flux.fromIterable(
+                [
+                    new JsonRpcWsMessage("100".bytes, null, "0xcff45d00e7"),
+                    new JsonRpcWsMessage("101".bytes, null, "0xcff45d00e7"),
+                    new JsonRpcWsMessage("102".bytes, null, "0xcff45d00e7"),
+                ]
+        )
+
+        def conn = Mock(WsConnectionImpl)
+        def ws = new WsSubscriptionsImpl(conn)
+
+        when:
+        def act = ws.subscribe("foo_bar")
+            .map { new String(it) }
+            .take(3)
+            .collectList().block(Duration.ofSeconds(1))
+
+        then:
+        act == ["100", "101", "102"]
+
+        1 * conn.callRpc({ JsonRpcRequest req ->
+            req.method == "eth_subscribe" && req.params == ["foo_bar"]
+        }) >> Mono.just(new JsonRpcResponse('"0xcff45d00e7"'.bytes, null))
+        1 * conn.getSubscribeResponses() >> answers
+    }
+
+    def "Produces only messages to the actual subscription"() {
+        setup:
+        def answers = Flux.fromIterable(
+                [
+                        new JsonRpcWsMessage("100".bytes, null, "0xcff45d00e7"),
+                        new JsonRpcWsMessage("AAA".bytes, null, "0x000001a0e7"),
+                        new JsonRpcWsMessage("101".bytes, null, "0xcff45d00e7"),
+                        new JsonRpcWsMessage("BBB".bytes, null, "0x000001a0e7"),
+                        new JsonRpcWsMessage("CCC".bytes, null, "0x000001a0e7"),
+                        new JsonRpcWsMessage("102".bytes, null, "0xcff45d00e7"),
+                        new JsonRpcWsMessage("DDD".bytes, null, "0x000001a0e7"),
+                ]
+        )
+
+        def conn = Mock(WsConnectionImpl)
+        def ws = new WsSubscriptionsImpl(conn)
+
+        when:
+        def act = ws.subscribe("foo_bar")
+                .map { new String(it) }
+                .take(3)
+                .collectList().block(Duration.ofSeconds(1))
+
+        then:
+        act == ["100", "101", "102"]
+
+        1 * conn.callRpc({ JsonRpcRequest req ->
+            req.method == "eth_subscribe" && req.params == ["foo_bar"]
+        }) >> Mono.just(new JsonRpcResponse('"0xcff45d00e7"'.bytes, null))
+        1 * conn.getSubscribeResponses() >> answers
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/AggregatedPendingTxesSpec.groovy

@@ -0,0 +1,92 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.etherjar.domain.TransactionId
+import reactor.core.publisher.Flux
+import spock.lang.Specification
+
+import java.time.Duration
+
+class AggregatedPendingTxesSpec extends Specification {
+
+    def "Produces values from two sources"() {
+        setup:
+        def source1 = Mock(PendingTxesSource)
+        def source2 = Mock(PendingTxesSource)
+
+        when:
+        def aggregate = new AggregatedPendingTxes([source1, source2])
+        def values = aggregate.connect(Selector.empty)
+                .collectList().block(Duration.ofSeconds(1))
+
+        then:
+        1 * source1.connect(Selector.empty) >> Flux.fromIterable(
+                [
+                        "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",
+                        "0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e"
+                ].collect { TransactionId.from(it) }
+        )
+        1 * source2.connect(Selector.empty) >> Flux.fromIterable(
+                [
+                        "0x9a9d4618b12d36d17a63d48c5b5efc05b461feead124ddd86803d8bca4015248",
+                        "0xa38173981f8eab96ee70cefe42735af0f574b7ef354565f2fea32a28e5ed9bd2"
+                ].collect { TransactionId.from(it) }
+        )
+
+        values.collect { it.toHex() }.toSorted() == [
+                "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",
+                "0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e",
+                "0x9a9d4618b12d36d17a63d48c5b5efc05b461feead124ddd86803d8bca4015248",
+                "0xa38173981f8eab96ee70cefe42735af0f574b7ef354565f2fea32a28e5ed9bd2"
+        ].toSorted()
+    }
+
+    def "Skip duplicates"() {
+        setup:
+        def source1 = Mock(PendingTxesSource)
+        def source2 = Mock(PendingTxesSource)
+
+        when:
+        def aggregate = new AggregatedPendingTxes([source1, source2])
+        def values = aggregate.connect(Selector.empty)
+                .collectList().block(Duration.ofSeconds(1))
+
+        then:
+        1 * source1.connect(Selector.empty) >> Flux.fromIterable(
+                [
+                        "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",
+                        "0x9a9d4618b12d36d17a63d48c5b5efc05b461feead124ddd86803d8bca4015248",
+                        "0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e"
+                ].collect { TransactionId.from(it) }
+        )
+        1 * source2.connect(Selector.empty) >> Flux.fromIterable(
+                [
+                        "0x9a9d4618b12d36d17a63d48c5b5efc05b461feead124ddd86803d8bca4015248",
+                        "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",
+                        "0xa38173981f8eab96ee70cefe42735af0f574b7ef354565f2fea32a28e5ed9bd2"
+                ].collect { TransactionId.from(it) }
+        )
+
+        values.collect { it.toHex() }.toSorted() == [
+                "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",
+                "0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e",
+                "0x9a9d4618b12d36d17a63d48c5b5efc05b461feead124ddd86803d8bca4015248",
+                "0xa38173981f8eab96ee70cefe42735af0f574b7ef354565f2fea32a28e5ed9bd2"
+        ].toSorted()
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/DshacklePendingTxesSourceSpec.groovy

@@ -0,0 +1,125 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import com.google.protobuf.ByteString
+import io.emeraldpay.api.proto.BlockchainGrpc
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.upstream.Selector
+import io.grpc.Channel
+import io.grpc.ManagedChannel
+import io.grpc.Server
+import io.grpc.inprocess.InProcessChannelBuilder
+import io.grpc.inprocess.InProcessServerBuilder
+import io.grpc.stub.StreamObserver
+import spock.lang.Specification
+
+import java.time.Duration
+
+class DshacklePendingTxesSourceSpec extends Specification {
+
+    def "Produces values"() {
+        setup:
+        BlockchainOuterClass.NativeSubscribeRequest receivedRequest
+
+        String uniqueName = InProcessServerBuilder.generateName();
+        Server server = InProcessServerBuilder.forName(uniqueName)
+                .directExecutor()
+                .addService(new BlockchainGrpc.BlockchainImplBase() {
+                    @Override
+                    void nativeSubscribe(BlockchainOuterClass.NativeSubscribeRequest request, StreamObserver<BlockchainOuterClass.NativeSubscribeReplyItem> responseObserver) {
+                        receivedRequest = request
+                        [
+                                '"0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c"',
+                                '"0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e"',
+                                '"0x9a9d4618b12d36d17a63d48c5b5efc05b461feead124ddd86803d8bca4015248"',
+                                '"0xa38173981f8eab96ee70cefe42735af0f574b7ef354565f2fea32a28e5ed9bd2"'
+                        ]
+                                    .collect { it.bytes }
+                                    .collect {
+                                        BlockchainOuterClass.NativeSubscribeReplyItem.newBuilder()
+                                            .setPayload(ByteString.copyFrom(it))
+                                            .build()
+                                    }.forEach {
+                                        responseObserver.onNext(it)
+                                    }
+                        responseObserver.onCompleted()
+                    }
+                })
+                .build().start()
+        ManagedChannel channel = InProcessChannelBuilder.forName(uniqueName)
+                .directExecutor()
+                .build()
+
+        def remote = ReactorBlockchainGrpc.newReactorStub(channel)
+        def pending = new DshacklePendingTxesSource(Chain.ETHEREUM, remote)
+
+        when:
+        pending.available = true
+        def txes = pending.connect(Selector.empty).take(3)
+                .collectList().block(Duration.ofSeconds(1))
+
+        then:
+        receivedRequest != null
+        receivedRequest.chainValue == Chain.ETHEREUM.id
+        receivedRequest.method == "newPendingTransactions"
+        txes.collect {it.toHex() } == [
+                "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",
+                "0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e",
+                "0x9a9d4618b12d36d17a63d48c5b5efc05b461feead124ddd86803d8bca4015248",
+        ]
+    }
+
+    def "available when method is enabled on remote"() {
+        setup:
+        def pending = new DshacklePendingTxesSource(Chain.ETHEREUM, ReactorBlockchainGrpc.newReactorStub(Stub(Channel)))
+        pending.available = false
+        when:
+        pending.update(
+                BlockchainOuterClass.DescribeChain.newBuilder()
+                .addAllSupportedMethods(["newPendingTransactions"])
+                .build()
+        )
+        then:
+        pending.available
+    }
+
+    def "unavailable when not method is enabled on remote"() {
+        setup:
+        def pending = new DshacklePendingTxesSource(Chain.ETHEREUM, ReactorBlockchainGrpc.newReactorStub(Stub(Channel)))
+        pending.available = false
+        when:
+        pending.update(
+                BlockchainOuterClass.DescribeChain.newBuilder()
+                        .addAllSupportedMethods(["other_method"])
+                        .build()
+        )
+        then:
+        !pending.available
+
+        when: "It was enabled before getting an update"
+        pending.available = true
+        pending.update(
+                BlockchainOuterClass.DescribeChain.newBuilder()
+                        .addAllSupportedMethods(["other_method"])
+                        .build()
+        )
+        then:
+        !pending.available
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxesSpec.groovy

@@ -0,0 +1,50 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
+import reactor.core.publisher.Flux
+import spock.lang.Specification
+
+import java.time.Duration
+
+class WebsocketPendingTxesSpec extends Specification {
+
+    def "Produces values"() {
+        setup:
+        def responses = [
+                '"0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c"',
+                '"0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e"',
+                '"0x67f22a3b441ea312306f97694ca8159f8d6faaccf0f5ce6442c84b13991f1d23"',
+                '"0xa38173981f8eab96ee70cefe42735af0f574b7ef354565f2fea32a28e5ed9bd2"',
+        ].collect { it.bytes }
+        def ws = Mock(WsSubscriptions)
+        def pending = new WebsocketPendingTxes(ws)
+
+        when:
+        def txes = pending.connect(Selector.empty).take(3)
+                .collectList().block(Duration.ofSeconds(1))
+
+        then:
+        1 * ws.subscribe("newPendingTransactions") >> Flux.fromIterable(responses)
+        txes.collect {it.toHex() } == [
+                "0xa61bab14fc9720ea8725622688c2f964666d7c2afdae38af7dad53f12f242d5c",
+                "0x911548eb0f3bf353a54e03a3506c7c3e747470d6c201f03babbc07ff6e14cd6e",
+                "0x67f22a3b441ea312306f97694ca8159f8d6faaccf0f5ce6442c84b13991f1d23",
+        ]
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy

@@ -1,6 +1,6 @@
 package io.emeraldpay.dshackle.upstream.rpcclient
 
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
+import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionImpl
 import reactor.core.Exceptions
 import spock.lang.Specification
 
@@ -10,7 +10,7 @@ class JsonRpcWsClientSpec extends Specification {
 
     def "Produce error if WS is not connected"() {
         setup:
-        def ws = Mock(WsConnection)
+        def ws = Mock(WsConnectionImpl)
         def client = new JsonRpcWsClient(ws)
         when:
         client.read(new JsonRpcRequest("foo_bar", [], 1))