StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "Native subscribe filter"

Browse Source
This commit is contained in:
Vyacheslav Shebanov
2022-09-22 10:56:49 +03:00
committed by GitHub
parent 63d96904b3
commit 2cf8451e4a
19 changed files with 72 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt
View File

@@ -17,7 +17,6 @@ package io.emeraldpay.dshackle.proxy
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.BlockchainOuterClass.Selector
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.ProxyConfig
import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp
@@ -140,7 +139,7 @@ class WebsocketHandler(
}
// produce actual responses
val responses = nativeSubscribe
.subscribe(blockchain, methodParams.first, methodParams.second, io.emeraldpay.dshackle.upstream.Selector.empty)
.subscribe(blockchain, methodParams.first, methodParams.second)
.map { event ->
WsSubscriptionResponse(params = WsSubscriptionData(event, subscriptionId))
}

22
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
View File

@@ -20,7 +20,6 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
@@ -51,17 +50,20 @@ open class NativeSubscribe(
.onErrorMap(this@NativeSubscribe::convertToStatus)
}
fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher<out Any> {
val chain = Chain.byId(request.chainValue)
fun start(it: BlockchainOuterClass.NativeSubscribeRequest): Publisher<out Any> {
val chain = Chain.byId(it.chainValue)
if (BlockchainType.from(chain) != BlockchainType.ETHEREUM_POS && BlockchainType.from(chain) != BlockchainType.ETHEREUM) {
return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}"))
}
val method = request.method
val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
objectMapper.readValue(it.newInput(), Map::class.java)
val method = it.method
val params: Any? = it.payload?.let { payload ->
if (payload.size() > 0) {
objectMapper.readValue(payload.newInput(), Map::class.java)
} else {
null
}
}
val matcher = Selector.convertToMatcher(request.selector)
return subscribe(chain, method, params, matcher)
return subscribe(chain, method, params)
}
fun convertToStatus(t: Throwable) = when (t) {
@@ -79,11 +81,11 @@ open class NativeSubscribe(
}
}
open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
open fun subscribe(chain: Chain, method: String, params: Any?): Flux<out Any> {
val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
return (up as EthereumLikeMultistream)
.getSubscribe()
.subscribe(method, params, matcher)
.subscribe(method, params)
}
fun convertToProto(value: Any): BlockchainOuterClass.NativeSubscribeReplyItem {

4
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt
View File

@@ -20,7 +20,6 @@ import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.config.TokensConfig
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.etherjar.domain.Address
@@ -91,8 +90,7 @@ class TrackERC20Address(
.getSubscribe().logs
.start(
listOf(tokenDefinition.token.contract),
listOf(EventId.fromSignature("Transfer", "address", "address", "uint256")),
Selector.empty
listOf(EventId.fromSignature("Transfer", "address", "address", "uint256"))
)
return ethereumAddresses.extract(request.address)

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

@@ -1,12 +1,8 @@
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
interface EthereumLikeMultistream : Upstream {
fun getReader(): EthereumReader
fun getSubscribe(): EthereumSubscribe
fun getHead(mather: Selector.Matcher): Head
}

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

@@ -20,7 +20,6 @@ import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ChainFees
import io.emeraldpay.dshackle.upstream.EmptyHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.Multistream
@@ -32,7 +31,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import org.springframework.util.ConcurrentReferenceHashMap
import reactor.core.publisher.Mono
@Suppress("UNCHECKED_CAST")
@@ -47,8 +45,6 @@ open class EthereumMultistream(
}
private var head: Head? = null
private val filteredHeads: MutableMap<String, Head> =
ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
private val subscribe = EthereumSubscribe(this)
@@ -78,7 +74,6 @@ open class EthereumMultistream(
override fun stop() {
super.stop()
reader.stop()
filteredHeads.clear()
}
override fun isRunning(): Boolean {
@@ -123,7 +118,6 @@ open class EthereumMultistream(
lagObserver.start()
newHead
}
filteredHeads[Selector.AnyLabelMatcher().describeInternal()] = head
onHeadUpdated(head)
return head
}
@@ -148,24 +142,6 @@ open class EthereumMultistream(
return subscribe
}
override fun getHead(mather: Selector.Matcher): Head =
filteredHeads.computeIfAbsent(mather.describeInternal().intern()) { _ ->
upstreams.filter { mather.matches(it) }
.apply {
log.debug("Found $size upstreams matching [${mather.describeInternal()}]")
}
.map { it.getHead() }
.let {
when (it.size) {
0 -> EmptyHead()
1 -> it.first()
else -> MergedHead(it, MostWorkForkChoice()).apply {
start()
}
}
}
}
override fun getFeeEstimation(): ChainFees {
return feeEstimation
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt
View File

@@ -1,6 +1,5 @@
package io.emeraldpay.dshackle.upstream.ethereum
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
@@ -22,9 +21,9 @@ open class EthereumSubscribe(
private val syncing = ConnectSyncing(upstream)
@Suppress("UNCHECKED_CAST")
open fun subscribe(method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
open fun subscribe(method: String, params: Any?): Flux<out Any> {
if (method == "newHeads") {
return newHeads.connect(matcher)
return newHeads.connect()
}
if (method == "logs") {
val paramsMap = try {
@@ -36,7 +35,7 @@ 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.start(paramsMap.address, paramsMap.topics)
}
if (method == "syncing") {
return syncing.connect()

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

@@ -19,16 +19,16 @@ import io.emeraldpay.dshackle.data.BlockContainer
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.ethereum.EthereumLikeMultistream
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.scheduler.Schedulers
import java.time.Duration
import java.util.LinkedList
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.locks.ReentrantLock
import java.util.concurrent.locks.ReentrantReadWriteLock
import kotlin.concurrent.read
import kotlin.concurrent.withLock
import kotlin.concurrent.write
class ConnectBlockUpdates(
@@ -46,19 +46,30 @@ class ConnectBlockUpdates(
*/
private val history = LinkedList<BlockContainer>()
private val historyUpdateLock = ReentrantReadWriteLock()
private val connected: MutableMap<String, Flux<Update>> = ConcurrentHashMap()
fun connect() = connect(Selector.empty)
fun connect(matcher: Selector.Matcher): Flux<Update> {
return connected.computeIfAbsent(matcher.describeInternal()) { key ->
extract(upstream.getHead(matcher))
private var connected: Flux<Update>? = null
private val connectLock = ReentrantLock()
fun connect(): Flux<Update> {
val current = connected
if (current != null) {
return current
}
connectLock.withLock {
val currentRecheck = connected
if (currentRecheck != null) {
return currentRecheck
}
val created = extract(upstream.getHead())
.publishOn(Schedulers.boundedElastic())
.publish()
.refCount(1, Duration.ofSeconds(60))
.doFinally {
// forget it on disconnect, so next time it's recreated
connected.remove(key)
connected = null
}
connected = created
return created
}
}

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

@@ -15,7 +15,6 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum.subscribe
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
import io.emeraldpay.etherjar.domain.Address
@@ -41,17 +40,17 @@ open class ConnectLogs(
private val produceLogs = ProduceLogs(upstream)
fun start(matcher: Selector.Matcher): Flux<LogMessage> {
return produceLogs.produce(connectBlockUpdates.connect(matcher))
fun start(): Flux<LogMessage> {
return produceLogs.produce(connectBlockUpdates.connect())
}
open fun start(addresses: List<Address>, topics: List<Hex32>, matcher: Selector.Matcher): Flux<LogMessage> {
open fun start(addresses: List<Address>, topics: List<Hex32>): Flux<LogMessage> {
// shortcut to the whole output if we don't have any filters
if (addresses.isEmpty() && topics.isEmpty()) {
return start(matcher)
return start()
}
// filtered output
return start(matcher)
return start()
.transform(filtered(addresses, topics))
}

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

@@ -15,14 +15,14 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum.subscribe
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.scheduler.Schedulers
import java.time.Duration
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
/**
* Connects/reconnects to the upstream to produce NewHeads messages
@@ -35,18 +35,30 @@ class ConnectNewHeads(
private val log = LoggerFactory.getLogger(ConnectNewHeads::class.java)
}
private val connected: MutableMap<String, Flux<NewHeadMessage>> = ConcurrentHashMap()
private var connected: Flux<NewHeadMessage>? = null
private val connectLock = ReentrantLock()
fun connect(matcher: Selector.Matcher): Flux<NewHeadMessage> =
connected.computeIfAbsent(matcher.describeInternal()) { key ->
ProduceNewHeads(upstream.getHead(matcher))
fun connect(): Flux<NewHeadMessage> {
val current = connected
if (current != null) {
return current
}
connectLock.withLock {
val currentRecheck = connected
if (currentRecheck != null) {
return currentRecheck
}
val created = ProduceNewHeads(upstream.getHead())
.start()
.publishOn(Schedulers.boundedElastic())
.publish()
.refCount(1, Duration.ofSeconds(60))
.doFinally {
// forget it on disconnect, so next time it's recreated
connected.remove(key)
connected = null
}
connected = created
return created
}
}
}

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

@@ -20,7 +20,6 @@ import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ChainFees
import io.emeraldpay.dshackle.upstream.EmptyHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.Multistream
@@ -32,7 +31,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import org.springframework.util.ConcurrentReferenceHashMap
import reactor.core.publisher.Mono
@Suppress("UNCHECKED_CAST")
@@ -51,8 +49,6 @@ open class EthereumPosMultiStream(
private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
private val feeEstimation = EthereumPriorityFees(this, reader, 256)
private val subscribe = EthereumSubscribe(this)
private val filteredHeads: MutableMap<String, Head> =
ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
init {
this.init()
@@ -73,7 +69,6 @@ open class EthereumPosMultiStream(
override fun stop() {
super.stop()
reader.stop()
filteredHeads.clear()
}
override fun isRunning(): Boolean {
@@ -142,24 +137,6 @@ open class EthereumPosMultiStream(
return subscribe
}
override fun getHead(mather: Selector.Matcher): Head =
filteredHeads.computeIfAbsent(mather.describeInternal().intern()) { _ ->
upstreams.filter { mather.matches(it) }
.apply {
log.debug("Found $size upstreams matching [${mather.describeInternal()}]")
}
.map { it.getHead() }
.let {
when (it.size) {
0 -> EmptyHead()
1 -> it.first()
else -> MergedHead(it, PriorityForkChoice()).apply {
start()
}
}
}
}
override fun getFeeEstimation(): ChainFees {
return feeEstimation
}