diff --git a/buildSrc/src/main/kotlin/chainsconfig.codegen.gradle.kts b/buildSrc/src/main/kotlin/chainsconfig.codegen.gradle.kts index 362f1073..19d80082 100644 --- a/buildSrc/src/main/kotlin/chainsconfig.codegen.gradle.kts +++ b/buildSrc/src/main/kotlin/chainsconfig.codegen.gradle.kts @@ -137,6 +137,7 @@ open class CodeGen(private val config: ChainsConfig) { "eth-beacon-chain" -> "BlockchainType.ETHEREUM_BEACON_CHAIN" "ton" -> "BlockchainType.TON" "cosmos" -> "BlockchainType.COSMOS" + "ripple" -> "BlockchainType.RIPPLE" else -> throw IllegalArgumentException("unknown blockchain type $type") } } diff --git a/emerald-grpc b/emerald-grpc index a8716b1a..cfbe85e4 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit a8716b1a3a91f8a8d9e176ad9b454b9d4711d60b +Subproject commit cfbe85e4f6cc29708cdf4c6ece55bd7e28b0e89d diff --git a/foundation/src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt b/foundation/src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt index 72f3dd8a..eab4591e 100644 --- a/foundation/src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt +++ b/foundation/src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt @@ -12,7 +12,8 @@ enum class BlockchainType( NEAR(ApiType.JSON_RPC), ETHEREUM_BEACON_CHAIN(ApiType.REST), COSMOS(ApiType.JSON_RPC), - TON(ApiType.REST); + TON(ApiType.REST), + RIPPLE(ApiType.JSON_RPC),; } enum class ApiType { diff --git a/foundation/src/main/resources/public b/foundation/src/main/resources/public index b0d89a5d..f3516dd2 160000 --- a/foundation/src/main/resources/public +++ b/foundation/src/main/resources/public @@ -1 +1 @@ -Subproject commit b0d89a5d942f0be0e0a81cd37d5bb5b8aebf2846 +Subproject commit f3516dd2a8859432056a3e7ab0534712a8c7a48e diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index a5f73f62..2d7b70dd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -16,6 +16,7 @@ */ package io.emeraldpay.dshackle.rpc +import com.fasterxml.jackson.databind.JsonNode import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass @@ -481,17 +482,24 @@ open class NativeCall( ReaderData(ctx.upstream, ctx.upstreamFilter, ctx.callQuorum, signer, tracer), ) val counter = reader.attempts() + val isRipple = ctx.upstream.getChain() in listOf(Chain.RIPPLE__MAINNET, Chain.RIPPLE__TESTNET) + var streamRequest = ctx.streamRequest + if (isRipple) { + streamRequest = false + } return SpannedReader(reader, tracer, RPC_READER) - .read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, ctx.streamRequest)) + .read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, streamRequest)) .map { val resolvedUpstreamData = it.resolvedUpstreamData.ifEmpty { ctx.upstream.getUpstreamSettingsData()?.run { listOf(this) } ?: emptyList() } if (it.stream == null) { - val bytes = ctx.resultDecorator.processResult(it) - validateResult(bytes, "remote", ctx) - CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, resolvedUpstreamData, ctx) + if (isRipple) { + callRippleResult(ctx, it, resolvedUpstreamData) + } else { + callResult(ctx, it, resolvedUpstreamData) + } } else { CallResult.ok(ctx.id, ctx.nonce, ByteArray(0), it.signature, resolvedUpstreamData, ctx, it.stream) } @@ -515,6 +523,67 @@ open class NativeCall( ) } + private fun callResult( + ctx: ValidCallContext, + it: RequestReader.Result, + resolvedUpstreamData: List, + ): CallResult { + val bytes = ctx.resultDecorator.processResult(it) + validateResult(bytes, "remote", ctx) + return CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, resolvedUpstreamData, ctx) + } + + private fun callRippleResult( + ctx: ValidCallContext, + it: RequestReader.Result, + resolvedUpstreamData: List, + ): CallResult { + return try { + val responseJson = Global.objectMapper.readTree(it.value) + if (isRippleErrorResponse(responseJson)) { + return createRippleErrorResult(ctx, it, responseJson) + } + callResult(ctx, it, resolvedUpstreamData) + } catch (e: Exception) { + log.warn("Failed to parse Ripple response: ${e.message}", e) + callResult(ctx, it, resolvedUpstreamData) + } + } + + private fun isRippleErrorResponse(responseJson: JsonNode): Boolean = + !responseJson.has("status") || responseJson.get("status").asText() == "error" + + private fun createRippleErrorResult( + ctx: ValidCallContext, + result: RequestReader.Result, + responseJson: JsonNode, + ): CallResult { + val error = responseJson.get("error")?.asText().orEmpty() + val errorMessage = responseJson.get("error_message")?.asText().orEmpty() + val fullErrorMessage = if (error.isNotEmpty() && errorMessage.isNotEmpty()) { + "$error: $errorMessage" + } else { + error.ifEmpty { errorMessage.ifEmpty { "Ripple request failed" } } + } + + val errorCode = responseJson.get("error_code")?.asInt() ?: RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE + + return CallResult.fail( + ctx.id, + ctx.nonce, + CallError( + ctx.id, + fullErrorMessage, + ChainCallError( + errorCode, + fullErrorMessage, + ), + String(result.value), + ), + ctx, + ) + } + private fun validateResult(bytes: ByteArray, origin: String, ctx: ValidCallContext) { if (bytes.isEmpty() || nullValue.contentEquals(bytes)) { log.warn("Empty result from origin $origin, method ${ctx.payload.method}, params ${ctx.payload.params}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt index 7704e2a4..6a22675b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt @@ -6,6 +6,7 @@ import io.emeraldpay.dshackle.BlockchainType.ETHEREUM import io.emeraldpay.dshackle.BlockchainType.ETHEREUM_BEACON_CHAIN import io.emeraldpay.dshackle.BlockchainType.NEAR import io.emeraldpay.dshackle.BlockchainType.POLKADOT +import io.emeraldpay.dshackle.BlockchainType.RIPPLE import io.emeraldpay.dshackle.BlockchainType.SOLANA import io.emeraldpay.dshackle.BlockchainType.STARKNET import io.emeraldpay.dshackle.BlockchainType.TON @@ -20,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.calls.DefaultCosmosMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.DefaultNearMethods import io.emeraldpay.dshackle.upstream.calls.DefaultPolkadotMethods +import io.emeraldpay.dshackle.upstream.calls.DefaultRippleMethods import io.emeraldpay.dshackle.upstream.calls.DefaultStarknetMethods import io.emeraldpay.dshackle.upstream.calls.DefaultTonHttpMethods import org.springframework.stereotype.Component @@ -51,6 +53,7 @@ class CallTargetsHolder { ETHEREUM_BEACON_CHAIN -> DefaultBeaconChainMethods() COSMOS -> DefaultCosmosMethods() TON -> DefaultTonHttpMethods(connection) + RIPPLE -> DefaultRippleMethods() UNKNOWN -> throw IllegalArgumentException("unknown chain") } callTargets[chain] = created diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultRippleMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultRippleMethods.kt new file mode 100644 index 00000000..abbb5747 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultRippleMethods.kt @@ -0,0 +1,95 @@ +package io.emeraldpay.dshackle.upstream.calls + +import io.emeraldpay.dshackle.quorum.AlwaysQuorum +import io.emeraldpay.dshackle.quorum.BroadcastQuorum +import io.emeraldpay.dshackle.quorum.CallQuorum +import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException + +class DefaultRippleMethods : CallMethods { + + private val all = setOf( + "account_channels", + "account_currencies", + "account_info", + "account_lines", + "account_nfts", + "account_objects", + "account_offers", + "account_tx", + "gateway_balances", + "noripple_check", + "ledger", + "ledger_closed", + "ledger_current", + "ledger_data", + "ledger_entry", + "transaction_entry", + "tx", + "tx_history", + "book_offers", + "deposit_authorized", + "nft_buy_offers", + "nft_sell_offers", + "path_find", + "ripple_path_find", + "channel_authorize", + "channel_verify", + "subscribe", + "unsubscribe", + "fee", + "manifest", + "server_info", + "server_state", + "ledger_index", + "nft_history", + "nft_info", + "nfts_by_issuer", + "ping", + "random", +// "amm_info", +// "book_changes", +// "get_aggregate_price", +// "server_definitions", +// "version", +// "server_info", +// "ledger", +// "mpt_holders", +// "version", + ) + + private val add = setOf( + "submit", + "submit_multisigned", + ) + + private val allowedMethods: Set = all + add + + override fun createQuorumFor(method: String): CallQuorum { + if (add.contains(method)) { + return BroadcastQuorum() + } + return AlwaysQuorum() + } + + override fun isCallable(method: String): Boolean { + return allowedMethods.contains(method) + } + + override fun isHardcoded(method: String): Boolean { + return false + } + + override fun executeHardcoded(method: String): ByteArray { + throw RpcException(-32601, "Method not found") + } + + override fun getGroupMethods(groupName: String): Set = + when (groupName) { + "default" -> getSupportedMethods() + else -> emptyList() + }.toSet() + + override fun getSupportedMethods(): Set { + return allowedMethods.toSortedSet() + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt index b060c113..65ae2d13 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -6,6 +6,7 @@ import io.emeraldpay.dshackle.BlockchainType.ETHEREUM import io.emeraldpay.dshackle.BlockchainType.ETHEREUM_BEACON_CHAIN import io.emeraldpay.dshackle.BlockchainType.NEAR import io.emeraldpay.dshackle.BlockchainType.POLKADOT +import io.emeraldpay.dshackle.BlockchainType.RIPPLE import io.emeraldpay.dshackle.BlockchainType.SOLANA import io.emeraldpay.dshackle.BlockchainType.STARKNET import io.emeraldpay.dshackle.BlockchainType.TON @@ -40,6 +41,7 @@ import io.emeraldpay.dshackle.upstream.finalization.FinalizationDetector import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.near.NearChainSpecific import io.emeraldpay.dshackle.upstream.polkadot.PolkadotChainSpecific +import io.emeraldpay.dshackle.upstream.ripple.RippleChainSpecific import io.emeraldpay.dshackle.upstream.solana.SolanaChainSpecific import io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific import io.emeraldpay.dshackle.upstream.ton.TonHttpSpecific @@ -112,6 +114,7 @@ object ChainSpecificRegistry { ETHEREUM_BEACON_CHAIN -> BeaconChainSpecific TON -> TonHttpSpecific COSMOS -> CosmosChainSpecific + RIPPLE -> RippleChainSpecific BITCOIN -> throw IllegalArgumentException("bitcoin should use custom streams implementation") UNKNOWN -> throw IllegalArgumentException("unknown chain") } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt new file mode 100644 index 00000000..a1b0a48d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt @@ -0,0 +1,235 @@ +package io.emeraldpay.dshackle.upstream.ripple + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties +import com.fasterxml.jackson.annotation.JsonProperty +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.foundation.ChainOptions.Options +import io.emeraldpay.dshackle.reader.ChainReader +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.GenericSingleCallValidator +import io.emeraldpay.dshackle.upstream.SingleValidator +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult +import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.math.BigInteger +import java.time.Instant + +object RippleChainSpecific : AbstractPollChainSpecific() { + + private val log = LoggerFactory.getLogger(RippleChainSpecific::class.java) + + override fun parseBlock(data: ByteArray, upstreamId: String, api: ChainReader): Mono { + val result = Global.objectMapper.readValue(data, RippleBlock::class.java) + val block = result.closed.ledger + var height: Long = 0 + try { + height = block.ledgerIndex.toLong() + } catch (e: NumberFormatException) { + log.error("Invalid ledgerIndex $block.ledgerIndex, upstreamId:$upstreamId ", result) + } + + return Mono.just( + BlockContainer( + height = height, + hash = BlockId.from(block.ledgerHash ?: ""), + difficulty = BigInteger.ZERO, + timestamp = Instant.EPOCH, + full = false, + json = data, + parsed = result, + transactions = emptyList(), + upstreamId = upstreamId, + parentHash = BlockId.from(block.parentHash), + ), + ) + } + + override fun getFromHeader(data: ByteArray, upstreamId: String, api: ChainReader): Mono { + throw NotImplementedError() + } + + override fun listenNewHeadsRequest(): ChainRequest { + throw NotImplementedError() + } + + override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest { + throw NotImplementedError() + } + + override fun upstreamValidators( + chain: Chain, + upstream: Upstream, + options: Options, + config: ChainConfig, + ): List> { + return listOf( + GenericSingleCallValidator( + ChainRequest("server_state", ListParams()), + upstream, + ) { data -> validate(data) }, + ) + } + + override fun upstreamSettingsValidators( + chain: Chain, + upstream: Upstream, + options: Options, + config: ChainConfig, + ): List> { + return listOf( + GenericSingleCallValidator( + ChainRequest("server_state", ListParams()), + upstream, + ) { data -> + validateSettings(data, chain) + }, + ) + } + + override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { + return RippleLowerBoundService(chain, upstream) + } + + fun validate(data: ByteArray): UpstreamAvailability { + val resp = Global.objectMapper.readValue(data, RippleState::class.java) + return when (resp.state.serverState) { + "full", "proposing" -> UpstreamAvailability.OK + "connected" -> UpstreamAvailability.SYNCING + else -> UpstreamAvailability.UNAVAILABLE + } + } + + fun validateSettings(data: ByteArray, chain: Chain): ValidateUpstreamSettingsResult { + val resp = Global.objectMapper.readValue(data, RippleState::class.java) + return if (chain.chainId.isNotEmpty() && resp.state.networkId.toString() != chain.chainId.lowercase()) { + ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR + } else { + ValidateUpstreamSettingsResult.UPSTREAM_VALID + } + } + + override fun latestBlockRequest(): ChainRequest = + ChainRequest("ledger", ListParams()) +} + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RippleBlock( + @JsonProperty("closed") var closed: RippleClosed, + @JsonProperty("open") var open: RippleOpen?, + @JsonProperty("status ") var status: String?, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RippleClosed( + @JsonProperty("ledger") var ledger: RippleClosedLedger, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RippleOpen( + @JsonProperty("ledger") var ledger: RippleOpenLedger, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RippleClosedLedger( + @JsonProperty("account_hash") var accountHash: String, + @JsonProperty("close_flags") var closeFlags: Short, + @JsonProperty("close_time") var closeTime: Long, + @JsonProperty("close_time_human") var closeTimeHuman: String, + @JsonProperty("close_time_iso") var closeTimeIso: String, + @JsonProperty("close_time_resolution") var closeTimeResolution: Short, + @JsonProperty("closed") var closed: Boolean, + @JsonProperty("ledger_hash") var ledgerHash: String, + @JsonProperty("ledger_index") var ledgerIndex: String, + @JsonProperty("parent_close_time") var parentCloseTime: Long, + @JsonProperty("parent_hash") var parentHash: String, + @JsonProperty("total_coins") var totalCoins: String, + @JsonProperty("transaction_hash") var transactionHash: String, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RippleOpenLedger( + @JsonProperty("closed") var closed: Boolean, + @JsonProperty("ledger_index") var ledgerIndex: String, + @JsonProperty("parent_hash") var parentHash: String, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RippleState( + @JsonProperty("state") var state: RippleServerState, + @JsonProperty("status") var status: String? = null, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RippleServerState( + @JsonProperty("build_version") val buildVersion: String, + @JsonProperty("complete_ledgers") val completeLedgers: String, + @JsonProperty("initial_sync_duration_us") val initialSyncDurationUs: String, + @JsonProperty("io_latency_ms") val ioLatencyMs: Int, + @JsonProperty("jq_trans_overflow") val jqTransOverflow: String, + @JsonProperty("last_close") val lastClose: LastClose, + @JsonProperty("load_base") val loadBase: Int, + @JsonProperty("load_factor") val loadFactor: Int, + @JsonProperty("load_factor_fee_escalation") val loadFactorFeeEscalation: Int, + @JsonProperty("load_factor_fee_queue") val loadFactorFeeQueue: Int, + @JsonProperty("load_factor_fee_reference") val loadFactorFeeReference: Int, + @JsonProperty("load_factor_server") val loadFactorServer: Int, + @JsonProperty("network_id") val networkId: Int, + @JsonProperty("peer_disconnects") val peerDisconnects: String, + @JsonProperty("peer_disconnects_resources") val peerDisconnectsResources: String, + @JsonProperty("peers") val peers: Int, + @JsonProperty("ports") val ports: List, + @JsonProperty("pubkey_node") val pubkeyNode: String, + @JsonProperty("server_state") val serverState: String, + @JsonProperty("server_state_duration_us") val serverStateDurationUs: String, + @JsonProperty("state_accounting") val stateAccounting: StateAccounting, + @JsonProperty("time") val time: String, + @JsonProperty("uptime") val uptime: Long, + @JsonProperty("validated_ledger") val validatedLedger: ValidatedLedger, + @JsonProperty("validation_quorum") val validationQuorum: Int, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class LastClose( + @JsonProperty("converge_time") val convergeTime: Int, + @JsonProperty("proposers") val proposers: Int, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class Port( + @JsonProperty("port") val port: String, + @JsonProperty("protocol") val protocol: List, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class StateAccounting( + @JsonProperty("connected") val connected: StateDuration, + @JsonProperty("disconnected") val disconnected: StateDuration, + @JsonProperty("full") val full: StateDuration, + @JsonProperty("syncing") val syncing: StateDuration, + @JsonProperty("tracking") val tracking: StateDuration, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class StateDuration( + @JsonProperty("duration_us") val durationUs: String, + @JsonProperty("transitions") val transitions: String, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class ValidatedLedger( + @JsonProperty("base_fee") val baseFee: Int, + @JsonProperty("close_time") val closeTime: Long, + @JsonProperty("hash") val hash: String, + @JsonProperty("reserve_base") val reserveBase: Long, + @JsonProperty("reserve_inc") val reserveInc: Long, + @JsonProperty("seq") val seq: Long, +) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundService.kt new file mode 100644 index 00000000..173f32f8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundService.kt @@ -0,0 +1,15 @@ +package io.emeraldpay.dshackle.upstream.ripple + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService + +class RippleLowerBoundService( + chain: Chain, + private val upstream: Upstream, +) : LowerBoundService(chain, upstream) { + override fun detectors(): List { + return listOf(RippleLowerBoundStateDetector(upstream)) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundStateDetector.kt new file mode 100644 index 00000000..5daebe30 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundStateDetector.kt @@ -0,0 +1,24 @@ +package io.emeraldpay.dshackle.upstream.ripple + +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import reactor.core.publisher.Flux + +class RippleLowerBoundStateDetector( + private val upstream: Upstream, +) : LowerBoundDetector(upstream.getChain()) { + + override fun period(): Long { + return 120 + } + + override fun internalDetectLowerBound(): Flux { + return Flux.just(LowerBoundData(1, LowerBoundType.STATE)) + } + + override fun types(): Set { + return setOf(LowerBoundType.STATE) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt index bdd9c429..b51ab438 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt @@ -218,7 +218,8 @@ abstract class ResponseParser { private val isResultSet = result != null || nullResult - val isRpcReady: Boolean = id != null && + // Ripple Response doesn't have `id` field + val isRpcReady: Boolean = // id != null && (error != null || isResultSet) val isSubReady: Boolean = subId != null && diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt index ca465105..b99c65c2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt @@ -30,14 +30,6 @@ class ResponseWSParser : ResponseParser() { } override fun build(state: Preparsed): WsResponse { - if (state.isRpcReady) { - return WsResponse( - Type.RPC, - state.id!!, - if (state.nullResult) NULL_RESULT else state.result, - state.error, - ) - } if (state.isSubReady) { return WsResponse( Type.SUBSCRIPTION, @@ -46,6 +38,14 @@ class ResponseWSParser : ResponseParser() { state.error, ) } + if (state.isRpcReady) { + return WsResponse( + Type.RPC, + state.id!!, + if (state.nullResult) NULL_RESULT else state.result, + state.error, + ) + } throw IllegalStateException("State is not ready") }