Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added block add event. #637

Merged
merged 5 commits into from
Apr 2, 2020
Merged

Added block add event. #637

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
22ea9f5
Added block add event.
Apr 1, 2020
0bfe48e
Merge branch 'master' into PIE-2242_2
davemec Apr 2, 2020
75279a4
Added more fields to TransactionReceipt. Added missing JavaDoc
Apr 2, 2020
5234bed
Merge branch 'PIE-2242_2' of github.com:davemec/besu into PIE-2242_2
Apr 2, 2020
1d3b013
Merge branch 'master' into PIE-2242_2
davemec Apr 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 57 additions & 5 deletions besu/src/main/java/org/hyperledger/besu/services/BesuEventsImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,11 @@
import org.hyperledger.besu.ethereum.core.Difficulty;
import org.hyperledger.besu.ethereum.core.LogTopic;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.TransactionReceipt;
import org.hyperledger.besu.ethereum.eth.sync.BlockBroadcaster;
import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
import org.hyperledger.besu.plugin.data.AddedBlockContext;
import org.hyperledger.besu.plugin.data.Address;
import org.hyperledger.besu.plugin.data.BlockHeader;
import org.hyperledger.besu.plugin.data.PropagatedBlockContext;
Expand Down Expand Up @@ -66,6 +68,38 @@ public void removeBlockPropagatedListener(final long listenerIdentifier) {
blockBroadcaster.unsubscribePropagateNewBlocks(listenerIdentifier);
}

@Override
public long addBlockAddedListener(final BlockAddedListener listener) {
return blockchain.observeBlockAdded(
(event, chain) ->
listener.onBlockAdded(
blockAddedContext(
event.getBlock()::getHeader,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wow... that compiled? I didn't know that was possible.

event.getBlock()::getBody,
event::getTransactionReceipts)));
}

@Override
public void removeBlockAddedListener(final long listenerIdentifier) {
blockchain.removeObserver(listenerIdentifier);
}

@Override
public long addBlockReorgListener(final BlockReorgListener listener) {
return blockchain.observeChainReorg(
(blockWithReceipts, chain) ->
listener.onBlockReorg(
blockAddedContext(
blockWithReceipts::getHeader,
blockWithReceipts.getBlock()::getBody,
blockWithReceipts::getReceipts)));
}

@Override
public void removeBlockReorgListener(final long listenerIdentifier) {
blockchain.removeObserver(listenerIdentifier);
}

@Override
public long addTransactionAddedListener(final TransactionAddedListener listener) {
return transactionPool.subscribePendingTransactions(listener::onTransactionAdded);
Expand Down Expand Up @@ -109,11 +143,7 @@ public long addLogListener(
.collect(toUnmodifiableList());
final List<List<LogTopic>> besuTopics =
topics.stream()
.map(
subList ->
subList.stream()
.map(bytes -> LogTopic.wrap(bytes))
.collect(toUnmodifiableList()))
.map(subList -> subList.stream().map(LogTopic::wrap).collect(toUnmodifiableList()))
.collect(toUnmodifiableList());

final LogsQuery logsQuery = new LogsQuery(besuAddresses, besuTopics);
Expand Down Expand Up @@ -152,4 +182,26 @@ public UInt256 getTotalDifficulty() {
}
};
}

private static AddedBlockContext blockAddedContext(
final Supplier<BlockHeader> blockHeaderSupplier,
final Supplier<BlockBody> blockBodySupplier,
final Supplier<List<TransactionReceipt>> transactionReceiptsSupplier) {
return new AddedBlockContext() {
@Override
public BlockHeader getBlockHeader() {
return blockHeaderSupplier.get();
}

@Override
public BlockBody getBlockBody() {
return blockBodySupplier.get();
}

@Override
public List<TransactionReceipt> getTransactionReceipts() {
return transactionReceiptsSupplier.get();
}
};
}
}
127 changes: 127 additions & 0 deletions besu/src/test/java/org/hyperledger/besu/services/BesuEventsImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
import org.hyperledger.besu.ethereum.core.Difficulty;
import org.hyperledger.besu.ethereum.core.TransactionReceipt;
import org.hyperledger.besu.ethereum.core.TransactionTestFixture;
import org.hyperledger.besu.ethereum.core.Wei;
import org.hyperledger.besu.ethereum.core.WorldState;
Expand All @@ -50,6 +51,7 @@
import org.hyperledger.besu.ethereum.storage.keyvalue.KeyValueStoragePrefixedKeyBlockchainStorage;
import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
import org.hyperledger.besu.plugin.data.AddedBlockContext;
import org.hyperledger.besu.plugin.data.LogWithMetadata;
import org.hyperledger.besu.plugin.data.PropagatedBlockContext;
import org.hyperledger.besu.plugin.data.SyncStatus;
Expand Down Expand Up @@ -228,6 +230,131 @@ public void newBlockEventUselessUnsubscribesCompletes() {
serviceImpl.removeBlockPropagatedListener(5L);
}

