mirrors/bisq
1
0
Fork 0
mirror of https://github.com/bisq-network/bisq.git synced 2025-02-24 15:10:44 +01:00
Code Issues Projects Releases Packages Wiki Activity

Use Thread for parsing BsqBlocks at BsqLiteNode

Browse source
This commit is contained in:
Manfred Karrer 2017-04-16 23:38:50 -05:00
parent 2955c0efff
commit 27462f0e83
6 changed files with 59 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
core/src/main/java/io/bisq/core/dao/blockchain/BsqBlockchainRequest.java
View file

@ -30,6 +30,7 @@ import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import javax.inject.Inject;
import java.util.List;
import java.util.function.Consumer;
// Used for non blocking access to blockchain data and parsing. Encapsulate thread context, so caller
@ -176,4 +177,40 @@ public class BsqBlockchainRequest {
public void addBlockHandler(Consumer<Block> blockHandler) {
bsqBlockchainService.registerBlockHandler(blockHandler);
}
// BsqLiteNode parse with delivered BsqBlocks. Much faster than requesting via RPC....
void parseBsqBlocks(List<BsqBlock> bsqBlockList,
int genesisBlockHeight,
String genesisTxId,
Consumer<BsqBlock> newBlockHandler,
ResultHandler resultHandler,
Consumer<Throwable> errorHandler) {
ListenableFuture<Void> future = parseBlocksExecutor.submit(() -> {
long startTs = System.currentTimeMillis();
bsqParser.parseBsqBlocks(bsqBlockList,
genesisBlockHeight,
genesisTxId,
newBsqBlock -> {
UserThread.execute(() -> newBlockHandler.accept(newBsqBlock));
});
log.info("parseBlocks took {} ms for {} blocks", System.currentTimeMillis() - startTs, bsqBlockList.size());
return null;
});
Futures.addCallback(future, new FutureCallback<Void>() {
@Override
public void onSuccess(Void ignore) {
UserThread.execute(() -> {
UserThread.execute(resultHandler::handleResult);
});
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.error(throwable.toString());
throwable.printStackTrace();
UserThread.execute(() -> errorHandler.accept(throwable));
}
});
}
}

3
core/src/main/java/io/bisq/core/dao/blockchain/BsqFullNode.java
View file

@ -49,7 +49,6 @@ import java.util.List;
@Slf4j
public class BsqFullNode extends BsqNode {
private final BsqBlockchainRequest bsqBlockchainRequest;
private final JsonExporter jsonExporter;
@Getter
private boolean parseBlockchainComplete;
@ -73,9 +72,9 @@ public class BsqFullNode extends BsqNode {
p2PService,
bsqChainState,
bsqParser,
bsqBlockchainRequest,
persistenceProtoResolver,
storageDir);
this.bsqBlockchainRequest = bsqBlockchainRequest;
this.jsonExporter = jsonExporter;
}

27
core/src/main/java/io/bisq/core/dao/blockchain/BsqLiteNode.java
View file

