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

send original json of ws subscription heads (#413)

Browse Source
This commit is contained in:
a10zn8
2024-02-02 14:56:02 +03:00
committed by GitHub
parent 46bb122f1b
commit 5e94e521a3
2 changed files with 32 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.SubscriptionConnect
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage
import reactor.core.publisher.Flux
import reactor.core.scheduler.Scheduler
import java.time.Duration
@@ -30,11 +29,11 @@ import java.util.concurrent.ConcurrentHashMap
class ConnectNewHeads(
private val upstream: Multistream,
private val scheduler: Scheduler,
) : SubscriptionConnect<NewHeadMessage> {
) : SubscriptionConnect<ByteArray> {
private val connected: MutableMap<String, Flux<NewHeadMessage>> = ConcurrentHashMap()
private val connected: MutableMap<String, Flux<ByteArray>> = ConcurrentHashMap()
override fun connect(matcher: Selector.Matcher): Flux<NewHeadMessage> =
override fun connect(matcher: Selector.Matcher): Flux<ByteArray> =
connected.computeIfAbsent(matcher.describeInternal()) { key ->
ProduceNewHeads(upstream.getHead(matcher))
.start()

52
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt
View File

@@ -15,6 +15,7 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum.subscribe
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage
import io.emeraldpay.etherjar.hex.HexData
@@ -29,31 +30,36 @@ class ProduceNewHeads(
val head: Head,
) {
fun start(): Flux<NewHeadMessage> {
fun start(): Flux<ByteArray> {
return head.getFlux()
.map {
val block = it.toBlock()
NewHeadMessage(
block.number,
block.hash,
block.parentHash,
block.timestamp,
block.difficulty,
block.gasLimit,
block.gasUsed,
block.logsBloom,
block.miner,
block.baseFeePerGas?.amount,
block.extraData ?: HexData.empty(),
block.mixHash,
block.nonce,
block.receiptsRoot,
block.sha3Uncles,
block.stateRoot,
block.transactionsRoot,
block.withdrawalsRoot,
it.upstreamId,
)
if (it.full || it.json == null) {
val block = it.toBlock()
val msg = NewHeadMessage(
block.number,
block.hash,
block.parentHash,
block.timestamp,
block.difficulty,
block.gasLimit,
block.gasUsed,
block.logsBloom,
block.miner,
block.baseFeePerGas?.amount,
block.extraData ?: HexData.empty(),
block.mixHash,
block.nonce,
block.receiptsRoot,
block.sha3Uncles,
block.stateRoot,
block.transactionsRoot,
block.withdrawalsRoot,
it.upstreamId,
)
Global.objectMapper.writeValueAsBytes(msg)
} else {
it.json
}
}
}
}