TransactionManager { public Disposable getTransactionByLoop(Transaction transaction) { Disposable disposable = mDisposableMap.get(transaction.getHash()); return disposable != null ? disposable : Flowable .interval(Constants.Common.TRANSACTION_STATUS_LOOP_TIME, TimeUnit.MILLISECONDS) .map(new Function<Long, Transaction>() { @Override public Transaction apply(Long aLong) throws Exception { Transaction tempTransaction = transaction.clone(); if (System.currentTimeMillis() - transaction.getTimestamp() >= NumberParserUtils.parseLong(AppConfigManager.getInstance().getTimeout())) { tempTransaction.setTxReceiptStatus(TransactionStatus.TIMEOUT.ordinal()); } else { TransactionReceipt transactionReceipt = getTransactionReceipt(tempTransaction.getHash()); tempTransaction.setTxReceiptStatus(transactionReceipt.getStatus()); tempTransaction.setTotalReward(transactionReceipt.getTotalReward()); tempTransaction.setBlockNumber(transactionReceipt.getBlockNumber()); tempTransaction.setTimestamp(transactionReceipt.getTimestamp() == null ? 0 : transactionReceipt.getTimestamp()); tempTransaction.setActualTxCost(TextUtils.isEmpty(transactionReceipt.getActualTxCost()) ? "0" : transactionReceipt.getActualTxCost()); if (tempTransaction.getTxType() == TransactionType.UNDELEGATE) { tempTransaction.setValue(BigDecimalUtil.add(transaction.getUnDelegation(), transactionReceipt.getTotalReward()).toPlainString()); } } return tempTransaction; } }) .takeUntil(new Predicate<Transaction>() { @Override public boolean test(Transaction transaction) throws Exception { return transaction.getTxReceiptStatus() != TransactionStatus.PENDING; } }) .filter(new Predicate<Transaction>() { @Override public boolean test(Transaction transaction) throws Exception { return transaction.getTxReceiptStatus() != TransactionStatus.PENDING; } }) .doOnNext(new Consumer<Transaction>() { @Override public void accept(Transaction transaction) throws Exception { removeTaskByHash(transaction.getHash()); removePendingTransaction(transaction.getFrom()); TransactionDao.insertTransaction(transaction.toTransactionEntity()); if (transaction.getTxReceiptStatus() == TransactionStatus.SUCCESSED || transaction.getTxReceiptStatus() == TransactionStatus.FAILED) { TransactionDao.deleteTransaction(transaction.getHash()); } EventPublisher.getInstance().sendUpdateTransactionEvent(transaction); } }) .toObservable() .subscribeOn(Schedulers.io()) .subscribe(); } private  TransactionManager(); static TransactionManager getInstance(); Wallet getBalanceByAddress(Wallet walletEntity); Disposable removeTaskByHash(String hash); void putPendingTransaction(String from, long timeStamp); long getPendingTransactionTimeStamp(String from); void removePendingTransaction(String from); long getSendTransactionTimeInterval(String from, long currentTime); boolean isAllowSendTransaction(String from, long currentTime); void putTask(String hash, Disposable disposable); void cancelTaskByHash(String hash); Single<RPCTransactionResult> sendContractTransaction(PlatOnContract platOnContract, Credentials credentials, PlatOnFunction platOnFunction, String nonce); Single<RPCTransactionResult> submitTransaction(String sign, String signedMessage, String remark); Single<BigInteger> getNonce(String from); RPCTransactionResult getTransactionResult(String hexValue); String sendContractTransaction(String signedMessage); PlatonSendTransaction sendTransactionReturnPlatonSendTransaction(String signedMessage); String signTransaction(Credentials credentials, String data, String toAddress, BigDecimal amount, BigInteger nonce, BigInteger gasPrice, BigInteger gasLimit); Single<Transaction> sendTransferTransaction(ECKeyPair ecKeyPair, String fromAddress, String toAddress, String walletName, BigDecimal transferAmount, BigDecimal feeAmount, BigInteger gasPrice, BigInteger gasLimit, BigInteger nonce, String remark); Disposable getTransactionByLoop(Transaction transaction);  }
@Test public void getTransactionByLoop(){ Transaction transaction =new Transaction.Builder() .hash("0x7c60553fe3dcdf19f641ee58c5691e49b63c4d7055b92fe1b09b517471f26143") .from("0xa577c0230df2cb329415bfebcb936496ab8ae2e4") .to("0x3d4cee0fb811034dab8ddda086f4448ee3124cc2") .senderWalletName("1111") .value("100000000") .chainId("101") .timestamp(System.currentTimeMillis()) .txReceiptStatus(TransactionStatus.PENDING.ordinal()) .actualTxCost("123580000") .unDelegation("20000000000000") .nodeName("节点02") .nodeId("0x411a6c3640b6cd13799e7d4ed286c95104e3a31fbb05d7ae0004463db648f26e93f7f5848ee9795fb4bbb5f83985afd63f750dc4cf48f53b0e84d26d6834c20c") .build(); Flowable .interval(Constants.Common.TRANSACTION_STATUS_LOOP_TIME, TimeUnit.MILLISECONDS) .map(new Function<Long, Transaction>() { @Override public Transaction apply(Long aLong) throws Exception { Transaction tempTransaction = transaction.clone(); if (System.currentTimeMillis() - transaction.getTimestamp() >= NumberParserUtils.parseLong(AppConfigManager.getInstance().getTimeout())) { tempTransaction.setTxReceiptStatus(TransactionStatus.TIMEOUT.ordinal()); } else { tempTransaction.setTxReceiptStatus(1); } return tempTransaction; } }) .takeUntil(new Predicate<Transaction>() { @Override public boolean test(Transaction transaction) throws Exception { return transaction.getTxReceiptStatus() != TransactionStatus.PENDING; } }) .filter(new Predicate<Transaction>() { @Override public boolean test(Transaction transaction) throws Exception { return transaction.getTxReceiptStatus() != TransactionStatus.PENDING; } }) .doOnNext(new Consumer<Transaction>() { @Override public void accept(Transaction transaction) throws Exception { if (transaction.getTxReceiptStatus() == TransactionStatus.SUCCESSED) { TransactionDao.deleteTransaction(transaction.getHash()); LogUtils.e("getIndividualTransactionByLoop 轮询交易成功" + transaction.toString()); } EventPublisher.getInstance().sendUpdateTransactionEvent(transaction); } }) .toObservable(); }