@Test
public void addedBlockEventFiresAfterSubscribe() {
final AtomicReference<AddedBlockContext> result = new AtomicReference<>();
serviceImpl.addBlockAddedListener(result::set);
assertThat(result.get()).isNull();

final var block =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash()));
List<TransactionReceipt> transactionReceipts = gen.receipts(block);
blockchain.appendBlock(block, transactionReceipts);
assertThat(result.get()).isNotNull();
assertThat(result.get().getBlockHeader()).isEqualTo(block.getHeader());
assertThat(result.get().getTransactionReceipts()).isEqualTo(transactionReceipts);
}

@Test
public void addedBlockEventDoesNotFireAfterUnsubscribe() {
final AtomicReference<AddedBlockContext> result = new AtomicReference<>();
final long id = serviceImpl.addBlockAddedListener(result::set);
assertThat(result.get()).isNull();

serviceImpl.removeBlockAddedListener(id);
result.set(null);

final var block =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash()));
blockchain.appendBlock(block, gen.receipts(block));
assertThat(result.get()).isNull();
}

@Test
public void additionWithoutSubscriptionsCompletes() {
final var block =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash()));
blockchain.appendBlock(block, gen.receipts(block));
}

@Test
public void addedBlockEventUselessUnsubscribesCompletes() {
serviceImpl.removeBlockAddedListener(5);
serviceImpl.removeBlockAddedListener(5L);
}

@Test
public void reorgedBlockEventFiresAfterSubscribe() {
final AtomicReference<AddedBlockContext> result = new AtomicReference<>();
serviceImpl.addBlockReorgListener(result::set);
assertThat(result.get()).isNull();

final var block =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash())
.setBlockNumber(blockchain.getGenesisBlock().getHeader().getNumber() + 1));
blockchain.appendBlock(block, gen.receipts(block));
assertThat(result.get()).isNull();

final var reorgBlock =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash())
.setBlockNumber(blockchain.getGenesisBlock().getHeader().getNumber() + 2));
List<TransactionReceipt> transactionReceipts = gen.receipts(reorgBlock);
blockchain.appendBlock(reorgBlock, transactionReceipts);
assertThat(result.get()).isNotNull();
assertThat(result.get().getBlockHeader()).isEqualTo(reorgBlock.getHeader());
assertThat(result.get().getTransactionReceipts()).isEqualTo(transactionReceipts);
}

@Test
public void reorgedBlockEventDoesNotFireAfterUnsubscribe() {
final AtomicReference<AddedBlockContext> result = new AtomicReference<>();
final long id = serviceImpl.addBlockReorgListener(result::set);
assertThat(result.get()).isNull();

serviceImpl.removeBlockReorgListener(id);
result.set(null);

final var block =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash())
.setBlockNumber(blockchain.getGenesisBlock().getHeader().getNumber() + 1));
blockchain.appendBlock(block, gen.receipts(block));
assertThat(result.get()).isNull();

final var reorgBlock =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash())
.setBlockNumber(blockchain.getGenesisBlock().getHeader().getNumber() + 1));
blockchain.appendBlock(reorgBlock, gen.receipts(reorgBlock));
assertThat(result.get()).isNull();
}

@Test
public void reorgWithoutSubscriptionsCompletes() {
final var block =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash())
.setBlockNumber(blockchain.getGenesisBlock().getHeader().getNumber() + 1));
blockchain.appendBlock(block, gen.receipts(block));

final var reorgBlock =
gen.block(
new BlockDataGenerator.BlockOptions()
.setParentHash(blockchain.getGenesisBlock().getHash())
.setBlockNumber(blockchain.getGenesisBlock().getHeader().getNumber() + 1));
List<TransactionReceipt> transactionReceipts = gen.receipts(reorgBlock);
blockchain.appendBlock(reorgBlock, transactionReceipts);
}

@Test
public void reorgedBlockEventUselessUnsubscribesCompletes() {
serviceImpl.removeBlockReorgListener(5);
serviceImpl.removeBlockReorgListener(5L);
}

@Test
public void transactionAddedEventFiresAfterSubscribe() {
final AtomicReference<Transaction> result = new AtomicReference<>();
Expand Down
3 changes: 2 additions & 1 deletion ...est/java/org/hyperledger/besu/consensus/ibft/blockcreation/IbftMiningCoordinatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,8 @@ public void setsTheExtraData() {
@Test
public void addsNewChainHeadEventWhenNewCanonicalHeadBlockEventReceived() throws Exception {
BlockAddedEvent headAdvancement =
BlockAddedEvent.createForHeadAdvancement(block, Collections.emptyList());
BlockAddedEvent.createForHeadAdvancement(
block, Collections.emptyList(), Collections.emptyList());
ibftMiningCoordinator.onBlockAdded(headAdvancement, blockChain);

assertThat(eventQueue.size()).isEqualTo(1);
Expand Down
4 changes: 2 additions & 2 deletions ...n/java/org/hyperledger/besu/ethereum/api/query/AutoTransactionLogBloomCachingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,9 @@ public void start() {
chainReorgSubscriptionId =
OptionalLong.of(
blockchain.observeChainReorg(
(header, __) ->
(blockWithReceipts, __) ->
transactionLogBloomCacher.cacheLogsBloomForBlockHeader(
header, Optional.empty(), true)));
blockWithReceipts.getHeader(), Optional.empty(), true)));

transactionLogBloomCacher
.getScheduler()
Expand Down
5 changes: 4 additions & 1 deletion ...org/hyperledger/besu/ethereum/api/jsonrpc/internal/filter/FilterManagerLogFilterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.hyperledger.besu.ethereum.core.LogWithMetadata;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;

import java.util.Collections;
import java.util.List;
import java.util.Optional;

Expand Down Expand Up @@ -149,7 +150,9 @@ private void recordNewBlockEvent() {
final Block block = gen.block();
filterManager.recordBlockEvent(
BlockAddedEvent.createForHeadAdvancement(
block, LogWithMetadata.generate(block, gen.receipts(block), false)),
block,
LogWithMetadata.generate(block, gen.receipts(block), false),
Collections.emptyList()),
blockchainQueries.getBlockchain());
}

