/**
 * scheduled task to sync Monitor Info per 5s
 * 
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Scheduled(cron = "0/5 * * * * ?")
public void syncMonitorInfo() throws ExecutionException, InterruptedException {
    log.debug("begin sync chain data");
    if (!constants.isMonitorEnabled()) {
        return;
    }
    Long currentTime = System.currentTimeMillis();
    // to do add more group
    for (Map.Entry<Integer, Web3j> entry : web3jMap.entrySet()) {
        Monitor monitor = new Monitor();
        CompletableFuture<BlockNumber> blockHeightFuture =
                entry.getValue().getBlockNumber().sendAsync();
        CompletableFuture<PbftView> pbftViewFuture = entry.getValue().getPbftView().sendAsync();
        CompletableFuture<PendingTxSize> pendingTxSizeFuture =
                entry.getValue().getPendingTxSize().sendAsync();

        monitor.setBlockHeight(blockHeightFuture.get().getBlockNumber());
        monitor.setPbftView(pbftViewFuture.get().getPbftView());
        monitor.setPendingTransactionCount(pendingTxSizeFuture.get().getPendingTxSize());
        monitor.setTimestamp(currentTime);
        monitor.setGroupId(entry.getKey());
        monitorRepository.save(monitor);
        log.debug("insert success =  " + monitor.getId());
    }
}
 

public void run() {
    System.out.println("MyRunnable.run()，" + new Date());

    System.out.println("begin sync chain data");
    Long currentTime = System.currentTimeMillis();
    //to do  add  more group
    Monitor monitor = new Monitor();
    CompletableFuture<BlockNumber> blockHeightFuture = web3j.getBlockNumber().sendAsync();
    CompletableFuture<PbftView> pbftViewFuture = web3j.getPbftView().sendAsync();
    CompletableFuture<PendingTxSize> pendingTxSizeFuture = web3j.getPendingTxSize().sendAsync();
    try {
        monitor.setBlockHeight(blockHeightFuture.get().getBlockNumber());
        monitor.setPbftView(pbftViewFuture.get().getPbftView());
        monitor.setPendingTransactionCount(pendingTxSizeFuture.get().getPendingTxSize());
        monitor.setTimestamp(currentTime);
        monitor.setGroupId(1);
        System.out.println(monitor);
    }catch (ExecutionException | InterruptedException e){
        System.out.println("sync chain data error " + e.getMessage());
    }
    System.out.println("insert success =  " + monitor.getId());

}
 

public static Long getBlockHeight(Web3j web3j, int timeout) throws BrokerException {
    try {
        BlockNumber blockNumber = web3j.getBlockNumber().sendAsync().get(timeout, TimeUnit.MILLISECONDS);
        // Web3sdk's rpc return null in "get".
        if (blockNumber == null) {
            return 0L;
        }
        Long blockHeight = blockNumber.getBlockNumber().longValue();
        log.debug("current block height: {}", blockHeight);
        return blockHeight;
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
        log.error("get block height failed due to InterruptedException|ExecutionException|TimeoutException", e);
        throw new BrokerException(ErrorCode.GET_BLOCK_HEIGHT_ERROR);
    }
}
 

@Override
public BigInteger getBlockNumberCache() {
    if (getLocalBlockNumber().intValue() == 1) {
        try {
            BlockNumber blockNumber = getBlockNumber().sendAsync().get();
            setBlockNumber(blockNumber.getBlockNumber());
        } catch (Exception e) {
            logger.error("Exception: " + e);
        }
    }
    return getLocalBlockNumber().add(new BigInteger(BlockLimit.blockLimit.toString()));
}
 

@Test
public void getBlockNumber() throws IOException, InterruptedException, ExecutionException {
  BlockNumber blockNumber = web3j.getBlockNumber().send();
  assertNotNull(blockNumber.getBlockNumber());
  blockNumber = web3j.getBlockNumber().sendAsync().get();
  assertNotNull(blockNumber.getBlockNumber());
}
 

@SuppressWarnings("unchecked")
void prepareBlockNumberRequest() throws IOException {
    BlockNumber ethBlockNumber = new BlockNumber();
    ethBlockNumber.setResult("0x1");

    Request<?, BlockNumber> ethBlockNumberRequest = mock(Request.class);
    when(ethBlockNumberRequest.send()).thenReturn(ethBlockNumber);
    when(web3j.getBlockNumber()).thenReturn((Request) ethBlockNumberRequest);
    when(web3j.getBlockNumberCache()).thenReturn(new BigInteger("1"));
}
 

private void queryBlockNumber(ChannelHandlerContext ctx) throws JsonProcessingException {

        final String host = ChannelHandlerContextHelper.getPeerHost(ctx);

        String seq = channelService.newSeq();

        BcosMessage bcosMessage = new BcosMessage();
        bcosMessage.setType((short) ChannelMessageType.CHANNEL_RPC_REQUEST.getType());
        bcosMessage.setSeq(seq);
        ChannelEthereumService channelEthereumService = new ChannelEthereumService();
        channelEthereumService.setChannelService(channelService);

        Request<Integer, BlockNumber> request =
                new Request<>(
                        "getBlockNumber",
                        Arrays.asList(channelService.getGroupId()),
                        channelEthereumService,
                        BlockNumber.class);

        bcosMessage.setData(ObjectMapperFactory.getObjectMapper().writeValueAsBytes(request));
        ByteBuf byteBuf = ctx.alloc().buffer();
        bcosMessage.writeHeader(byteBuf);
        bcosMessage.writeExtra(byteBuf);
        ctx.writeAndFlush(byteBuf);

        String content = new String(bcosMessage.getData());
        logger.info(" query block number host: {}, seq: {}, content: {}", host, seq, content);

        channelService
                .getSeq2Callback()
                .put(
                        seq,
                        new BcosResponseCallback() {
                            @Override
                            public void onResponse(BcosResponse response) {
                                try {
                                    BlockNumber blockNumber =
                                            ObjectMapperFactory.getObjectMapper()
                                                    .readValue(
                                                            response.getContent(),
                                                            BlockNumber.class);

                                    SocketChannel socketChannel = (SocketChannel) ctx.channel();
                                    InetSocketAddress socketAddress = socketChannel.remoteAddress();
                                    channelService
                                            .getNodeToBlockNumberMap()
                                            .put(
                                                    socketAddress.getAddress().getHostAddress()
                                                            + socketAddress.getPort(),
                                                    blockNumber.getBlockNumber());

                                    logger.info(
                                            " query blocknumer, host:{}, blockNumber: {} ",
                                            host,
                                            blockNumber.getBlockNumber());
                                } catch (Exception e) {
                                    logger.error(
                                            " query blocknumer failed, host: {}, message: {} ",
                                            host,
                                            e.getMessage());

                                    throw new MessageDecodingException(response.getContent());
                                }
                            }
                        });
    }
 

@Override
public synchronized Request<?, BlockNumber> getBlockNumber() {
    return new Request<>(
            "getBlockNumber", Arrays.asList(groupId), web3jService, BlockNumber.class);
}
 

Request<?, BlockNumber> getBlockNumber();