diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 620efcb2..7905b3dd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -46,19 +46,12 @@ class StreamHead( .getFlux() .map { asProto(chain, it!!) } .onErrorContinue { t, _ -> - log.warn("Head error: ${t.message}") + log.warn("Head subscription error: ${t.message}") } } } fun asProto(chain: Chain, block: BlockContainer): BlockchainOuterClass.ChainHead { - if (BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM) { - return asEthereumProto(chain, block) - } - throw IllegalArgumentException("Unsupported blockchain ${chain}") - } - - fun asEthereumProto(chain: Chain, block: BlockContainer): BlockchainOuterClass.ChainHead { return BlockchainOuterClass.ChainHead.newBuilder() .setChainValue(chain.id) .setHeight(block.height) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 368ddff4..d5dc1f7a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -78,7 +78,7 @@ abstract class AbstractHead : Head { } override fun getFlux(): Flux { - return Flux.merge( + return Flux.concat( Mono.justOrEmpty(head.get()), Flux.from(stream) ).onBackpressureLatest()