Expand Down
3 changes: 2 additions & 1 deletion ...est/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/filter/FilterManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,8 @@ private Hash appendBlockToBlockchain() {
new BlockDataGenerator.BlockOptions().setBlockNumber(blockNumber).setParentHash(parentHash);
currentBlock = blockGenerator.block(options);
filterManager.recordBlockEvent(
BlockAddedEvent.createForHeadAdvancement(currentBlock, Collections.emptyList()),
BlockAddedEvent.createForHeadAdvancement(
currentBlock, Collections.emptyList(), Collections.emptyList()),
blockchainQueries.getBlockchain());
return currentBlock.getHash();
}
Expand Down
24 changes: 18 additions & 6 deletions .../test/java/org/hyperledger/besu/ethereum/blockcreation/AbstractMiningCoordinatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,9 @@ public void shouldNotStartWhenBlockAddedAndOutOfSync() {
miningCoordinator.start();

miningCoordinator.onBlockAdded(
BlockAddedEvent.createForHeadAdvancement(BLOCK, Collections.emptyList()), blockchain);
BlockAddedEvent.createForHeadAdvancement(
BLOCK, Collections.emptyList(), Collections.emptyList()),
blockchain);

verifyNoMoreInteractions(minerExecutor, blockMiner);
}
Expand All @@ -118,7 +120,9 @@ public void shouldRestartMiningWhenBlockAddedAndInSync() {
miningCoordinator.start();

miningCoordinator.onBlockAdded(
BlockAddedEvent.createForHeadAdvancement(BLOCK, Collections.emptyList()), blockchain);
BlockAddedEvent.createForHeadAdvancement(
BLOCK, Collections.emptyList(), Collections.emptyList()),
blockchain);

verify(blockMiner).cancel();
verify(minerExecutor, times(2)).startAsyncMining(any(), any(), any());
Expand Down Expand Up @@ -175,7 +179,9 @@ public void shouldNotStartMiningWhenBlockAddedAndInSyncIfMinerNotEnabled() {
miningCoordinator.start();
when(syncState.isInSync()).thenReturn(true);
miningCoordinator.onBlockAdded(
BlockAddedEvent.createForHeadAdvancement(BLOCK, Collections.emptyList()), blockchain);
BlockAddedEvent.createForHeadAdvancement(
BLOCK, Collections.emptyList(), Collections.emptyList()),
blockchain);

verifyNoMoreInteractions(minerExecutor, blockMiner);
}
Expand All @@ -185,7 +191,9 @@ public void shouldNotStartMiningWhenBlockAddedAndInSyncIfMinerNotStarted() {
miningCoordinator.enable();
when(syncState.isInSync()).thenReturn(true);
miningCoordinator.onBlockAdded(
BlockAddedEvent.createForHeadAdvancement(BLOCK, Collections.emptyList()), blockchain);
BlockAddedEvent.createForHeadAdvancement(
BLOCK, Collections.emptyList(), Collections.emptyList()),
blockchain);
verifyNoMoreInteractions(minerExecutor, blockMiner);
}

Expand All @@ -198,7 +206,9 @@ public void shouldNotStartMiningWhenBlockAddedAndInSyncIfStopped() {

when(syncState.isInSync()).thenReturn(true);
miningCoordinator.onBlockAdded(
BlockAddedEvent.createForHeadAdvancement(BLOCK, Collections.emptyList()), blockchain);
BlockAddedEvent.createForHeadAdvancement(
BLOCK, Collections.emptyList(), Collections.emptyList()),
blockchain);

verifyNoMoreInteractions(minerExecutor, blockMiner);
}
Expand All @@ -213,7 +223,9 @@ public void shouldNotStartMiningWhenBlockAddedAndInSyncIfStoppedThenStarted() {

when(syncState.isInSync()).thenReturn(true);
miningCoordinator.onBlockAdded(
BlockAddedEvent.createForHeadAdvancement(BLOCK, Collections.emptyList()), blockchain);
BlockAddedEvent.createForHeadAdvancement(
BLOCK, Collections.emptyList(), Collections.emptyList()),
blockchain);
verifyNoMoreInteractions(minerExecutor, blockMiner);
}

Expand Down
Loading