@ -29,7 +29,6 @@ import io.bisq.common.storage.Storage;
import io.bisq.common.util.Utilities;
import io.bisq.core.app.BisqEnvironment;
import io.bisq.core.dao.blockchain.exceptions.BlockNotConnectingException;
import io.bisq.core.dao.blockchain.exceptions.BsqBlockchainException;
import io.bisq.core.dao.blockchain.p2p.GetBsqBlocksRequest;
import io.bisq.core.dao.blockchain.p2p.GetBsqBlocksResponse;
import io.bisq.core.dao.blockchain.p2p.NewBsqBlockBroadcastMsg;
@ -60,12 +59,14 @@ public class BsqLiteNode extends BsqNode {
P2PService p2PService,
BsqChainState bsqChainState,
BsqParser bsqParser,
BsqBlockchainRequest bsqBlockchainRequest,
PersistenceProtoResolver persistenceProtoResolver,
@Named(Storage.STORAGE_DIR) File storageDir) {
super(bisqEnvironment,
p2PService,
bsqChainState,
bsqParser,
bsqBlockchainRequest,
persistenceProtoResolver,
storageDir);
}
@ -134,16 +135,20 @@ public class BsqLiteNode extends BsqNode {
byte[] bsqBlocksBytes = getBsqBlocksResponse.getBsqBlocksBytes();
List<BsqBlock> bsqBlockList = Utilities.<ArrayList<BsqBlock>>deserialize(bsqBlocksBytes);
log.debug("received msg with {} items", bsqBlockList.size());
try {
bsqParser.parseBsqBlocks(bsqBlockList, getGenesisBlockHeight(), getGenesisTxId(),
this::onNewBsqBlock);
onParseBlockchainComplete(getGenesisBlockHeight(), getGenesisTxId());
} catch (BsqBlockchainException e) {
e.printStackTrace();
} catch (BlockNotConnectingException e) {
e.printStackTrace();
}
bsqBlockchainRequest.parseBsqBlocks(bsqBlockList,
getGenesisBlockHeight(),
getGenesisTxId(),
this::onNewBsqBlock,
() -> {
onParseBlockchainComplete(getGenesisBlockHeight(), getGenesisTxId());
}, throwable -> {
if (throwable instanceof BlockNotConnectingException) {
startReOrgFromLastSnapshot(((BlockNotConnectingException) throwable).getBlock());
} else {
log.error(throwable.toString());
throwable.printStackTrace();
}
});
} else if (parseBlockchainComplete && msg instanceof NewBsqBlockBroadcastMsg) {
NewBsqBlockBroadcastMsg newBsqBlockBroadcastMsg = (NewBsqBlockBroadcastMsg) msg;
byte[] bsqBlockBytes = newBsqBlockBroadcastMsg.getBsqBlockBytes();

4
core/src/main/java/io/bisq/core/dao/blockchain/BsqNode.java
View file

@ -103,6 +103,8 @@ public abstract class BsqNode {
@SuppressWarnings("WeakerAccess")
protected final BsqParser bsqParser;
@SuppressWarnings("WeakerAccess")
protected final BsqBlockchainRequest bsqBlockchainRequest;
@SuppressWarnings("WeakerAccess")
protected final List<BsqChainStateListener> bsqChainStateListeners = new ArrayList<>();
private final Storage<BsqChainState> snapshotBsqChainStateStorage;
@ -123,12 +125,14 @@ public abstract class BsqNode {
P2PService p2PService,
BsqChainState bsqChainState,
BsqParser bsqParser,
BsqBlockchainRequest bsqBlockchainRequest,
PersistenceProtoResolver persistenceProtoResolver,
@Named(Storage.STORAGE_DIR) File storageDir) {
this.p2PService = p2PService;
this.bsqChainState = bsqChainState;
this.bsqParser = bsqParser;
this.bsqBlockchainRequest = bsqBlockchainRequest;
isRegTest = bisqEnvironment.getBitcoinNetwork() == BitcoinNetwork.REGTEST;
snapshotBsqChainStateStorage = new Storage<>(storageDir, persistenceProtoResolver);

4
core/src/main/java/io/bisq/core/dao/blockchain/BsqParser.java
View file

@ -57,8 +57,6 @@ public class BsqParser {
String genesisTxId,
Consumer<BsqBlock> newBlockHandler)
throws BsqBlockchainException, BlockNotConnectingException {
long startTs = System.currentTimeMillis();
// TODO thread
for (BsqBlock bsqBlock : bsqBlocks) {
parseBsqBlock(bsqBlock,
genesisBlockHeight,
@ -66,8 +64,6 @@ public class BsqParser {
bsqChainState.addBlock(bsqBlock);
newBlockHandler.accept(bsqBlock);
}
log.info("parseBlocks took {} ms for {} blocks", System.currentTimeMillis() - startTs, bsqBlocks.size());
}
void parseBsqBlock(BsqBlock bsqBlock,

2
gui/src/main/java/io/bisq/gui/main/portfolio/pendingtrades/PendingTradesViewModel.java
View file

@ -321,7 +321,7 @@ public class PendingTradesViewModel extends ActivatableWithDataModel<PendingTrad
private void onTradeStateChanged(Trade.State tradeState) {
Log.traceCall(tradeState.toString());
log.info("UI tradeState={}, id={}",
log.debug("UI tradeState={}, id={}",
tradeState,
trade != null ? trade.getShortId() : "trade is null");
// TODO what is first valid state for trade?