diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt index b007c738..1d0e0046 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt @@ -28,6 +28,12 @@ class TxContainer( ) : SourceContainer(json, parsed) { companion object { + @JvmStatic + fun from(raw: ByteArray): TxContainer { + val tx = Global.objectMapper.readValue(raw, TransactionJson::class.java) + return from(tx, raw) + } + @JvmStatic fun from(tx: TransactionJson): TxContainer { return from(tx, Global.objectMapper.writeValueAsBytes(tx)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt index b30ff025..3d5d466e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt @@ -15,20 +15,18 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.Reader -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory -import org.springframework.beans.BeanUtils import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.util.function.Tuple2 +import reactor.util.function.Tuples +import java.nio.ByteBuffer +import java.util.function.BiFunction /** * Reads blocks with full transactions details. Based on data contained in readers for blocks @@ -46,40 +44,91 @@ class EthereumFullBlocksReader( private val log = LoggerFactory.getLogger(EthereumFullBlocksReader::class.java) } - override fun read(key: BlockId): Mono { - return blocks.read(key).flatMap { block -> - val block = Global.objectMapper.readValue(block.json, BlockJson::class.java) as BlockJson - val fullBlock = if (block.transactions == null || block.transactions.isEmpty()) { - // in fact it's not necessary to create a copy, made just for code clarity but it may be a performance loss - val fullBlock = BlockJson() - BeanUtils.copyProperties(block, fullBlock) - Mono.just(fullBlock) - } else { - Flux.fromIterable(block.transactions) - .map { TxId.from(it.hash) } - .flatMap { txes.read(it) } - .collectList() - .flatMap { list -> - if (block.transactions.size != list.size) { - Mono.empty>() - } else { - val fullBlock = BlockJson() - BeanUtils.copyProperties(block, fullBlock) - fullBlock.transactions = list.map { - Global.objectMapper.readValue(it.json, TransactionJson::class.java) - } - Mono.just(fullBlock) - } - } - } - fullBlock - .map { block -> - BlockContainer(block.number, BlockId.from(block.hash), block.totalDifficulty, block.timestamp, true, - Global.objectMapper.writeValueAsBytes(block), - block.transactions.map { tx -> TxId.from(tx) } - ) - } + private val accumulate: BiFunction = BiFunction { buf, x -> + if (buf.remaining() < x.size) { + val resize = ByteBuffer.allocate(buf.capacity() + buf.capacity() / 4 + x.size) + resize.put(buf.flip()).put(x) + } else { + buf.put(x) } } + override fun read(key: BlockId): Mono { + return blocks.read(key).flatMap { block -> + if (block.transactions.isEmpty()) { + // in fact it's not necessary to create a copy, made just for code clarity but it may be a performance loss + val fullBlock = BlockContainer( + block.height, block.hash, block.difficulty, block.timestamp, + true, + block.json, + block.parsed, + block.transactions + ) + return@flatMap Mono.just(fullBlock) + } + + val blockSplit = splitByTransactions(block.json!!) + + val transactions = Flux.fromIterable(block.transactions) + .flatMap { txes.read(it) } + .collectList() + + return@flatMap transactions.flatMap { transactionsData -> + // make sure that all transaction are loaded, otherwise just return empty because cannot make full block data + if (transactionsData.size != block.transactions.size) { + log.warn("No data to fill the block") + Mono.empty() + } else { + joinWithTransactions(blockSplit.t1, blockSplit.t2, Flux.fromIterable(transactionsData).map { it.json!! }) + .reduce(ByteBuffer.allocate(block.json.size * 4), accumulate) + .map { it.flip().array() } + .map { json -> + BlockContainer(block.height, block.hash, block.difficulty, block.timestamp, + true, + json, + null, + block.transactions + ) + } + } + } + } + } + + fun splitByTransactions(json: ByteArray): Tuple2 { + //TODO find a lib that implements Knuth-Morris-Pratt Pattern Matching Algorithm for byte arrays + // and reimplement without making a string copy from bytes + + val s = String(json) + val fieldStart = s.indexOf("\"transactions\"") + val arrayStart = s.indexOf("[", fieldStart) + val arrayEnd = s.indexOf("]", arrayStart) + + val head = s.substring(0, arrayStart + 1) + val tail = s.substring(arrayEnd, s.length) + + return Tuples.of(head.toByteArray(), tail.toByteArray()) + } + + fun joinWithTransactions(head: ByteArray, tail: ByteArray, transactions: Flux): Flux { + val separator = Flux.range(0, Integer.MAX_VALUE) + .map { it != 0 } + + val transactionsWithSeparator = transactions.zipWith(separator) + .flatMap { + val tx = Flux.just(it.t1) + if (it.t2) { + Flux.concat(Flux.just(",".toByteArray()), tx) + } else { + tx + } + } + + return Flux.concat( + Flux.just(head), + transactionsWithSeparator, + Flux.just(tail) + ) + } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt index 14fa8c3b..e80fa386 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt @@ -158,11 +158,15 @@ class NativeCallRouter( throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number") } val withTx = params[1].toString().toBoolean() - return if (withTx) { - log.warn("Block by number is not implemented") - null + var block = reader.blocksByHeightAsCont() + .read(number) + block = if (withTx) { + block.flatMap { + fullBlocksReader.read(it.hash) + } } else { - reader.blocksByHeightAsCont().read(number).map { it.json!! } + block } + return block.map { it.json!! } } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy index 326a966b..3c50485f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy @@ -15,6 +15,8 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import com.fasterxml.jackson.core.PrettyPrinter +import com.fasterxml.jackson.core.util.DefaultPrettyPrinter import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.BlocksMemCache @@ -22,6 +24,8 @@ import io.emeraldpay.dshackle.cache.TxMemCache import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxContainer +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.test.ReaderMock import io.emeraldpay.dshackle.test.TestingCommons import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId @@ -200,7 +204,7 @@ class EthereumFullBlocksReaderSpec extends Specification { act.transactions.size() == 0 } - def "Return nothing if no transactions"() { + def "Return nothing if some of tx are unavailable"() { setup: def txes = new TxMemCache() def blocks = new BlocksMemCache() @@ -234,4 +238,140 @@ class EthereumFullBlocksReaderSpec extends Specification { then: act == null } + + def "Doesn't change original cached json"() { + setup: + def txes = new TxMemCache() + def blocks = new BlocksMemCache() + def blockJson = ''' + { + "number": "0x100001", + "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446", + "timestamp": "0x56cc7b8c", + "totalDifficulty": "0x6baba0399a0f2e73", + "extraField": "extraValue", + "transactions": [ + "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77", + "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e" + ], + "extraField2": "extraValue2" + } + ''' + blocks.add(BlockContainer.from(blockJson.bytes)) + + def tx1 = ''' + { + "hash": "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77", + "blockHash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446", + "blockNumber": "0x100001", + "extraField3": "extraValue3" + } + ''' + def tx2 = ''' + { + "hash": "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e", + "blockHash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446", + "blockNumber": "0x100001", + "from": "0x2a65aca4d5fc5b5c859090a6c34d164135398226", + "extraField4": "extraValue4" + } + ''' + txes.add(TxContainer.from(tx1.bytes)) + txes.add(TxContainer.from(tx2.bytes)) + + def full = new EthereumFullBlocksReader(blocks, txes) + + def blockJsonExpected = ''' + { + "number": "0x100001", + "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446", + "timestamp": "0x56cc7b8c", + "totalDifficulty": "0x6baba0399a0f2e73", + "extraField": "extraValue", + "transactions": [ + ''' + tx1 + ', ' + tx2 + + ''' ], + "extraField2": "extraValue2" + } + ''' + blockJsonExpected = Global.objectMapper.readValue(blockJsonExpected, Map) + + def prettyJson = Global.objectMapper.writer(new DefaultPrettyPrinter()) + blockJsonExpected = prettyJson.writeValueAsString(blockJsonExpected) + + when: + def act = full.read(BlockId.from("0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446")).block() + act = prettyJson.writeValueAsString(Global.objectMapper.readValue(act.json, Map)) + + then: + act.contains("extraValue") + act.contains("extraValue2") + act.contains("extraValue3") + act.contains("extraValue4") + act == blockJsonExpected + } + + def "Split block with tx"() { + setup: + def blockJson = ''' + { + "number": "0x100001", + "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446", + "timestamp": "0x56cc7b8c", + "totalDifficulty": "0x6baba0399a0f2e73", + "extraField": "extraValue", + "transactions": [ + "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77", + "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e" + ], + "extraField2": "extraValue2" + } + ''' + def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader)) + + when: + def act = reader.splitByTransactions(blockJson.bytes) + then: + new String(act.getT1()).endsWith('"transactions": [') + new String(act.getT2()).startsWith('],') + new String(act.getT2()).trim().endsWith('}') + } + + def "Split block with tx if formatted with space"() { + setup: + def blockJson = ''' + { + "number": "0x100001", + "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446", + "timestamp": "0x56cc7b8c", + "totalDifficulty": "0x6baba0399a0f2e73", + "extraField": "extraValue", + "transactions" : [ + "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77", + "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e" + ] , + "extraField2": "extraValue2" + } + ''' + def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader)) + + when: + def act = reader.splitByTransactions(blockJson.bytes) + then: + new String(act.getT1()).endsWith('"transactions" : [') + new String(act.getT2()).startsWith('] ,') + new String(act.getT2()).trim().endsWith('}') + } + + def "Split block with tx if formatted without space"() { + setup: + def blockJson = '{"extraField": "extraValue","transactions":["0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77","0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"]}' + def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader)) + + when: + def act = reader.splitByTransactions(blockJson.bytes) + then: + new String(act.getT1()).endsWith('"transactions":[') + new String(act.getT2()) == ']}' + } }