package cc.mrbird.febs.mall.chain.service;
|
|
import cc.mrbird.febs.common.exception.FebsException;
|
import cc.mrbird.febs.mall.chain.enums.ChainEnum;
|
import cc.mrbird.febs.mall.chain.enums.EthService;
|
import cn.hutool.core.util.ObjectUtil;
|
import io.reactivex.Flowable;
|
import io.reactivex.disposables.Disposable;
|
import io.reactivex.schedulers.Schedulers;
|
import lombok.extern.slf4j.Slf4j;
|
import org.java_websocket.WebSocket;
|
import org.java_websocket.exceptions.WebsocketNotConnectedException;
|
import org.web3j.crypto.Credentials;
|
import org.web3j.protocol.Web3j;
|
import org.web3j.protocol.core.DefaultBlockParameter;
|
import org.web3j.protocol.core.DefaultBlockParameterName;
|
import org.web3j.protocol.core.DefaultBlockParameterNumber;
|
import org.web3j.protocol.core.methods.request.EthFilter;
|
import org.web3j.protocol.http.HttpService;
|
import org.web3j.protocol.websocket.WebSocketClient;
|
import org.web3j.protocol.websocket.WebSocketService;
|
import org.web3j.tx.gas.StaticGasProvider;
|
|
import java.math.BigInteger;
|
import java.net.ConnectException;
|
import java.net.URI;
|
import java.net.URISyntaxException;
|
import java.util.HashMap;
|
import java.util.Map;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.ScheduledExecutorService;
|
import java.util.concurrent.TimeUnit;
|
|
/**
|
* @author
|
* @date 2022-03-23
|
**/
|
@Slf4j
|
public class ChainService {
|
private final static Map<String, ContractChainService> contractMap = new HashMap<>();
|
|
static {
|
for (ChainEnum chain : ChainEnum.values()) {
|
contractMap.put(chain.name(), new EthService(chain.getUrl(), chain.getAddress(), chain.getPrivateKey(), chain.getContractAddress()));
|
}
|
}
|
|
private ChainService() {
|
}
|
|
public final static ChainService INSTANCE = new ChainService();
|
|
public static ContractChainService getInstance(String chainType) {
|
ContractChainService contract = contractMap.get(chainType);
|
if (contract == null) {
|
throw new FebsException("参数错误");
|
}
|
|
return contract;
|
}
|
|
/**
|
* 监听合约事件
|
*
|
* @param startBlock 开始区块
|
*/
|
public static void contractEventListener(BigInteger startBlock, ContractEventService event, String type) {
|
contractEventListener(startBlock, null, event, type);
|
}
|
|
public static void contractEventListener(BigInteger startBlock, BigInteger endBlock, ContractEventService event, String type) {
|
ChainEnum chain = ChainEnum.getValueByName(type);
|
assert chain != null;
|
|
EthUsdtContract contract = contract(chain.getPrivateKey(), chain.getContractAddress(), chain.getUrl());
|
EthFilter filter = getFilter(startBlock, endBlock, chain.getContractAddress());
|
|
Flowable<EthUsdtContract.TransferEventResponse> eventFlowable = contract.transferEventFlowable(filter);
|
eventFlowable.subscribe(e -> {
|
event.compile(e);
|
}, error -> {
|
log.error("合约监听启动报错", error);
|
});
|
}
|
|
public static void wssContractEventListener(BigInteger startBlock, ContractEventService event, String type) {
|
WebSocketService ws = null;
|
WebSocketClient webSocketClient = null;
|
Web3j web3j = null;
|
ChainEnum chain = ChainEnum.getValueByName(type);
|
|
// try {
|
// webSocketClient = new WebSocketClient(new URI("wss://bsc-mainnet.nodereal.io/ws/v1/78074065950e4915aef4f12b6f357d16"));
|
// ws = new WebSocketService(webSocketClient, true);
|
// ws.connect();
|
// } catch (Exception e) {
|
// try {
|
// ws.close();
|
// //TODO
|
// if(ObjectUtil.isEmpty(webSocketClient)){
|
// webSocketClient = new WebSocketClient(new URI("wss://bsc-mainnet.nodereal.io/ws/v1/78074065950e4915aef4f12b6f357d16"));
|
// ws = new WebSocketService(webSocketClient, true);
|
// ws.connect();
|
// }
|
// WebSocket.READYSTATE readyState = webSocketClient.getReadyState();
|
// if((WebSocket.READYSTATE.OPEN != readyState
|
// && WebSocket.READYSTATE.CONNECTING != readyState)){
|
// webSocketClient = new WebSocketClient(new URI("wss://bsc-mainnet.nodereal.io/ws/v1/78074065950e4915aef4f12b6f357d16"));
|
// ws = new WebSocketService(webSocketClient, true);
|
// ws.connect();
|
// }
|
// } catch (Exception connectException) {
|
// connectException.printStackTrace();
|
// }
|
// }
|
|
|
try {
|
webSocketClient = new WebSocketClient(new URI("wss://bsc-mainnet.nodereal.io/ws/v1/78074065950e4915aef4f12b6f357d16"));
|
ws = new WebSocketService(webSocketClient, true);
|
ws.connect();
|
web3j = Web3j.build(ws);
|
|
assert chain != null;
|
try{
|
EthUsdtContract ethUsdtContract = wssContract(chain.getPrivateKey(), chain.getContractAddress(), web3j);
|
EthFilter filter = getFilter(startBlock, startBlock, chain.getContractAddress());
|
Flowable<EthUsdtContract.TransferEventResponse> eventFlowable = ethUsdtContract.transferEventFlowable(filter);
|
Disposable subscribe = eventFlowable.subscribe(event::compile, error -> {
|
log.error("币安监听异常", error);
|
|
});
|
}catch(WebsocketNotConnectedException e){
|
ws.close();
|
webSocketClient = new WebSocketClient(new URI("wss://bsc-mainnet.nodereal.io/ws/v1/78074065950e4915aef4f12b6f357d16"));
|
ws = new WebSocketService(webSocketClient, true);
|
ws.connect();
|
web3j = Web3j.build(ws);
|
assert chain != null;
|
EthUsdtContract ethUsdtContract = wssContract(chain.getPrivateKey(), chain.getContractAddress(), web3j);
|
EthFilter filter = getFilter(startBlock, startBlock, chain.getContractAddress());
|
Flowable<EthUsdtContract.TransferEventResponse> eventFlowable = ethUsdtContract.transferEventFlowable(filter);
|
Disposable subscribe = eventFlowable.subscribe(event::compile, error -> {
|
log.error("币安监听异常", error);
|
|
});
|
}
|
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
|
}
|
|
public static void wssContractEventListener2(BigInteger startBlock, ContractEventService event, String type) {
|
ChainEnum chain = ChainEnum.getValueByName(type);
|
|
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
|
WebSocketService ws = null;
|
try{
|
WebSocketClient webSocketClient = new WebSocketClient(new URI("wss://bsc-mainnet.nodereal.io/ws/v1/fcb1ae31845147dcabb183db57336218"));
|
ws = new WebSocketService(webSocketClient, true);
|
ws.connect();
|
Web3j web3j = Web3j.build(ws);
|
|
EthUsdtContract ethUsdtContract = wssContract(chain.getPrivateKey(), chain.getContractAddress(), web3j);
|
EthFilter filter = getFilter(startBlock, startBlock, chain.getContractAddress());
|
Flowable<EthUsdtContract.TransferEventResponse> eventFlowable = ethUsdtContract.transferEventFlowable(filter);
|
|
// 订阅事件流，将事件编译和处理放到单独的线程中执行
|
Disposable subscribe = eventFlowable.observeOn(Schedulers.from(executor))
|
.subscribe(event::compile, error -> {
|
log.error("币安监听异常", error);
|
error = error.getCause();
|
if (error instanceof WebsocketNotConnectedException) {
|
log.error("WebSocket连接已断开，正在尝试重新连接...");
|
// 异常情况下，递归重试连接和订阅
|
int retries = 3;
|
while (retries-- > 0) {
|
try {
|
Thread.sleep(1000);
|
wssContractEventListener2(startBlock, event, type);
|
return;
|
} catch (Exception ex) {
|
log.error("币安监听异常", ex);
|
}
|
}
|
} else {
|
log.error("币安监听异常", error);
|
}
|
});
|
|
// 使用try-with-resources语句，确保WebSocket连接和Disposable对象能够被及时关闭和释放
|
try{
|
// 等待订阅事件流的线程结束，然后关闭连接和释放资源
|
executor.schedule(ws::close, 1, TimeUnit.SECONDS);
|
executor.awaitTermination(10, TimeUnit.SECONDS);
|
subscribe.dispose();
|
} catch (InterruptedException e) {
|
e.printStackTrace();
|
// 将递归重试连接和订阅改为有限次数的重试，避免出现无限递归的情况
|
int retries = 3;
|
while (retries-- > 0) {
|
try {
|
Thread.sleep(1000);
|
wssContractEventListener2(startBlock, event, type);
|
return;
|
} catch (Exception ex) {
|
log.error("币安监听异常", ex);
|
}
|
}
|
}
|
} catch (Exception e) {
|
log.error("币安监听异常", e);
|
|
// 将递归重试连接和订阅改为有限次数的重试，避免出现无限递归的情况
|
int retries = 3;
|
while (retries-- > 0) {
|
try {
|
Thread.sleep(1000);
|
wssContractEventListener2(startBlock, event, type);
|
return;
|
} catch (Exception ex) {
|
log.error("币安监听异常", ex);
|
}
|
}
|
} finally {
|
executor.shutdownNow();
|
}
|
}
|
|
|
|
public static void wssBaseCoinEventListener(BigInteger startBlock, BaseCoinService event) {
|
WebSocketService ws = null;
|
WebSocketClient webSocketClient = null;
|
Web3j web3j = null;
|
|
try {
|
webSocketClient = new WebSocketClient(new URI("wss://bsc-mainnet.nodereal.io/ws/v1/fcb1ae31845147dcabb183db57336218"));
|
ws = new WebSocketService(webSocketClient, true);
|
ws.connect();
|
web3j = Web3j.build(ws);
|
|
Disposable subscribe = web3j.replayPastAndFutureTransactionsFlowable(new DefaultBlockParameterNumber(startBlock)).subscribe(event::compile, error ->{
|
log.error("监听链上异常", error);
|
});
|
}catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
|
|
private static EthUsdtContract contract(String privateKey, String contractAddress, String url) {
|
Credentials credentials = Credentials.create(privateKey);
|
HttpService httpService = new HttpService(url);
|
// httpService.addHeader("Authorization", "Bearer " + Base64.encode("tfc:tfc123".getBytes()));
|
// httpService.addHeader("Authorization", "Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJhdWQiOiJwdWJsaWMiLCJleHAiOjE2NTk5MzcxOTAsImp0aSI6IjRiMjNkYTVjLWRlZWEtNDYzNi04YjMwLWNmMmZmMjVkM2NlYyIsImlhdCI6MTY1OTkzMzU5MCwiaXNzIjoiQW5rciIsIm5iZiI6MTY1OTkzMzU5MCwic3ViIjoiZmNiNjY0YjItOGEwNC00N2E5LTg3ZjMtNTJhMjE2ODVlMzEzIn0.YfEwvDByU2MGHywsblZpEmKMIbjv4cWYkn5CaFglXY0TSANzd2pCSbIe40yU_R9_nV6xZeE8Uk74jJOdd_QvMpFyUgo-MMNWZP6uiEaYvK_K3tlpk5yzeZq9D4ruWaq8rFKggr-iaRGzu6coRSAOFv2prWll3a7NdEbmkM-y5Y85xYD6g1N-TPIpE_Y-_-WPf3JUavk744kG8YyHhGvAmk2IL0N2xePfC6CHesdJhwvmJJXzr_53dbPwit1y5KljS0iTZz3mGTML2bq4hGaEHbQxeY2fBpZOSm8sPMz-zB9IVJQKzH5-DXlPKz01mJ9XiBJlubfHsN72RdqFD-O2Tw");
|
return EthUsdtContract.load(contractAddress,
|
Web3j.build(httpService),
|
credentials,
|
new StaticGasProvider(BigInteger.valueOf(4500000L), BigInteger.valueOf(200000L)));
|
}
|
|
private static EthUsdtContract wssContract(String privateKey, String contractAddress, Web3j web3j) {
|
Credentials credentials = Credentials.create(privateKey);
|
// httpService.addHeader("Authorization", "Bearer " + Base64.encode("tfc:tfc123".getBytes()));
|
// httpService.addHeader("Authorization", "Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJhdWQiOiJwdWJsaWMiLCJleHAiOjE2NTk5MzcxOTAsImp0aSI6IjRiMjNkYTVjLWRlZWEtNDYzNi04YjMwLWNmMmZmMjVkM2NlYyIsImlhdCI6MTY1OTkzMzU5MCwiaXNzIjoiQW5rciIsIm5iZiI6MTY1OTkzMzU5MCwic3ViIjoiZmNiNjY0YjItOGEwNC00N2E5LTg3ZjMtNTJhMjE2ODVlMzEzIn0.YfEwvDByU2MGHywsblZpEmKMIbjv4cWYkn5CaFglXY0TSANzd2pCSbIe40yU_R9_nV6xZeE8Uk74jJOdd_QvMpFyUgo-MMNWZP6uiEaYvK_K3tlpk5yzeZq9D4ruWaq8rFKggr-iaRGzu6coRSAOFv2prWll3a7NdEbmkM-y5Y85xYD6g1N-TPIpE_Y-_-WPf3JUavk744kG8YyHhGvAmk2IL0N2xePfC6CHesdJhwvmJJXzr_53dbPwit1y5KljS0iTZz3mGTML2bq4hGaEHbQxeY2fBpZOSm8sPMz-zB9IVJQKzH5-DXlPKz01mJ9XiBJlubfHsN72RdqFD-O2Tw");
|
return EthUsdtContract.load(contractAddress,
|
web3j,
|
credentials,
|
new StaticGasProvider(BigInteger.valueOf(4500000L), BigInteger.valueOf(200000L)));
|
}
|
|
private static EthFilter getFilter(BigInteger startBlock, String contractAddress) {
|
return getFilter(startBlock, null, contractAddress);
|
}
|
|
private static EthFilter getFilter(BigInteger startBlock, BigInteger endBlock, String contractAddress) {
|
DefaultBlockParameter startParameterName = null;
|
DefaultBlockParameter endParameterName = null;
|
if (startBlock != null) {
|
startParameterName = new DefaultBlockParameterNumber(startBlock);
|
} else {
|
startParameterName = DefaultBlockParameterName.EARLIEST;
|
}
|
|
if (endBlock != null) {
|
endParameterName = new DefaultBlockParameterNumber(endBlock);
|
} else {
|
endParameterName = DefaultBlockParameterName.LATEST;
|
}
|
|
return new EthFilter(startParameterName, endParameterName, contractAddress);
|
}
|
|
public static void main(String[] args) {
|
// ChainEnum chain = ChainEnum.getValueByName(ChainEnum.BSC_TFC.name());
|
// assert chain != null;
|
//
|
// EthUsdtContract contract = contract(chain.getPrivateKey(), chain.getContractAddress(), chain.getUrl());
|
// EthFilter filter = getFilter(new BigInteger("18097238"), chain.getContractAddress());
|
//
|
// contract.transferEventFlowable(filter).subscribe(e -> {
|
// System.out.println(1);
|
// }, error -> {
|
// log.error("--->", error);
|
// });
|
|
// System.out.println(ChainService.getInstance(ChainEnum.BSC_TFC.name()).totalSupply());
|
}
|
|
}
|
