diff --git a/build.gradle b/build.gradle
index 65e72c0fb73..a03474d6785 100644
--- a/build.gradle
+++ b/build.gradle
@@ -7,6 +7,17 @@ plugins {
ext {
grpcVersion = "1.83.0"
+ // Netty 4.2 split io.netty.handler.codec.protobuf out of netty-codec into its
+ // own artifact. Both :framework and :p2p put the varint32 framing codecs on
+ // their channel pipelines and so must declare it explicitly. Netty itself
+ // arrives transitively through grpc-netty, so this version has to move with
+ // grpcVersion above — keeping it here makes that coupling visible instead of
+ // leaving two literals to drift apart.
+ nettyVersion = "4.2.15.Final"
+ // Shared by :protocol and :p2p, which both generate from .proto files.
+ protobufVersion = "3.25.8"
+ // Shared by :framework, :plugins and :p2p.
+ checkstyleVersion = "8.7"
}
allprojects {
diff --git a/common/build.gradle b/common/build.gradle
index 14d3eb4e637..fbab0385d7c 100644
--- a/common/build.gradle
+++ b/common/build.gradle
@@ -21,23 +21,7 @@ dependencies {
api 'org.aspectj:aspectjrt:1.9.8'
api 'org.aspectj:aspectjweaver:1.9.8'
api 'org.aspectj:aspectjtools:1.9.8'
- api group: 'io.github.tronprotocol', name: 'libp2p', version: '2.2.9',{
- exclude group: 'io.grpc', module: 'grpc-context'
- exclude group: 'io.grpc', module: 'grpc-core'
- exclude group: 'io.grpc', module: 'grpc-netty'
- exclude group: 'com.google.protobuf', module: 'protobuf-java'
- exclude group: 'com.google.protobuf', module: 'protobuf-java-util'
- // https://github.com/dom4j/dom4j/pull/116
- // https://github.com/gradle/gradle/issues/13656
- // https://github.com/dom4j/dom4j/issues/99
- exclude group: 'jaxen', module: 'jaxen'
- exclude group: 'javax.xml.stream', module: 'stax-api'
- exclude group: 'net.java.dev.msv', module: 'xsdlib'
- exclude group: 'pull-parser', module: 'pull-parser'
- exclude group: 'xpp3', module: 'xpp3'
- exclude group: 'org.bouncycastle', module: 'bcprov-jdk18on'
- exclude group: 'org.bouncycastle', module: 'bcutil-jdk18on'
- }
+ api project(":p2p")
api project(":protocol")
api project(":platform")
}
diff --git a/framework/build.gradle b/framework/build.gradle
index 8255fc30d18..8edd6714f95 100644
--- a/framework/build.gradle
+++ b/framework/build.gradle
@@ -11,9 +11,7 @@ apply plugin: 'checkstyle'
mainClassName = 'org.tron.program.FullNode'
-def versions = [
- checkstyle: '8.7',
-]
+
@@ -40,7 +38,7 @@ dependencies {
// end local libraries
implementation group: 'com.beust', name: 'jcommander', version: '1.78'
implementation group: 'io.dropwizard.metrics', name: 'metrics-core', version: '3.1.2'
- implementation('io.netty:netty-codec-protobuf:4.2.15.Final') {
+ implementation("io.netty:netty-codec-protobuf:${rootProject.nettyVersion}") {
exclude group: 'com.google.protobuf'
exclude group: 'com.google.protobuf.nano'
}
@@ -61,17 +59,28 @@ dependencies {
testImplementation group: 'org.springframework', name: 'spring-test', version: "${springVersion}"
testImplementation group: 'javax.portlet', name: 'portlet-api', version: '3.0.1'
+
implementation group: 'org.zeromq', name: 'jeromq', version: '0.5.3'
api project(":chainbase")
api project(":protocol")
api project(":actuator")
api project(":consensus")
+ // org.tron.p2p is used directly in 17 files under src/main/java (org.tron.core.net
+ // and org.tron.core.config.args). It currently arrives only transitively, three
+ // hops away, because :common exposes it via `api project(":p2p")` -- and :common
+ // has to, since CommonParameter publishes P2pConfig/PublishConfig in its own API.
+ // Declare the direct use here as well, so framework keeps compiling if that
+ // transitive chain is ever narrowed. api, not implementation: framework does
+ // re-export p2p types -- P2pEventHandlerImpl extends org.tron.p2p.P2pEventHandler,
+ // HelloMessage.getFrom() returns org.tron.p2p.discover.Node, PeerManager takes
+ // org.tron.p2p.connection.Channel, Args.loadDnsPublishConfig returns PublishConfig.
+ api project(":p2p")
}
check.dependsOn 'lint'
checkstyle {
- toolVersion = "${versions.checkstyle}"
+ toolVersion = "${rootProject.checkstyleVersion}"
configFile = file("config/checkstyle/checkStyleAll.xml")
maxWarnings = 0
}
@@ -187,8 +196,14 @@ def binaryRelease(taskName, jarName, mainClass) {
}
// explicit_dependency
+ // :p2p is included because :common now exposes it via `api project(":p2p")`,
+ // so p2p-1.0.0.jar is on runtimeClasspath and gets zipped into the fat jar.
+ // Without it Gradle reports an implicit_dependency and disables execution
+ // optimizations, and a parallel build could assemble FullNode.jar before
+ // :p2p:jar has been written.
dependsOn (project(':actuator').jar, project(':consensus').jar, project(':chainbase').jar,
- project(':crypto').jar, project(':common').jar, project(':protocol').jar, project(':platform').jar)
+ project(':crypto').jar, project(':common').jar, project(':protocol').jar,
+ project(':platform').jar, project(':p2p').jar)
from {
configurations.runtimeClasspath.collect {
diff --git a/gradle/verification-metadata.xml b/gradle/verification-metadata.xml
index 6a3e641d5d6..d603780e242 100644
--- a/gradle/verification-metadata.xml
+++ b/gradle/verification-metadata.xml
@@ -448,6 +448,14 @@
+
+
+
+
+
+
+
+
@@ -463,6 +471,11 @@
+
+
+
+
+
@@ -1985,6 +1998,14 @@
+
+
+
+
+
+
+
+
diff --git a/p2p/.gitignore b/p2p/.gitignore
new file mode 100644
index 00000000000..ebb224e762b
--- /dev/null
+++ b/p2p/.gitignore
@@ -0,0 +1,2 @@
+# protobuf generated code (rebuilt by ./gradlew :p2p:generateProto)
+src/main/java/org/tron/p2p/protos/
diff --git a/p2p/README.md b/p2p/README.md
new file mode 100644
index 00000000000..a94c4587bd1
--- /dev/null
+++ b/p2p/README.md
@@ -0,0 +1,453 @@
+# p2p
+
+Peer discovery, connection management and DNS-based node lists for java-tron.
+
+> **Vendored from [tronprotocol/libp2p](https://github.com/tronprotocol/libp2p)
+> v2.2.9.** It lives here as the `:p2p` Gradle module and is consumed as a
+> project dependency, not as a published artifact — `:common` exposes it via
+> `api project(":p2p")`. Upstream's own README follows, edited where
+> internalizing changed the facts; treat this file as the module's docs rather
+> than as a mirror of upstream.
+
+## Running it on its own
+
+`./gradlew :p2p:build` produces two jars:
+
+| | |
+|---|---|
+| `p2p-1.0.0.jar` | thin — what java-tron depends on, no `Main-Class` |
+| `p2p-standalone.jar` | all dependencies bundled, `Main-Class` set to `StartApp` |
+
+Use the standalone one to drive the module without starting java-tron:
+
+```bash
+./gradlew :p2p:buildStandaloneJar
+java -jar p2p/build/libs/p2p-standalone.jar --help
+```
+
+`-PbinaryRelease=false` skips building it, matching `:framework` and `:plugins`.
+
+This module can run on its own, or be used as a library.
+
+# 1. Run independently
+
+command of start a p2p node:
+
+```bash
+$ java -jar p2p/build/libs/p2p-standalone.jar [options]
+```
+
+available cli options:
+
+```bash
+usage: available p2p discovery cli options:
+ -a,--active-nodes active node(s),
+ ip:port[,ip:port[...]]
+ -d,--discover enable p2p discover, 0/1, default 1
+ -h,--help print help message
+ -M,--max-connection max connection number, int, default
+ 50
+ -m,--min-connection min connection number, int, default 8
+ -ma,--min-active-connection min active connection number, int,
+ default 2
+ -p,--port UDP & TCP port, int, default 18888
+ -s,--seed-nodes seed node(s), required,
+ ip:port[,ip:port[...]]
+ -t,--trust-ips trust ip(s), ip[,ip[...]]
+ -v,--version p2p version, int, default 1
+
+available dns read cli options:
+ -u,--url-schemes dns url(s) to get nodes, url format
+ tree://{pubkey}@{domain}, url[,url[...]]
+
+available dns publish cli options:
+ --access-key-id access key id of aws or aliyun api,
+ required, string
+ --access-key-secret access key secret of aws or aliyun api,
+ required, string
+ --aliyun-dns-endpoint if server-type is aliyun, it's endpoint
+ of aws dns server, required, string
+ --aws-region if server-type is aws, it's region of
+ aws api, such as "eu-south-1", required,
+ string
+ --change-threshold change threshold of add and delete to
+ publish, optional, should be > 0 and <
+ 1.0, default 0.1
+ --dns-private dns private key used to publish,
+ required, hex string of length 64
+ --domain dns domain to publish nodes, required,
+ string
+ --host-zone-id if server-type is aws, it's host zone id
+ of aws's domain, optional, string
+ --known-urls known dns urls to publish, url format
+ tree://{pubkey}@{domain}, optional,
+ url[,url[...]]
+ --max-merge-size max merge size to merge node to a leaf
+ node in dns tree, optional, should be
+ [1~5], default 5
+ -publish,--publish enable dns publish
+ --server-type dns server to publish, required, only
+ aws or aliyun is support
+ --static-nodes static nodes to publish, if exist then
+ nodes from kad will be ignored,
+ optional, ip:port[,ip:port[...]]
+```
+
+For details please
+check [StartApp](src/main/java/org/tron/p2p/example/StartApp.java)
+.
+
+## 1.1 Construct a p2p network
+
+For example
+Node A, starts with default configuration parameters. Let's say its IP is 127.0.0.1
+
+```bash
+$ java -jar p2p/build/libs/p2p-standalone.jar
+```
+
+Node B, start with seed nodes(127.0.0.1:18888). Let's say its IP is 127.0.0.2
+
+```bash
+$ java -jar p2p/build/libs/p2p-standalone.jar -s 127.0.0.1:18888
+```
+
+Node C, start with with seed nodes(127.0.0.1:18888). Let's say its IP is 127.0.0.3
+
+```bash
+$ java -jar p2p/build/libs/p2p-standalone.jar -s 127.0.0.1:18888
+```
+
+After the three nodes are successfully started, the usual situation is that node B can discover node
+C (or node C can discover B), and the three of them can establish a TCP connection with each other.
+
+## 1.2 Publish our nodes on one domain
+
+Libp2p support publish nodes on dns domain. Before publishing, you must enable p2p
+discover. Node lists can be deployed to any DNS provider such as CloudFlare DNS, dnsimple, Amazon
+Route 53, Aliyun Cloud using their respective client libraries. But we only support Amazon Route 53
+and Aliyun Cloud.
+You can see more detail on https://eips.ethereum.org/EIPS/eip-1459, we implement this eip, but have
+some difference in data structure.
+
+### 1.2.1 Acquire your apikey from Amazon Route 53 or Aliyun Cloud
+
+* Amazon Route 53 include: AWS Access Key ID、AWS Access Key Secret、Route53 Zone ID、AWS Region, get more info
+* Aliyun Cloud include: accessKeyId、accessKeySecret、endpoint, get more info
+
+### 1.2.2 Publish nodes
+
+Suppose you have a domain example.org hosted by Amazon Route 53, you can publish your nodes automatically
+like this:
+
+```bash
+java -jar p2p/build/libs/p2p-standalone.jar -p 18888 -v 201910292 -d 1 -s 127.0.0.1:18888 \
+-publish \
+--dns-private b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291 \
+--server-type aws \
+--access-key-id \
+--access-key-secret \
+--aws-region us-east-1 \
+--host-zone-id \
+--domain nodes.example.org
+```
+
+This program will do following periodically:
+
+* get nodes from p2p discover service and construct a tree using these nodes
+* collect txt records from dns domain with API
+* compare tree with the txt records
+* submit changes to dns domain with API if necessary.
+
+We can get tree's url from log:
+
+```
+tree://APFGGTFOBVE2ZNAB3CSMNNX6RRK3ODIRLP2AA5U4YFAA6MSYZUYTQ@nodes.example.org
+```
+
+The compressed public Key APFGGTFOBVE2ZNAB3CSMNNX6RRK3ODIRLP2AA5U4YFAA6MSYZUYTQ is responsed to
+above dns-private key.
+
+### 1.2.3 Verify your dns txt records
+
+You can query dns record by following command and check if a TXT type record exists:
+
+```bash
+dig nodes.example.org TXT
+```
+
+At last we can release the tree's url on anywhere later, such as github. So others can download this
+tree to get nodes dynamically.
+
+# 2. Use as a dependency
+
+## 2.1 Core classes
+
+* [P2pService](src/main/java/org/tron/p2p/P2pService.java)
+ is the entry class of p2p service and provides the startup interface of p2p service and the main
+ interfaces provided by p2p module.
+* [P2pConfig](src/main/java/org/tron/p2p/P2pConfig.java)
+ defines all the configurations of the p2p module, such as the listening port, the maximum number
+ of connections, etc.
+* [P2pEventHandler](src/main/java/org/tron/p2p/P2pEventHandler.java)
+ is the abstract class for p2p event handler.
+* [Channel](src/main/java/org/tron/p2p/connection/Channel.java)
+ is an implementation of the TCP connection channel in the p2p module. The new connection channel
+ is obtained through the `P2pEventHandler.onConnect` method.
+
+## 2.2 Interface
+
+* `P2pService.start`
+ - @param: p2pConfig P2pConfig
+ - @return: void
+ - desc: the startup interface of p2p service
+* `P2pService.close`
+ - @param:
+ - @return: void
+ - desc: the close interface of p2p service
+* `P2pService.register`
+ - @param: p2PEventHandler P2pEventHandler
+ - @return: void
+ - desc: register p2p event handler
+* `P2pService.connect`
+ - @param: address InetSocketAddress
+ - @return: void
+ - desc: connect to a node with a socket address
+* `P2pService.getAllNodes`
+ - @param:
+ - @return: List
+ - desc: get all the nodes
+* `P2pService.getTableNodes`
+ - @param:
+ - @return: List
+ - desc: get all the nodes that in the hash table
+* `P2pService.getConnectableNodes`
+ - @param:
+ - @return: List
+ - desc: get all the nodes that can be connected
+* `P2pService.getP2pStats()`
+ - @param:
+ - @return: void
+ - desc: get statistics information of p2p service
+* `Channel.send`
+ - @param: data byte[]
+ - @return: void
+ - desc: send messages to the peer node through the channel
+* `Channel.close`
+ - @param:
+ - @return: void
+ - desc: the close interface of channel
+
+## 2.3 Steps for usage
+
+1. Config p2p discover parameters
+2. (optional) Config dns parameters
+3. Implement P2pEventHandler and register p2p event handler
+4. Start p2p service
+5. Use Channel's send and close interfaces as needed
+6. Use P2pService's interfaces as needed
+
+### 2.3.1 Config discover parameters
+
+New p2p config instance
+
+```bash
+P2pConfig config = new P2pConfig();
+```
+
+Set p2p networkId (also called p2p version)
+
+```bash
+config.setNetworkId(11111);
+```
+
+Set TCP and UDP listen port
+
+```bash
+config.setPort(18888);
+```
+
+Turn node discovery on or off
+
+```bash
+config.setDiscoverEnable(true);
+```
+
+Set discover seed nodes
+
+```bash
+List seedNodeList = new ArrayList<>();
+seedNodeList.add(new InetSocketAddress("13.124.62.58", 18888));
+seedNodeList.add(new InetSocketAddress("2600:1f13:908:1b00:e1fd:5a84:251c:a32a", 18888));
+seedNodeList.add(new InetSocketAddress("[2600:1f13:908:1b00:e1fd:5a84:251c:1234]", 18888));
+seedNodeList.add(new InetSocketAddress("127.0.0.4", 18888));
+config.setSeedNodes(seedNodeList);
+```
+
+Set active nodes
+```bash
+List activeNodeList = new ArrayList<>();
+activeNodeList.add(new InetSocketAddress("127.0.0.2", 18888));
+activeNodeList.add(new InetSocketAddress("127.0.0.3", 18888));
+config.setActiveNodes(activeNodeList);
+```
+
+Set trust ips
+
+```bash
+List trustNodeList = new ArrayList<>();
+trustNodeList.add((new InetSocketAddress("127.0.0.2", 18888)).getAddress());
+config.setTrustNodes(trustNodeList);
+```
+
+Set the minimum number of connections
+
+```bash
+config.setMinConnections(8);
+```
+
+Set the minimum number of actively established connections
+
+```bash
+config.setMinActiveConnections(2);
+```
+
+Set the maximum number of connections
+
+```bash
+config.setMaxConnections(30);
+```
+
+Set the maximum number of connections with the same IP
+
+```bash
+config.setMaxConnectionsWithSameIp(2);
+```
+
+### 2.3.2 (optional) Config dns parameters if needed
+Suppose these scenes:
+* you don't want to config one or many fixed seed nodes in mobile app such as wallet, because nodes may be out of service but you cannot update the app timely
+* you don't known any seed node but you still want to establish tcp connection
+
+You can config a dns tree regardless of whether discovery service is enabled or not. Assume you have a tree url of Tron's nile or shasta or mainnet nodes that publish on github like:
+```azure
+tree://APFGGTFOBVE2ZNAB3CSMNNX6RRK3ODIRLP2AA5U4YFAA6MSYZUYTQ@nodes.example.org
+```
+You can config the parameters like that:
+```bash
+config.setDiscoverEnable(false);
+String[] urls = new String[] {"tree://APFGGTFOBVE2ZNAB3CSMNNX6RRK3ODIRLP2AA5U4YFAA6MSYZUYTQ@nodes.example.org"};
+config.setTreeUrls(Arrays.asList(urls));
+```
+After that, the module will download the nodes from nile.nftderby1.net periodically.
+
+### 2.3.3 TCP Handler
+
+Implement definition message
+
+```bash
+public class TestMessage {
+ protected MessageTypes type;
+ protected byte[] data;
+ public TestMessage(byte[] data) {
+ this.type = MessageTypes.TEST;
+ this.data = data;
+ }
+
+}
+
+public enum MessageTypes {
+
+ FIRST((byte)0x00),
+
+ TEST((byte)0x01),
+
+ LAST((byte)0x8f);
+
+ private final byte type;
+
+ MessageTypes(byte type) {
+ this.type = type;
+ }
+
+ public byte getType() {
+ return type;
+ }
+
+ private static final Map map = new HashMap<>();
+
+ static {
+ for (MessageTypes value : values()) {
+ map.put(value.type, value);
+ }
+ }
+
+ public static MessageTypes fromByte(byte type) {
+ return map.get(type);
+ }
+ }
+```
+
+Inheritance implements the P2pEventHandler class.
+
+* `onConnect` is called back after the TCP connection is established.
+* `onDisconnect` is called back after the TCP connection is closed.
+* `onMessage` is called back after receiving a message on the channel. Note that `data[0]` is the
+ message type.
+
+```bash
+public class MyP2pEventHandler extends P2pEventHandler {
+
+ public MyP2pEventHandler() {
+ this.typeSet = new HashSet<>();
+ this.typeSet.add(MessageTypes.TEST.getType());
+ }
+
+ @Override
+ public void onConnect(Channel channel) {
+ channels.put(channel.getInetSocketAddress(), channel);
+ }
+
+ @Override
+ public void onDisconnect(Channel channel) {
+ channels.remove(channel.getInetSocketAddress());
+ }
+
+ @Override
+ public void onMessage(Channel channel, byte[] data) {
+ byte type = data[0];
+ byte[] messageData = ArrayUtils.subarray(data, 1, data.length);
+ switch (MessageTypes.fromByte(type)) {
+ case TEST:
+ TestMessage message = new TestMessage(messageData);
+ // process TestMessage
+ break;
+ default:
+ // todo
+ }
+ }
+}
+```
+
+### 2.3.4 Start p2p service
+
+Start p2p service with P2pConfig and P2pEventHandler
+
+```bash
+P2pService p2pService = new P2pService();
+MyP2pEventHandler myP2pEventHandler = new MyP2pEventHandler();
+try {
+ p2pService.register(myP2pEventHandler);
+} catch (P2pException e) {
+ // todo process exception
+}
+p2pService.start(config);
+```
+
+For details please
+The former `ImportUsing`, `DnsExample1` and `DnsExample2` reference classes have
+been replaced by
+[ExampleUsageTest](src/test/java/org/tron/p2p/example/ExampleUsageTest.java),
+which asserts the same configuration shapes instead of only compiling them.
+
+
diff --git a/p2p/build.gradle b/p2p/build.gradle
new file mode 100644
index 00000000000..55bd5ec8ec5
--- /dev/null
+++ b/p2p/build.gradle
@@ -0,0 +1,212 @@
+apply plugin: 'com.google.protobuf'
+apply plugin: 'checkstyle'
+
+// Unit tests live in src/test/java/, alongside the code they cover.
+
+checkstyle {
+ toolVersion = "${rootProject.checkstyleVersion}"
+ configFile = file("${rootDir}/config/checkstyle/checkStyleAll.xml")
+ maxWarnings = 0
+}
+
+checkstyleMain {
+ source = 'src/main/java'
+ exclude '**/protos/**'
+}
+
+checkstyleTest {
+ source = 'src/test/java'
+}
+
+// The plain jar stays thin -- it is what :common depends on. Declaring
+// Main-Class on it would be a trap: `java -jar p2p-1.0.0.jar` would resolve the
+// entry point and then die on the first dependency it touches
+// (NoClassDefFoundError: org/apache/commons/cli/ParseException). The runnable
+// artifact is p2p-standalone.jar below.
+//
+// Same shape as :framework's FullNode.jar and :plugins' Toolkit.jar, including
+// the -PbinaryRelease=false opt-out and the signature/SPI exclusions.
+def releaseBinary = hasProperty('binaryRelease') ? getProperty('binaryRelease') : 'true'
+if (releaseBinary == 'true') {
+ artifacts {
+ archives(tasks.create('buildStandaloneJar', Jar) {
+ baseName = 'p2p-standalone'
+ version = null
+ from(sourceSets.main.output) {
+ include '/**'
+ }
+ from {
+ configurations.runtimeClasspath.collect {
+ it.isDirectory() ? it : zipTree(it)
+ }
+ }
+ duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+ // Bouncy Castle ships signed jars; merged signatures fail verification.
+ exclude 'META-INF/*.SF'
+ exclude 'META-INF/*.DSA'
+ exclude 'META-INF/*.RSA'
+ // dnsjava's resolver SPI, see HADOOP-19288
+ exclude 'META-INF/services/java.net.spi.InetAddressResolverProvider'
+ manifest {
+ attributes 'Main-Class': 'org.tron.p2p.example.StartApp'
+ }
+ })
+ }
+}
+
+// .proto files sit in src/main/proto, the protobuf plugin's default, so no
+// sourceSet override is needed. The previous layout, src/main/protos with an
+// explicit srcDir, builds fine but IDEA's protobuf plugin does not read the
+// Gradle override -- it resolves imports against the default path -- so
+// `import "Discover.proto"` and every type it brings in showed as unresolved
+// in the editor.
+//
+// Generated sources still land in src/main/java/org/tron/p2p/protos via
+// generatedFilesBaseDir below, and are gitignored.
+
+// These exclusions used to live on the `libp2p` dependency in common/build.gradle.
+// Internalizing the module moves the same transitive dom4j tail (pulled in by the
+// Aliyun/Route53 SDKs) here, so the exclusions move with it — dropping them would
+// silently re-admit artifacts the project has excluded since:
+// https://github.com/dom4j/dom4j/pull/116
+// https://github.com/gradle/gradle/issues/13656
+// https://github.com/dom4j/dom4j/issues/99
+configurations.configureEach {
+ exclude group: 'jaxen', module: 'jaxen'
+ exclude group: 'javax.xml.stream', module: 'stax-api'
+ exclude group: 'net.java.dev.msv', module: 'msv'
+ exclude group: 'net.java.dev.msv', module: 'xsdlib'
+ exclude group: 'relaxngDatatype', module: 'relaxngDatatype'
+ exclude group: 'pull-parser', module: 'pull-parser'
+ exclude group: 'xpp3', module: 'xpp3'
+}
+
+dependencies {
+ // protobuf & grpc (implementation scope: not leaked to consumers)
+ implementation "com.google.protobuf:protobuf-java:${rootProject.protobufVersion}"
+ implementation "com.google.protobuf:protobuf-java-util:${rootProject.protobufVersion}"
+ // grpc-netty provides Netty transitively, which p2p uses for TCP/UDP transport.
+ // grpc itself is not used (p2p protos define only messages, no services).
+ // Track the project's grpc version rather than pinning libp2p's, so p2p
+ // cannot drift from the Netty that the rest of java-tron resolves.
+ implementation "io.grpc:grpc-netty:${rootProject.grpcVersion}"
+ // Netty 4.2 split io.netty.handler.codec.protobuf out of netty-codec into its
+ // own artifact, so the varint32 framing codecs p2p puts on every channel
+ // pipeline no longer arrive transitively. framework/build.gradle declares the
+ // same dependency for the same reason; both track rootProject.nettyVersion.
+ implementation("io.netty:netty-codec-protobuf:${rootProject.nettyVersion}") {
+ exclude group: 'com.google.protobuf'
+ exclude group: 'com.google.protobuf.nano'
+ }
+
+ // p2p-specific dependencies
+ implementation 'org.xerial.snappy:snappy-java:1.1.10.5'
+ // Matches the bcprov-jdk18on version root build.gradle gives every
+ // subproject; bcpkix is not provided there, so declare it explicitly.
+ implementation 'org.bouncycastle:bcpkix-jdk18on:1.84'
+ implementation 'dnsjava:dnsjava:3.6.2'
+ implementation 'commons-cli:commons-cli:1.5.0'
+ implementation('software.amazon.awssdk:route53:2.18.41') {
+ exclude group: 'io.netty', module: 'netty-codec-http2'
+ exclude group: 'io.netty', module: 'netty-codec-http'
+ exclude group: 'io.netty', module: 'netty-common'
+ exclude group: 'io.netty', module: 'netty-buffer'
+ exclude group: 'io.netty', module: 'netty-transport'
+ exclude group: 'io.netty', module: 'netty-codec'
+ exclude group: 'io.netty', module: 'netty-handler'
+ exclude group: 'io.netty', module: 'netty-resolver'
+ exclude group: 'io.netty', module: 'netty-transport-classes-epoll'
+ exclude group: 'io.netty', module: 'netty-transport-native-unix-common'
+ exclude group: 'software.amazon.awssdk', module: 'netty-nio-client'
+ }
+ implementation('com.aliyun:alidns20150109:3.0.1') {
+ exclude group: 'org.bouncycastle', module: 'bcprov-jdk15on'
+ exclude group: 'org.bouncycastle', module: 'bcpkix-jdk15on'
+ exclude group: 'pull-parser', module: 'pull-parser'
+ exclude group: 'xpp3', module: 'xpp3'
+ }
+
+ // commons-lang3: root provides 3.4 as 'implementation' (not on compile classpath),
+ // so it has to be re-declared here for p2p to compile.
+ //
+ // The version must stay 3.18.0. The external libp2p:2.2.9 POM declared
+ // commons-lang3:3.18.0 at runtime scope, which won conflict resolution against the
+ // root's 3.4 and put 3.18.0 on :framework:runtimeClasspath. Internalizing the module
+ // removes that requester, so pinning 3.4 here would silently downgrade the shipped
+ // node to a 2015 release and reintroduce CVE-2025-48924
+ // (ClassUtils.getAbbreviatedName uncontrolled recursion, fixed in 3.18.0).
+ // Verify with:
+ // ./gradlew :framework:dependencyInsight --configuration runtimeClasspath \
+ // --dependency commons-lang3
+ //
+ // The source still uses the 3.0-compatible `new BasicThreadFactory.Builder()` rather
+ // than `builder()`, so the module keeps compiling if this ever has to move back down.
+ implementation 'org.apache.commons:commons-lang3:3.18.0'
+
+ // provided by root build.gradle for all subprojects:
+ // slf4j-api, logback, bcprov-jdk18on, lombok, junit, mockito
+}
+
+protobuf {
+ generatedFilesBaseDir = "$projectDir/src"
+ protoc {
+ artifact = "com.google.protobuf:protoc:${rootProject.protobufVersion}"
+ }
+ generateProtoTasks {
+ all().each { task ->
+ task.builtins {
+ java { outputSubDir = "java" }
+ }
+ }
+ }
+}
+
+clean.doFirst {
+ delete "src/main/java/org/tron/p2p/protos"
+}
+
+processResources.dependsOn(generateProto)
+
+// The module reports its own coverage now that it has a test sourceSet. CI
+// collects **/build/reports/jacoco/test/jacocoTestReport.xml across every
+// module, so this is picked up without any wiring in :framework.
+jacocoTestReport {
+ dependsOn test
+ reports {
+ xml.required = true
+ html.required = false
+ }
+ // Generated protobuf code, matching the checkstyle exclusion above.
+ //
+ // org/tron/p2p/example holds StartApp, the standalone CLI entry point for
+ // driving this module without java-tron. It is argument parsing, option
+ // declarations and a main() that starts services and blocks -- not module
+ // logic, and not code the node runs. It also sat in the checkstyle- and
+ // coverage-exempt `example` sourceSet until this PR, so excluding it keeps
+ // the measured surface the same rather than hiding newly counted logic.
+ // The parsing helpers it does own are covered by StartAppArgsTest.
+ classDirectories.setFrom(files(classDirectories.files.collect {
+ fileTree(dir: it, excludes: ['**/protos/**', '**/example/**'])
+ }))
+
+ // :framework's own tests -- org.tron.core.net and friends -- execute a good
+ // deal of this module's code. That coverage is real, and while p2p's classes
+ // hung off :framework:jacocoTestReport it was counted. Reading framework's
+ // exec data here keeps counting it now that the classes live in this
+ // module's report instead. Without this the coverage gate sees a ~0.3 point
+ // drop for code that is still being exercised, just no longer measured.
+ //
+ // The fileTree is empty when :framework:test has not run, so :p2p:build on
+ // its own still works -- it just reports this module's tests alone.
+ executionData.from(fileTree("${rootProject.projectDir}/framework/build/jacoco")
+ .include('**/*.exec'))
+ mustRunAfter ':framework:test'
+}
+
+// Prints the module's runtime classpath, so StartApp can be launched standalone
+// against a thin jar. See README.md.
+tasks.register('printRuntimeClasspath') {
+ doLast {
+ println sourceSets.main.runtimeClasspath.asPath
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/P2pConfig.java b/p2p/src/main/java/org/tron/p2p/P2pConfig.java
new file mode 100644
index 00000000000..9a3aef6b9df
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/P2pConfig.java
@@ -0,0 +1,37 @@
+package org.tron.p2p;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import lombok.Data;
+import org.tron.p2p.dns.update.PublishConfig;
+import org.tron.p2p.utils.NetUtil;
+
+@Data
+public class P2pConfig {
+
+ private List seedNodes = new CopyOnWriteArrayList<>();
+ private List activeNodes = new CopyOnWriteArrayList<>();
+ private List trustNodes = new CopyOnWriteArrayList<>();
+ private byte[] nodeID = NetUtil.getNodeId();
+ private String ip = NetUtil.getExternalIpV4();
+ private String lanIp = NetUtil.getLanIP();
+ private String ipv6 = NetUtil.getExternalIpV6();
+ private int port = 18888;
+ private int networkId = 1;
+ private int minConnections = 8;
+ private int maxConnections = 50;
+ private int minActiveConnections = 2;
+ private int maxConnectionsWithSameIp = 2;
+ private boolean discoverEnable = true;
+ private boolean disconnectionPolicyEnable = false;
+ private boolean nodeDetectEnable = false;
+
+ //dns read config
+ private List treeUrls = new ArrayList<>();
+
+ //dns publish config
+ private PublishConfig publishConfig = new PublishConfig();
+}
diff --git a/p2p/src/main/java/org/tron/p2p/P2pEventHandler.java b/p2p/src/main/java/org/tron/p2p/P2pEventHandler.java
new file mode 100644
index 00000000000..7ca3f235049
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/P2pEventHandler.java
@@ -0,0 +1,20 @@
+package org.tron.p2p;
+
+import java.util.Set;
+import lombok.Getter;
+import org.tron.p2p.connection.Channel;
+
+public abstract class P2pEventHandler {
+
+ @Getter
+ protected Set messageTypes;
+
+ public void onConnect(Channel channel) {
+ }
+
+ public void onDisconnect(Channel channel) {
+ }
+
+ public void onMessage(Channel channel, byte[] data) {
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/P2pService.java b/p2p/src/main/java/org/tron/p2p/P2pService.java
new file mode 100644
index 00000000000..5e0b05e56c8
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/P2pService.java
@@ -0,0 +1,90 @@
+package org.tron.p2p;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.connection.ChannelManager;
+import org.tron.p2p.discover.Node;
+import org.tron.p2p.discover.NodeManager;
+import org.tron.p2p.dns.DnsManager;
+import org.tron.p2p.exception.P2pException;
+import org.tron.p2p.stats.P2pStats;
+import org.tron.p2p.stats.StatsManager;
+
+@Slf4j(topic = "net")
+public class P2pService {
+
+ private StatsManager statsManager = new StatsManager();
+ private volatile boolean isShutdown = false;
+
+ public void start(P2pConfig p2pConfig) {
+ Parameter.p2pConfig = p2pConfig;
+ NodeManager.init();
+ ChannelManager.init();
+ DnsManager.init();
+ logger.info("P2p service started");
+
+ Runtime.getRuntime().addShutdownHook(new Thread(this::close));
+ }
+
+ public void close() {
+ if (isShutdown) {
+ return;
+ }
+ isShutdown = true;
+ DnsManager.close();
+ NodeManager.close();
+ ChannelManager.close();
+ logger.info("P2p service closed");
+ }
+
+ public void register(P2pEventHandler p2PEventHandler) throws P2pException {
+ Parameter.addP2pEventHandle(p2PEventHandler);
+ }
+
+ @Deprecated
+ public void connect(InetSocketAddress address) {
+ ChannelManager.connect(address);
+ }
+
+ public ChannelFuture connect(Node node, ChannelFutureListener future) {
+ return ChannelManager.connect(node, future);
+ }
+
+ public P2pStats getP2pStats() {
+ return statsManager.getP2pStats();
+ }
+
+ public List getTableNodes() {
+ return NodeManager.getTableNodes();
+ }
+
+ public List getConnectableNodes() {
+ Set nodes = new HashSet<>();
+ nodes.addAll(NodeManager.getConnectableNodes());
+ nodes.addAll(DnsManager.getDnsNodes());
+ return new ArrayList<>(nodes);
+ }
+
+ public List getAllNodes() {
+ Set nodes = new HashSet<>();
+ nodes.addAll(NodeManager.getAllNodes());
+ nodes.addAll(DnsManager.getDnsNodes());
+ return new ArrayList<>(nodes);
+ }
+
+ public void updateNodeId(Channel channel, String nodeId) {
+ ChannelManager.updateNodeId(channel, nodeId);
+ }
+
+ public int getVersion() {
+ return Parameter.version;
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/base/Constant.java b/p2p/src/main/java/org/tron/p2p/base/Constant.java
new file mode 100644
index 00000000000..b87cfa287f5
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/base/Constant.java
@@ -0,0 +1,16 @@
+package org.tron.p2p.base;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class Constant {
+
+ public static final int NODE_ID_LEN = 64;
+ public static final List ipV4Urls = Arrays.asList(
+ "http://checkip.amazonaws.com", "https://ifconfig.me/ip", "https://4.ipw.cn/");
+ public static final List ipV6Urls = Arrays.asList(
+ "https://v6.ident.me", "http://6.ipw.cn/", "https://api6.ipify.org",
+ "https://ipv6.icanhazip.com");
+ public static final String ipV4Hex = "00000000"; //32 bit
+ public static final String ipV6Hex = "00000000000000000000000000000000"; //128 bit
+}
diff --git a/p2p/src/main/java/org/tron/p2p/base/Parameter.java b/p2p/src/main/java/org/tron/p2p/base/Parameter.java
new file mode 100644
index 00000000000..a53f74b9a33
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/base/Parameter.java
@@ -0,0 +1,74 @@
+package org.tron.p2p.base;
+
+import com.google.protobuf.ByteString;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.Data;
+import org.apache.commons.lang3.StringUtils;
+import org.tron.p2p.P2pConfig;
+import org.tron.p2p.P2pEventHandler;
+import org.tron.p2p.exception.P2pException;
+import org.tron.p2p.exception.P2pException.TypeEnum;
+import org.tron.p2p.protos.Discover;
+import org.tron.p2p.utils.ByteArray;
+
+@Data
+public class Parameter {
+
+ public static int version = 1;
+
+ public static final int TCP_NETTY_WORK_THREAD_NUM = 0;
+
+ public static final int UDP_NETTY_WORK_THREAD_NUM = 1;
+
+ public static final int CONN_MAX_QUEUE_SIZE = 10;
+
+ public static final int NODE_CONNECTION_TIMEOUT = 2000;
+
+ public static final int KEEP_ALIVE_TIMEOUT = 20_000;
+
+ public static final int PING_TIMEOUT = 20_000;
+
+ public static final int NETWORK_TIME_DIFF = 1000;
+
+ public static final long DEFAULT_BAN_TIME = 60_000;
+
+ public static final int MAX_MESSAGE_LENGTH = 5 * 1024 * 1024;
+
+ public static volatile P2pConfig p2pConfig;
+
+ public static volatile List handlerList = new ArrayList<>();
+
+ public static volatile Map handlerMap = new HashMap<>();
+
+ public static void addP2pEventHandle(P2pEventHandler p2PEventHandler) throws P2pException {
+ if (p2PEventHandler.getMessageTypes() != null) {
+ for (Byte type : p2PEventHandler.getMessageTypes()) {
+ if (handlerMap.get(type) != null) {
+ throw new P2pException(TypeEnum.TYPE_ALREADY_REGISTERED, "type:" + type);
+ }
+ }
+ for (Byte type : p2PEventHandler.getMessageTypes()) {
+ handlerMap.put(type, p2PEventHandler);
+ }
+ }
+ handlerList.add(p2PEventHandler);
+ }
+
+ public static Discover.Endpoint getHomeNode() {
+ Discover.Endpoint.Builder builder = Discover.Endpoint.newBuilder()
+ .setNodeId(ByteString.copyFrom(Parameter.p2pConfig.getNodeID()))
+ .setPort(Parameter.p2pConfig.getPort());
+ if (StringUtils.isNotEmpty(Parameter.p2pConfig.getIp())) {
+ builder.setAddress(ByteString.copyFrom(
+ ByteArray.fromString(Parameter.p2pConfig.getIp())));
+ }
+ if (StringUtils.isNotEmpty(Parameter.p2pConfig.getIpv6())) {
+ builder.setAddressIpv6(ByteString.copyFrom(
+ ByteArray.fromString(Parameter.p2pConfig.getIpv6())));
+ }
+ return builder.build();
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/Channel.java b/p2p/src/main/java/org/tron/p2p/connection/Channel.java
new file mode 100644
index 00000000000..811904b7d27
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/Channel.java
@@ -0,0 +1,204 @@
+package org.tron.p2p.connection;
+
+import com.google.common.base.Throwables;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.CorruptedFrameException;
+import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
+import io.netty.handler.timeout.ReadTimeoutException;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.business.upgrade.UpgradeController;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.handshake.HelloMessage;
+import org.tron.p2p.connection.socket.MessageHandler;
+import org.tron.p2p.connection.socket.P2pProtobufVarint32FrameDecoder;
+import org.tron.p2p.discover.Node;
+import org.tron.p2p.exception.P2pException;
+import org.tron.p2p.stats.TrafficStats;
+import org.tron.p2p.utils.ByteArray;
+
+@Slf4j(topic = "net")
+public class Channel {
+
+ public volatile boolean waitForPong = false;
+ public volatile long pingSent = System.currentTimeMillis();
+
+ @Getter
+ private HelloMessage helloMessage;
+ @Getter
+ private Node node;
+ @Getter
+ private int version;
+ @Getter
+ private ChannelHandlerContext ctx;
+ @Getter
+ private InetSocketAddress inetSocketAddress;
+ @Getter
+ private InetAddress inetAddress;
+ @Getter
+ private volatile long disconnectTime;
+ @Getter
+ @Setter
+ private volatile boolean isDisconnect = false;
+ @Getter
+ @Setter
+ private long lastSendTime = System.currentTimeMillis();
+ @Getter
+ private final long startTime = System.currentTimeMillis();
+ @Getter
+ private boolean isActive = false;
+ @Getter
+ private boolean isTrustPeer;
+ @Getter
+ @Setter
+ private volatile boolean finishHandshake;
+ @Getter
+ @Setter
+ private String nodeId;
+ @Setter
+ @Getter
+ private boolean discoveryMode;
+ @Getter
+ private long avgLatency;
+ private long count;
+
+ public void init(ChannelPipeline pipeline, String nodeId, boolean discoveryMode) {
+ this.discoveryMode = discoveryMode;
+ this.nodeId = nodeId;
+ this.isActive = StringUtils.isNotEmpty(nodeId);
+ MessageHandler messageHandler = new MessageHandler(this);
+ pipeline.addLast("readTimeoutHandler", new ReadTimeoutHandler(60, TimeUnit.SECONDS));
+ pipeline.addLast(TrafficStats.tcp);
+ pipeline.addLast("protoPrepend", new ProtobufVarint32LengthFieldPrepender());
+ pipeline.addLast("protoDecode", new P2pProtobufVarint32FrameDecoder(this));
+ pipeline.addLast("messageHandler", messageHandler);
+ }
+
+ public void processException(Throwable throwable) {
+ Throwable baseThrowable = throwable;
+ try {
+ baseThrowable = Throwables.getRootCause(baseThrowable);
+ } catch (IllegalArgumentException e) {
+ baseThrowable = e.getCause();
+ logger.warn("Loop in causal chain detected");
+ }
+ SocketAddress address = ctx.channel().remoteAddress();
+ if (throwable instanceof ReadTimeoutException
+ || throwable instanceof IOException
+ || throwable instanceof CorruptedFrameException) {
+ logger.warn("Close peer {}, reason: {}", address, throwable.getMessage());
+ } else if (baseThrowable instanceof P2pException) {
+ logger.warn("Close peer {}, type: ({}), info: {}",
+ address, ((P2pException) baseThrowable).getType(), baseThrowable.getMessage());
+ } else {
+ logger.error("Close peer {}, exception caught", address, throwable);
+ }
+ close();
+ }
+
+ public void setHelloMessage(HelloMessage helloMessage) {
+ this.helloMessage = helloMessage;
+ this.node = helloMessage.getFrom();
+ this.nodeId = node.getHexId(); //update node id from handshake
+ this.version = helloMessage.getVersion();
+ }
+
+ public void setChannelHandlerContext(ChannelHandlerContext ctx) {
+ this.ctx = ctx;
+ this.inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
+ this.inetAddress = inetSocketAddress.getAddress();
+ this.isTrustPeer = Parameter.p2pConfig.getTrustNodes().contains(inetAddress);
+ }
+
+ public void close(long banTime) {
+ this.isDisconnect = true;
+ this.disconnectTime = System.currentTimeMillis();
+ ChannelManager.banNode(this.inetAddress, banTime);
+ ctx.close();
+ }
+
+ public void close() {
+ close(Parameter.DEFAULT_BAN_TIME);
+ }
+
+ public void send(Message message) {
+ if (message.needToLog()) {
+ logger.info("Send message to channel {}, {}", inetSocketAddress, message);
+ } else {
+ logger.debug("Send message to channel {}, {}", inetSocketAddress, message);
+ }
+ send(message.getSendData());
+ }
+
+ public void send(byte[] data) {
+ try {
+ byte type = data[0];
+ if (isDisconnect) {
+ logger.warn("Send to {} failed as channel has closed, message-type:{} ",
+ ctx.channel().remoteAddress(), type);
+ return;
+ }
+
+ if (finishHandshake) {
+ data = UpgradeController.codeSendData(version, data);
+ }
+
+ ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
+ ctx.writeAndFlush(byteBuf).addListener((ChannelFutureListener) future -> {
+ if (!future.isSuccess() && !isDisconnect) {
+ logger.warn("Send to {} failed, message-type:{}, cause:{}",
+ ctx.channel().remoteAddress(), ByteArray.byte2int(type),
+ future.cause().getMessage());
+ }
+ });
+ setLastSendTime(System.currentTimeMillis());
+ } catch (Exception e) {
+ logger.warn("Send message to {} failed, {}", inetSocketAddress, e.getMessage());
+ ctx.channel().close();
+ }
+ }
+
+ public void updateAvgLatency(long latency) {
+ long total = this.avgLatency * this.count;
+ this.count++;
+ this.avgLatency = (total + latency) / this.count;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Channel channel = (Channel) o;
+ return Objects.equals(inetSocketAddress, channel.inetSocketAddress);
+ }
+
+ @Override
+ public int hashCode() {
+ return inetSocketAddress.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s | %s", inetSocketAddress,
+ StringUtils.isEmpty(nodeId) ? "" : nodeId);
+ }
+
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/ChannelManager.java b/p2p/src/main/java/org/tron/p2p/connection/ChannelManager.java
new file mode 100644
index 00000000000..d32fe2f1a3a
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/ChannelManager.java
@@ -0,0 +1,307 @@
+package org.tron.p2p.connection;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.bouncycastle.util.encoders.Hex;
+import org.tron.p2p.P2pEventHandler;
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.business.detect.NodeDetectService;
+import org.tron.p2p.connection.business.handshake.DisconnectCode;
+import org.tron.p2p.connection.business.handshake.HandshakeService;
+import org.tron.p2p.connection.business.keepalive.KeepAliveService;
+import org.tron.p2p.connection.business.pool.ConnPoolService;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.MessageType;
+import org.tron.p2p.connection.message.base.P2pDisconnectMessage;
+import org.tron.p2p.connection.socket.PeerClient;
+import org.tron.p2p.connection.socket.PeerServer;
+import org.tron.p2p.discover.Node;
+import org.tron.p2p.exception.P2pException;
+import org.tron.p2p.exception.P2pException.TypeEnum;
+import org.tron.p2p.protos.Connect.DisconnectReason;
+import org.tron.p2p.utils.ByteArray;
+import org.tron.p2p.utils.NetUtil;
+
+@Slf4j(topic = "net")
+public class ChannelManager {
+
+ @Getter
+ private static NodeDetectService nodeDetectService;
+
+ private static PeerServer peerServer;
+
+ @Getter
+ private static PeerClient peerClient;
+
+ @Getter
+ private static ConnPoolService connPoolService;
+
+ private static KeepAliveService keepAliveService;
+
+ @Getter
+ private static HandshakeService handshakeService;
+
+ @Getter
+ private static final Map channels = new ConcurrentHashMap<>();
+
+ @Getter
+ private static final Cache bannedNodes = CacheBuilder
+ .newBuilder().maximumSize(2000).build(); //ban timestamp
+
+ private static boolean isInit = false;
+ public static volatile boolean isShutdown = false;
+
+ public static void init() {
+ isInit = true;
+ peerServer = new PeerServer();
+ peerClient = new PeerClient();
+ keepAliveService = new KeepAliveService();
+ connPoolService = new ConnPoolService();
+ handshakeService = new HandshakeService();
+ nodeDetectService = new NodeDetectService();
+ peerServer.init();
+ peerClient.init();
+ keepAliveService.init();
+ connPoolService.init(peerClient);
+ nodeDetectService.init(peerClient);
+ }
+
+ public static void connect(InetSocketAddress address) {
+ peerClient.connect(address.getAddress().getHostAddress(), address.getPort(),
+ ByteArray.toHexString(NetUtil.getNodeId()));
+ }
+
+ public static ChannelFuture connect(Node node, ChannelFutureListener future) {
+ return peerClient.connect(node, future);
+ }
+
+ public static void notifyDisconnect(Channel channel) {
+ if (channel.getInetSocketAddress() == null) {
+ logger.warn("Notify Disconnect peer has no address.");
+ return;
+ }
+ channels.remove(channel.getInetSocketAddress());
+ Parameter.handlerList.forEach(h -> h.onDisconnect(channel));
+ InetAddress inetAddress = channel.getInetAddress();
+ if (inetAddress != null) {
+ banNode(inetAddress, Parameter.DEFAULT_BAN_TIME);
+ }
+ }
+
+ public static int getConnectionNum(InetAddress inetAddress) {
+ int cnt = 0;
+ for (Channel channel : channels.values()) {
+ if (channel.getInetAddress().equals(inetAddress)) {
+ cnt++;
+ }
+ }
+ return cnt;
+ }
+
+ public static synchronized DisconnectCode processPeer(Channel channel) {
+
+ if (!channel.isActive() && !channel.isTrustPeer()) {
+ InetAddress inetAddress = channel.getInetAddress();
+ if (bannedNodes.getIfPresent(inetAddress) != null
+ && bannedNodes.getIfPresent(inetAddress) > System.currentTimeMillis()) {
+ logger.info("Peer {} recently disconnected", channel);
+ return DisconnectCode.TIME_BANNED;
+ }
+
+ if (channels.size() >= Parameter.p2pConfig.getMaxConnections()) {
+ logger.info("Too many peers, disconnected with {}", channel);
+ return DisconnectCode.TOO_MANY_PEERS;
+ }
+
+ int num = getConnectionNum(channel.getInetAddress());
+ if (num >= Parameter.p2pConfig.getMaxConnectionsWithSameIp()) {
+ logger.info("Max connection with same ip {}", channel);
+ return DisconnectCode.MAX_CONNECTION_WITH_SAME_IP;
+ }
+ }
+
+ if (StringUtils.isNotEmpty(channel.getNodeId())) {
+ for (Channel c : channels.values()) {
+ if (channel.getNodeId().equals(c.getNodeId())) {
+ if (c.getStartTime() > channel.getStartTime()) {
+ c.close();
+ } else {
+ logger.info("Duplicate peer {}, exist peer {}", channel, c);
+ return DisconnectCode.DUPLICATE_PEER;
+ }
+ }
+ }
+ }
+
+ channels.put(channel.getInetSocketAddress(), channel);
+
+ logger.info("Add peer {}, total channels: {}", channel.getInetSocketAddress(), channels.size());
+ return DisconnectCode.NORMAL;
+ }
+
+ public static DisconnectReason getDisconnectReason(DisconnectCode code) {
+ DisconnectReason disconnectReason;
+ switch (code) {
+ case DIFFERENT_VERSION:
+ disconnectReason = DisconnectReason.DIFFERENT_VERSION;
+ break;
+ case TIME_BANNED:
+ disconnectReason = DisconnectReason.RECENT_DISCONNECT;
+ break;
+ case DUPLICATE_PEER:
+ disconnectReason = DisconnectReason.DUPLICATE_PEER;
+ break;
+ case TOO_MANY_PEERS:
+ disconnectReason = DisconnectReason.TOO_MANY_PEERS;
+ break;
+ case MAX_CONNECTION_WITH_SAME_IP:
+ disconnectReason = DisconnectReason.TOO_MANY_PEERS_WITH_SAME_IP;
+ break;
+ default: {
+ disconnectReason = DisconnectReason.UNKNOWN;
+ }
+ }
+ return disconnectReason;
+ }
+
+ public static void logDisconnectReason(Channel channel, DisconnectReason reason) {
+ logger.info("Try to close channel: {}, reason: {}", channel.getInetSocketAddress(),
+ reason.name());
+ }
+
+ public static void banNode(InetAddress inetAddress, Long banTime) {
+ long now = System.currentTimeMillis();
+ if (bannedNodes.getIfPresent(inetAddress) == null
+ || bannedNodes.getIfPresent(inetAddress) < now) {
+ bannedNodes.put(inetAddress, now + banTime);
+ }
+ }
+
+ public static void close() {
+ if (!isInit || isShutdown) {
+ return;
+ }
+ isShutdown = true;
+ connPoolService.close();
+ keepAliveService.close();
+ peerServer.close();
+ peerClient.close();
+ nodeDetectService.close();
+ }
+
+
+ public static void processMessage(Channel channel, byte[] data) throws P2pException {
+ if (data == null || data.length == 0) {
+ throw new P2pException(TypeEnum.EMPTY_MESSAGE, "");
+ }
+ if (data[0] >= 0) {
+ handMessage(channel, data);
+ return;
+ }
+
+ Message message = Message.parse(data);
+
+ if (message.needToLog()) {
+ logger.info("Receive message from channel: {}, {}", channel.getInetSocketAddress(), message);
+ } else {
+ logger.debug("Receive message from channel {}, {}", channel.getInetSocketAddress(), message);
+ }
+
+ if (channel.isDiscoveryMode() && message.getType() != MessageType.STATUS) {
+ logger.debug("Discovery channel {} received unexpected message {}, close it",
+ channel.getInetSocketAddress(), message.getType());
+ channel.close();
+ return;
+ }
+
+ switch (message.getType()) {
+ case KEEP_ALIVE_PING:
+ case KEEP_ALIVE_PONG:
+ keepAliveService.processMessage(channel, message);
+ break;
+ case HANDSHAKE_HELLO:
+ handshakeService.processMessage(channel, message);
+ break;
+ case STATUS:
+ nodeDetectService.processMessage(channel, message);
+ break;
+ case DISCONNECT:
+ channel.close();
+ break;
+ default:
+ throw new P2pException(P2pException.TypeEnum.NO_SUCH_MESSAGE, "type:" + data[0]);
+ }
+ }
+
+ private static void handMessage(Channel channel, byte[] data) throws P2pException {
+ P2pEventHandler handler = Parameter.handlerMap.get(data[0]);
+ if (handler == null) {
+ throw new P2pException(P2pException.TypeEnum.NO_SUCH_MESSAGE, "type:" + data[0]);
+ }
+ if (channel.isDiscoveryMode()) {
+ channel.send(new P2pDisconnectMessage(DisconnectReason.DISCOVER_MODE));
+ channel.getCtx().close();
+ return;
+ }
+
+ if (!channel.isFinishHandshake()) {
+ channel.setFinishHandshake(true);
+ DisconnectCode code = processPeer(channel);
+ if (!DisconnectCode.NORMAL.equals(code)) {
+ DisconnectReason disconnectReason = getDisconnectReason(code);
+ channel.send(new P2pDisconnectMessage(disconnectReason));
+ channel.getCtx().close();
+ return;
+ }
+ Parameter.handlerList.forEach(h -> h.onConnect(channel));
+ }
+
+ handler.onMessage(channel, data);
+ }
+
+ public static synchronized void updateNodeId(Channel channel, String nodeId) {
+ channel.setNodeId(nodeId);
+ if (nodeId.equals(Hex.toHexString(Parameter.p2pConfig.getNodeID()))) {
+ logger.warn("Channel {} is myself", channel.getInetSocketAddress());
+ channel.send(new P2pDisconnectMessage(DisconnectReason.DUPLICATE_PEER));
+ channel.close();
+ return;
+ }
+
+ List list = new ArrayList<>();
+ channels.values().forEach(c -> {
+ if (nodeId.equals(c.getNodeId())) {
+ list.add(c);
+ }
+ });
+ if (list.size() <= 1) {
+ return;
+ }
+ Channel c1 = list.get(0);
+ Channel c2 = list.get(1);
+ if (c1.getStartTime() > c2.getStartTime()) {
+ logger.info("Close channel {}, other channel {} is earlier", c1, c2);
+ c1.send(new P2pDisconnectMessage(DisconnectReason.DUPLICATE_PEER));
+ c1.close();
+ } else {
+ logger.info("Close channel {}, other channel {} is earlier", c2, c1);
+ c2.send(new P2pDisconnectMessage(DisconnectReason.DUPLICATE_PEER));
+ c2.close();
+ }
+ }
+
+ public static void triggerConnect(InetSocketAddress address) {
+ connPoolService.triggerConnect(address);
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/MessageProcess.java b/p2p/src/main/java/org/tron/p2p/connection/business/MessageProcess.java
new file mode 100644
index 00000000000..cf731e23398
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/MessageProcess.java
@@ -0,0 +1,8 @@
+package org.tron.p2p.connection.business;
+
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.connection.message.Message;
+
+public interface MessageProcess {
+ void processMessage(Channel channel, Message message);
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/detect/NodeDetectService.java b/p2p/src/main/java/org/tron/p2p/connection/business/detect/NodeDetectService.java
new file mode 100644
index 00000000000..3e642c04f4e
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/detect/NodeDetectService.java
@@ -0,0 +1,229 @@
+package org.tron.p2p.connection.business.detect;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.connection.business.MessageProcess;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.detect.StatusMessage;
+import org.tron.p2p.connection.socket.PeerClient;
+import org.tron.p2p.discover.Node;
+import org.tron.p2p.discover.NodeManager;
+
+@Slf4j(topic = "net")
+public class NodeDetectService implements MessageProcess {
+
+ private PeerClient peerClient;
+
+ private Map nodeStatMap = new ConcurrentHashMap<>();
+
+ @Getter
+ private static final Cache badNodesCache = CacheBuilder
+ .newBuilder().maximumSize(5000).expireAfterWrite(1, TimeUnit.HOURS).build();
+
+ private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
+ new BasicThreadFactory.Builder().namingPattern("nodeDetectService").build());
+
+ private final long NODE_DETECT_THRESHOLD = 5 * 60 * 1000;
+
+ private final long NODE_DETECT_MIN_THRESHOLD = 30 * 1000;
+
+ private final long NODE_DETECT_TIMEOUT = 2 * 1000;
+
+ private final int MAX_NODE_SLOW_DETECT = 3;
+
+ private final int MAX_NODE_NORMAL_DETECT = 10;
+
+ private final int MAX_NODE_FAST_DETECT = 100;
+
+ private final int MAX_NODES = 300;
+
+ private final int MIN_NODES = 200;
+
+
+ public void init(PeerClient peerClient) {
+ if (!Parameter.p2pConfig.isNodeDetectEnable()) {
+ return;
+ }
+ this.peerClient = peerClient;
+ executor.scheduleWithFixedDelay(() -> {
+ try {
+ work();
+ } catch (Exception t) {
+ logger.warn("Exception in node detect worker, {}", t.getMessage());
+ }
+ }, 1, 5, TimeUnit.SECONDS);
+ }
+
+ public void close() {
+ executor.shutdown();
+ }
+
+ public void work() {
+ trimNodeMap();
+ if (nodeStatMap.size() < MIN_NODES) {
+ loadNodes();
+ }
+
+ List nodeStats = getSortedNodeStats();
+ if (nodeStats.size() == 0) {
+ return;
+ }
+
+ NodeStat nodeStat = nodeStats.get(0);
+ if (nodeStat.getLastDetectTime() > System.currentTimeMillis() - NODE_DETECT_MIN_THRESHOLD) {
+ return;
+ }
+
+ int n = MAX_NODE_NORMAL_DETECT;
+ if (nodeStat.getLastDetectTime() > System.currentTimeMillis() - NODE_DETECT_THRESHOLD) {
+ n = MAX_NODE_SLOW_DETECT;
+ }
+
+ n = StrictMath.min(n, nodeStats.size());
+
+ for (int i = 0; i < n; i++) {
+ detect(nodeStats.get(i));
+ }
+ }
+
+ public void trimNodeMap() {
+ long now = System.currentTimeMillis();
+ nodeStatMap.forEach((k, v) -> {
+ if (!v.finishDetect() && v.getLastDetectTime() < now - NODE_DETECT_TIMEOUT) {
+ nodeStatMap.remove(k);
+ badNodesCache.put(k.getAddress(), System.currentTimeMillis());
+ }
+ });
+ }
+
+ private void loadNodes() {
+ int size = nodeStatMap.size();
+ int count = 0;
+ List nodes = NodeManager.getConnectableNodes();
+ for (Node node : nodes) {
+ InetSocketAddress socketAddress = node.getPreferInetSocketAddress();
+ if (socketAddress != null
+ && !nodeStatMap.containsKey(socketAddress)
+ && badNodesCache.getIfPresent(socketAddress.getAddress()) == null) {
+ NodeStat nodeStat = new NodeStat(node);
+ nodeStatMap.put(socketAddress, nodeStat);
+ detect(nodeStat);
+ count++;
+ if (count >= MAX_NODE_FAST_DETECT || count + size >= MAX_NODES) {
+ break;
+ }
+ }
+ }
+ }
+
+ private void detect(NodeStat stat) {
+ try {
+ stat.setTotalCount(stat.getTotalCount() + 1);
+ setLastDetectTime(stat);
+ peerClient.connectAsync(stat.getNode(), true);
+ } catch (Exception e) {
+ logger.warn("Detect node {} failed, {}",
+ stat.getNode().getPreferInetSocketAddress(), e.getMessage());
+ nodeStatMap.remove(stat.getSocketAddress());
+ }
+ }
+
+ public synchronized void processMessage(Channel channel, Message message) {
+ StatusMessage statusMessage = (StatusMessage) message;
+
+ if (!channel.isActive()) {
+ channel.setDiscoveryMode(true);
+ channel.send(new StatusMessage());
+ channel.getCtx().close();
+ return;
+ }
+
+ InetSocketAddress socketAddress = channel.getInetSocketAddress();
+ NodeStat nodeStat = nodeStatMap.get(socketAddress);
+ if (nodeStat == null) {
+ return;
+ }
+
+ long cost = System.currentTimeMillis() - nodeStat.getLastDetectTime();
+ if (cost > NODE_DETECT_TIMEOUT
+ || statusMessage.getRemainConnections() == 0) {
+ badNodesCache.put(socketAddress.getAddress(), cost);
+ nodeStatMap.remove(socketAddress);
+ }
+
+ nodeStat.setLastSuccessDetectTime(nodeStat.getLastDetectTime());
+ setStatusMessage(nodeStat, statusMessage);
+
+ channel.getCtx().close();
+ }
+
+ public void notifyDisconnect(Channel channel) {
+
+ if (!channel.isActive()) {
+ return;
+ }
+
+ InetSocketAddress socketAddress = channel.getInetSocketAddress();
+ if (socketAddress == null) {
+ return;
+ }
+
+ NodeStat nodeStat = nodeStatMap.get(socketAddress);
+ if (nodeStat == null) {
+ return;
+ }
+
+ if (nodeStat.getLastDetectTime() != nodeStat.getLastSuccessDetectTime()) {
+ badNodesCache.put(socketAddress.getAddress(), System.currentTimeMillis());
+ nodeStatMap.remove(socketAddress);
+ }
+ }
+
+ private synchronized List getSortedNodeStats() {
+ List nodeStats = new ArrayList<>(nodeStatMap.values());
+ nodeStats.sort(Comparator.comparingLong(o -> o.getLastDetectTime()));
+ return nodeStats;
+ }
+
+ private synchronized void setLastDetectTime(NodeStat nodeStat) {
+ nodeStat.setLastDetectTime(System.currentTimeMillis());
+ }
+
+ private synchronized void setStatusMessage(NodeStat nodeStat, StatusMessage message) {
+ nodeStat.setStatusMessage(message);
+ }
+
+ public synchronized List getConnectableNodes() {
+ List stats = new ArrayList<>();
+ List nodes = new ArrayList<>();
+ nodeStatMap.values().forEach(stat -> {
+ if (stat.getStatusMessage() != null) {
+ stats.add(stat);
+ }
+ });
+
+ if (stats.isEmpty()) {
+ return nodes;
+ }
+
+ stats.sort(Comparator.comparingInt(o -> -o.getStatusMessage().getRemainConnections()));
+ stats.forEach(stat -> nodes.add(stat.getNode()));
+ return nodes;
+ }
+
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/detect/NodeStat.java b/p2p/src/main/java/org/tron/p2p/connection/business/detect/NodeStat.java
new file mode 100644
index 00000000000..395df70e314
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/detect/NodeStat.java
@@ -0,0 +1,25 @@
+package org.tron.p2p.connection.business.detect;
+
+import java.net.InetSocketAddress;
+import lombok.Data;
+import org.tron.p2p.connection.message.detect.StatusMessage;
+import org.tron.p2p.discover.Node;
+
+@Data
+public class NodeStat {
+ private int totalCount;
+ private long lastDetectTime;
+ private long lastSuccessDetectTime;
+ private StatusMessage statusMessage;
+ private Node node;
+ private InetSocketAddress socketAddress;
+
+ public NodeStat(Node node) {
+ this.node = node;
+ this.socketAddress = node.getPreferInetSocketAddress();
+ }
+
+ public boolean finishDetect() {
+ return this.lastDetectTime == this.lastSuccessDetectTime;
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/handshake/DisconnectCode.java b/p2p/src/main/java/org/tron/p2p/connection/business/handshake/DisconnectCode.java
new file mode 100644
index 00000000000..fc4c9224988
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/handshake/DisconnectCode.java
@@ -0,0 +1,30 @@
+package org.tron.p2p.connection.business.handshake;
+
+public enum DisconnectCode {
+ NORMAL(0),
+ TOO_MANY_PEERS(1),
+ DIFFERENT_VERSION(2),
+ TIME_BANNED(3),
+ DUPLICATE_PEER(4),
+ MAX_CONNECTION_WITH_SAME_IP(5),
+ UNKNOWN(256);
+
+ private final Integer value;
+
+ DisconnectCode(Integer value) {
+ this.value = value;
+ }
+
+ public Integer getValue() {
+ return value;
+ }
+
+ public static DisconnectCode forNumber(int code) {
+ for (DisconnectCode disconnectCode : values()) {
+ if (disconnectCode.value == code) {
+ return disconnectCode;
+ }
+ }
+ return UNKNOWN;
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/handshake/HandshakeService.java b/p2p/src/main/java/org/tron/p2p/connection/business/handshake/HandshakeService.java
new file mode 100644
index 00000000000..38ba25c22ed
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/handshake/HandshakeService.java
@@ -0,0 +1,90 @@
+package org.tron.p2p.connection.business.handshake;
+
+import static org.tron.p2p.connection.ChannelManager.getDisconnectReason;
+import static org.tron.p2p.connection.ChannelManager.logDisconnectReason;
+
+import lombok.extern.slf4j.Slf4j;
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.connection.ChannelManager;
+import org.tron.p2p.connection.business.MessageProcess;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.base.P2pDisconnectMessage;
+import org.tron.p2p.connection.message.handshake.HelloMessage;
+import org.tron.p2p.protos.Connect.DisconnectReason;
+
+@Slf4j(topic = "net")
+public class HandshakeService implements MessageProcess {
+
+ private final int networkId = Parameter.p2pConfig.getNetworkId();
+
+ public void startHandshake(Channel channel) {
+ sendHelloMsg(channel, DisconnectCode.NORMAL, channel.getStartTime());
+ }
+
+ @Override
+ public void processMessage(Channel channel, Message message) {
+ HelloMessage msg = (HelloMessage) message;
+
+ if (channel.isFinishHandshake()) {
+ logger.warn("Close channel {}, handshake is finished", channel.getInetSocketAddress());
+ channel.send(new P2pDisconnectMessage(DisconnectReason.DUP_HANDSHAKE));
+ channel.close();
+ return;
+ }
+
+ channel.setHelloMessage(msg);
+
+ DisconnectCode code = ChannelManager.processPeer(channel);
+ if (code != DisconnectCode.NORMAL) {
+ if (!channel.isActive()) {
+ sendHelloMsg(channel, code, msg.getTimestamp());
+ }
+ logDisconnectReason(channel, getDisconnectReason(code));
+ channel.close();
+ return;
+ }
+
+ ChannelManager.updateNodeId(channel, msg.getFrom().getHexId());
+ if (channel.isDisconnect()) {
+ return;
+ }
+
+ if (channel.isActive()) {
+ if (msg.getCode() != DisconnectCode.NORMAL.getValue()
+ || (msg.getNetworkId() != networkId && msg.getVersion() != networkId)) {
+ DisconnectCode disconnectCode = DisconnectCode.forNumber(msg.getCode());
+ //v0.1 have version, v0.2 both have version and networkId
+ logger.info("Handshake failed {}, code: {}, reason: {}, networkId: {}, version: {}",
+ channel.getInetSocketAddress(),
+ msg.getCode(),
+ disconnectCode.name(),
+ msg.getNetworkId(),
+ msg.getVersion());
+ logDisconnectReason(channel, getDisconnectReason(disconnectCode));
+ channel.close();
+ return;
+ }
+ } else {
+
+ if (msg.getNetworkId() != networkId) {
+ logger.info("Peer {} different network id, peer->{}, me->{}",
+ channel.getInetSocketAddress(), msg.getNetworkId(), networkId);
+ sendHelloMsg(channel, DisconnectCode.DIFFERENT_VERSION, msg.getTimestamp());
+ logDisconnectReason(channel, DisconnectReason.DIFFERENT_VERSION);
+ channel.close();
+ return;
+ }
+ sendHelloMsg(channel, DisconnectCode.NORMAL, msg.getTimestamp());
+ }
+ channel.setFinishHandshake(true);
+ channel.updateAvgLatency(System.currentTimeMillis() - channel.getStartTime());
+ Parameter.handlerList.forEach(h -> h.onConnect(channel));
+ }
+
+ private void sendHelloMsg(Channel channel, DisconnectCode code, long time) {
+ HelloMessage helloMessage = new HelloMessage(code, time);
+ channel.send(helloMessage);
+ }
+
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/keepalive/KeepAliveService.java b/p2p/src/main/java/org/tron/p2p/connection/business/keepalive/KeepAliveService.java
new file mode 100644
index 00000000000..47fa9437e98
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/keepalive/KeepAliveService.java
@@ -0,0 +1,70 @@
+package org.tron.p2p.connection.business.keepalive;
+
+import static org.tron.p2p.base.Parameter.KEEP_ALIVE_TIMEOUT;
+import static org.tron.p2p.base.Parameter.PING_TIMEOUT;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.connection.ChannelManager;
+import org.tron.p2p.connection.business.MessageProcess;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.base.P2pDisconnectMessage;
+import org.tron.p2p.connection.message.keepalive.PingMessage;
+import org.tron.p2p.connection.message.keepalive.PongMessage;
+import org.tron.p2p.protos.Connect.DisconnectReason;
+
+@Slf4j(topic = "net")
+public class KeepAliveService implements MessageProcess {
+
+ private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
+ new BasicThreadFactory.Builder().namingPattern("keepAlive").build());
+
+ public void init() {
+ executor.scheduleWithFixedDelay(() -> {
+ try {
+ long now = System.currentTimeMillis();
+ ChannelManager.getChannels().values().stream()
+ .filter(p -> !p.isDisconnect())
+ .forEach(p -> {
+ if (p.waitForPong) {
+ if (now - p.pingSent > KEEP_ALIVE_TIMEOUT) {
+ p.send(new P2pDisconnectMessage(DisconnectReason.PING_TIMEOUT));
+ p.close();
+ }
+ } else {
+ if (now - p.getLastSendTime() > PING_TIMEOUT && p.isFinishHandshake()) {
+ p.send(new PingMessage());
+ p.waitForPong = true;
+ p.pingSent = now;
+ }
+ }
+ });
+ } catch (Exception t) {
+ logger.error("Exception in keep alive task", t);
+ }
+ }, 2, 2, TimeUnit.SECONDS);
+ }
+
+ public void close() {
+ executor.shutdown();
+ }
+
+ @Override
+ public void processMessage(Channel channel, Message message) {
+ switch (message.getType()) {
+ case KEEP_ALIVE_PING:
+ channel.send(new PongMessage());
+ break;
+ case KEEP_ALIVE_PONG:
+ channel.updateAvgLatency(System.currentTimeMillis() - channel.pingSent);
+ channel.waitForPong = false;
+ break;
+ default:
+ break;
+ }
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/pool/ConnPoolService.java b/p2p/src/main/java/org/tron/p2p/connection/business/pool/ConnPoolService.java
new file mode 100644
index 00000000000..ee83df03d46
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/pool/ConnPoolService.java
@@ -0,0 +1,346 @@
+package org.tron.p2p.connection.business.pool;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.bouncycastle.util.encoders.Hex;
+import org.tron.p2p.P2pConfig;
+import org.tron.p2p.P2pEventHandler;
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.connection.ChannelManager;
+import org.tron.p2p.connection.message.base.P2pDisconnectMessage;
+import org.tron.p2p.connection.socket.PeerClient;
+import org.tron.p2p.discover.Node;
+import org.tron.p2p.discover.NodeManager;
+import org.tron.p2p.dns.DnsManager;
+import org.tron.p2p.dns.DnsNode;
+import org.tron.p2p.exception.P2pException;
+import org.tron.p2p.protos.Connect.DisconnectReason;
+import org.tron.p2p.utils.CollectionUtils;
+import org.tron.p2p.utils.NetUtil;
+
+@Slf4j(topic = "net")
+public class ConnPoolService extends P2pEventHandler {
+
+ private final List activePeers = Collections.synchronizedList(new ArrayList<>());
+ private final Cache peerClientCache = CacheBuilder.newBuilder()
+ .maximumSize(1000).expireAfterWrite(120, TimeUnit.SECONDS).recordStats().build();
+ @Getter
+ private final AtomicInteger passivePeersCount = new AtomicInteger(0);
+ @Getter
+ private final AtomicInteger activePeersCount = new AtomicInteger(0);
+ @Getter
+ private final AtomicInteger connectingPeersCount = new AtomicInteger(0);
+ private final ScheduledThreadPoolExecutor poolLoopExecutor = new ScheduledThreadPoolExecutor(1,
+ new BasicThreadFactory.Builder().namingPattern("connPool").build());
+ private final ScheduledExecutorService disconnectExecutor =
+ Executors.newSingleThreadScheduledExecutor(
+ new BasicThreadFactory.Builder().namingPattern("randomDisconnect").build());
+
+ public P2pConfig p2pConfig = Parameter.p2pConfig;
+ private PeerClient peerClient;
+ private final List configActiveNodes = new ArrayList<>();
+ private int minCandidateSize = 50;
+
+ public ConnPoolService() {
+ this.messageTypes = new HashSet<>(); //no message type registers
+ try {
+ Parameter.addP2pEventHandle(this);
+ configActiveNodes.addAll(p2pConfig.getActiveNodes());
+ } catch (P2pException e) {
+ //no exception will throw
+ }
+ }
+
+ public void init(PeerClient peerClient) {
+ this.peerClient = peerClient;
+ poolLoopExecutor.scheduleWithFixedDelay(() -> {
+ try {
+ connect(false);
+ } catch (Exception t) {
+ logger.error("Exception in poolLoopExecutor worker", t);
+ }
+ }, 200, 3600, TimeUnit.MILLISECONDS);
+
+ if (p2pConfig.isDisconnectionPolicyEnable()) {
+ disconnectExecutor.scheduleWithFixedDelay(() -> {
+ try {
+ check();
+ } catch (Exception t) {
+ logger.error("Exception in disconnectExecutor worker", t);
+ }
+ }, 30, 30, TimeUnit.SECONDS);
+ }
+ }
+
+ private void addNode(Set inetSet, Node node) {
+ if (node != null) {
+ if (node.getInetSocketAddressV4() != null) {
+ inetSet.add(node.getInetSocketAddressV4());
+ }
+ if (node.getInetSocketAddressV6() != null) {
+ inetSet.add(node.getInetSocketAddressV6());
+ }
+ }
+ }
+
+ private void connect(boolean isFilterActiveNodes) {
+ List connectNodes = new ArrayList<>();
+
+ //collect already used nodes in channelManager
+ Set addressInUse = new HashSet<>();
+ Set inetInUse = new HashSet<>();
+ Set nodesInUse = new HashSet<>();
+ nodesInUse.add(Hex.toHexString(p2pConfig.getNodeID()));
+ ChannelManager.getChannels().values().forEach(channel -> {
+ if (StringUtils.isNotEmpty(channel.getNodeId())) {
+ nodesInUse.add(channel.getNodeId());
+ }
+ addressInUse.add(channel.getInetAddress());
+ inetInUse.add(channel.getInetSocketAddress());
+ addNode(inetInUse, channel.getNode());
+ });
+
+ addNode(inetInUse, new Node(Parameter.p2pConfig.getNodeID(), Parameter.p2pConfig.getIp(),
+ Parameter.p2pConfig.getIpv6(), Parameter.p2pConfig.getPort()));
+
+ p2pConfig.getActiveNodes().forEach(address -> {
+ if (!isFilterActiveNodes && !inetInUse.contains(address) && !addressInUse.contains(
+ address.getAddress())) {
+ addressInUse.add(address.getAddress());
+ inetInUse.add(address);
+ Node node = new Node(address); //use a random NodeId for config activeNodes
+ if (node.getPreferInetSocketAddress() != null) {
+ connectNodes.add(node);
+ }
+ }
+ });
+
+ //calculate lackSize exclude config activeNodes
+ int activeLackSize = p2pConfig.getMinActiveConnections() - connectingPeersCount.get();
+ int size = StrictMath.max(
+ p2pConfig.getMinConnections() - connectingPeersCount.get() - passivePeersCount.get(),
+ activeLackSize);
+ if (p2pConfig.getMinConnections() <= activePeers.size() && activeLackSize <= 0) {
+ size = 0;
+ }
+ int lackSize = size;
+ if (lackSize > 0) {
+ List connectableNodes = ChannelManager.getNodeDetectService().getConnectableNodes();
+ for (Node node : connectableNodes) {
+ // nodesInUse and inetInUse don't change in method `validNode`
+ if (validNode(node, nodesInUse, inetInUse, null)) {
+ connectNodes.add(node);
+ nodesInUse.add(node.getHexId());
+ inetInUse.add(node.getPreferInetSocketAddress());
+ lackSize -= 1;
+ if (lackSize <= 0) {
+ break;
+ }
+ }
+ }
+ }
+
+ if (lackSize > 0) {
+ List connectableNodes = NodeManager.getConnectableNodes();
+ // nodesInUse and inetInUse don't change in method `getNodes`
+ List newNodes = getNodes(nodesInUse, inetInUse, connectableNodes, lackSize);
+ connectNodes.addAll(newNodes);
+ for (Node node : newNodes) {
+ nodesInUse.add(node.getHexId());
+ inetInUse.add(node.getPreferInetSocketAddress());
+ }
+ lackSize -= newNodes.size();
+ }
+
+ if (lackSize > 0 && !p2pConfig.getTreeUrls().isEmpty()) {
+ List dnsNodes = DnsManager.getDnsNodes();
+ List filtered = new ArrayList<>();
+ Collections.shuffle(dnsNodes);
+ for (DnsNode node : dnsNodes) {
+ if (validNode(node, nodesInUse, inetInUse, null)) {
+ DnsNode copyNode = (DnsNode) node.clone();
+ copyNode.setId(NetUtil.getNodeId());
+ //for node1 {ipv4_1, ipv6}, node2 {ipv4_2, ipv6}, we will not connect it twice
+ addNode(inetInUse, node);
+ filtered.add(copyNode);
+ }
+ }
+ List newNodes = CollectionUtils.truncate(filtered, lackSize);
+ connectNodes.addAll(newNodes);
+ }
+
+ logger.debug("Lack size:{}, connectNodes size:{}, is disconnect trigger: {}",
+ size, connectNodes.size(), isFilterActiveNodes);
+ //establish tcp connection with chose nodes by peerClient
+ {
+ connectNodes.forEach(n -> {
+ logger.info("Connect to peer {}", n.getPreferInetSocketAddress());
+ peerClient.connectAsync(n, false);
+ peerClientCache.put(n.getPreferInetSocketAddress().getAddress(),
+ System.currentTimeMillis());
+ if (!configActiveNodes.contains(n.getPreferInetSocketAddress())) {
+ connectingPeersCount.incrementAndGet();
+ }
+ });
+ }
+ }
+
+ public List getNodes(Set nodesInUse, Set inetInUse,
+ List connectableNodes, int limit) {
+ List filtered = new ArrayList<>();
+ Set dynamicInetInUse = new HashSet<>(inetInUse);
+ for (Node node : connectableNodes) {
+ if (validNode(node, nodesInUse, inetInUse, dynamicInetInUse)) {
+ filtered.add((Node) node.clone());
+ addNode(dynamicInetInUse, node);
+ }
+ }
+
+ filtered.sort(Comparator.comparingLong(node -> -node.getUpdateTime()));
+ int candidateSize = StrictMath.max(limit * 10, minCandidateSize);
+ if (filtered.size() > candidateSize) {
+ filtered = filtered.subList(0, candidateSize);
+ }
+ Collections.shuffle(filtered);
+ return CollectionUtils.truncate(filtered, limit);
+ }
+
+ private boolean validNode(Node node, Set nodesInUse, Set inetInUse,
+ Set dynamicInet) {
+ long now = System.currentTimeMillis();
+ InetSocketAddress inetSocketAddress = node.getPreferInetSocketAddress();
+ InetAddress inetAddress = inetSocketAddress.getAddress();
+ Long forbiddenTime = ChannelManager.getBannedNodes().getIfPresent(inetAddress);
+ if ((forbiddenTime != null && now <= forbiddenTime)
+ || (ChannelManager.getConnectionNum(inetAddress)
+ >= p2pConfig.getMaxConnectionsWithSameIp())
+ || (node.getId() != null && nodesInUse.contains(node.getHexId()))
+ || (peerClientCache.getIfPresent(inetAddress) != null)
+ || inetInUse.contains(inetSocketAddress)
+ || (dynamicInet != null && dynamicInet.contains(inetSocketAddress))) {
+ return false;
+ }
+ return true;
+ }
+
+ private void check() {
+ if (ChannelManager.getChannels().size() < p2pConfig.getMaxConnections()) {
+ return;
+ }
+
+ List channels = new ArrayList<>(activePeers);
+ Collection peers = channels.stream()
+ .filter(peer -> !peer.isDisconnect())
+ .filter(peer -> !peer.isTrustPeer())
+ .filter(peer -> !peer.isActive())
+ .collect(Collectors.toList());
+
+ // if len(peers) >= 0, disconnect randomly
+ if (!peers.isEmpty()) {
+ List list = new ArrayList<>(peers);
+ Channel peer = list.get(new Random().nextInt(peers.size()));
+ logger.info("Disconnect with peer randomly: {}", peer);
+ peer.send(new P2pDisconnectMessage(DisconnectReason.RANDOM_ELIMINATION));
+ peer.close();
+ }
+ }
+
+ private synchronized void logActivePeers() {
+ logger.info("Peer stats: channels {}, activePeers {}, active {}, passive {}",
+ ChannelManager.getChannels().size(), activePeers.size(), activePeersCount.get(),
+ passivePeersCount.get());
+ }
+
+ public void triggerConnect(InetSocketAddress address) {
+ if (configActiveNodes.contains(address)) {
+ return;
+ }
+ connectingPeersCount.decrementAndGet();
+ if (poolLoopExecutor.getQueue().size() >= Parameter.CONN_MAX_QUEUE_SIZE) {
+ logger.warn("ConnPool task' size is greater than or equal to {}",
+ Parameter.CONN_MAX_QUEUE_SIZE);
+ return;
+ }
+ try {
+ if (!ChannelManager.isShutdown) {
+ poolLoopExecutor.submit(() -> {
+ try {
+ connect(true);
+ } catch (Exception t) {
+ logger.error("Exception in poolLoopExecutor worker", t);
+ }
+ });
+ }
+ } catch (Exception e) {
+ logger.warn("Submit task failed, message:{}", e.getMessage());
+ }
+ }
+
+ @Override
+ public synchronized void onConnect(Channel peer) {
+ if (!activePeers.contains(peer)) {
+ if (!peer.isActive()) {
+ passivePeersCount.incrementAndGet();
+ } else {
+ activePeersCount.incrementAndGet();
+ }
+ activePeers.add(peer);
+ }
+ logActivePeers();
+ }
+
+ @Override
+ public synchronized void onDisconnect(Channel peer) {
+ if (activePeers.contains(peer)) {
+ if (!peer.isActive()) {
+ passivePeersCount.decrementAndGet();
+ } else {
+ activePeersCount.decrementAndGet();
+ }
+ activePeers.remove(peer);
+ }
+ logActivePeers();
+ }
+
+ @Override
+ public void onMessage(Channel channel, byte[] data) {
+ //do nothing
+ }
+
+ public void close() {
+ List channels = new ArrayList<>(activePeers);
+ try {
+ channels.forEach(p -> {
+ if (!p.isDisconnect()) {
+ p.send(new P2pDisconnectMessage(DisconnectReason.PEER_QUITING));
+ p.close();
+ }
+ });
+ poolLoopExecutor.shutdownNow();
+ disconnectExecutor.shutdownNow();
+ } catch (Exception e) {
+ logger.warn("Problems shutting down executor", e);
+ }
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/business/upgrade/UpgradeController.java b/p2p/src/main/java/org/tron/p2p/connection/business/upgrade/UpgradeController.java
new file mode 100644
index 00000000000..8e204dea08b
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/business/upgrade/UpgradeController.java
@@ -0,0 +1,38 @@
+package org.tron.p2p.connection.business.upgrade;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import java.io.IOException;
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.exception.P2pException;
+import org.tron.p2p.exception.P2pException.TypeEnum;
+import org.tron.p2p.protos.Connect.CompressMessage;
+import org.tron.p2p.utils.ProtoUtil;
+
+public class UpgradeController {
+
+ public static byte[] codeSendData(int version, byte[] data) throws IOException {
+ if (!supportCompress(version)) {
+ return data;
+ }
+ return ProtoUtil.compressMessage(data).toByteArray();
+ }
+
+ public static byte[] decodeReceiveData(int version, byte[] data)
+ throws P2pException, IOException {
+ if (!supportCompress(version)) {
+ return data;
+ }
+ CompressMessage compressMessage;
+ try {
+ compressMessage = CompressMessage.parseFrom(data);
+ } catch (InvalidProtocolBufferException e) {
+ throw new P2pException(TypeEnum.PARSE_MESSAGE_FAILED, e);
+ }
+ return ProtoUtil.uncompressMessage(compressMessage);
+ }
+
+ private static boolean supportCompress(int version) {
+ return Parameter.version >= 1 && version >= 1;
+ }
+
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/message/Message.java b/p2p/src/main/java/org/tron/p2p/connection/message/Message.java
new file mode 100644
index 00000000000..38a860a54a1
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/message/Message.java
@@ -0,0 +1,78 @@
+package org.tron.p2p.connection.message;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.tron.p2p.connection.message.base.P2pDisconnectMessage;
+import org.tron.p2p.connection.message.detect.StatusMessage;
+import org.tron.p2p.connection.message.handshake.HelloMessage;
+import org.tron.p2p.connection.message.keepalive.PingMessage;
+import org.tron.p2p.connection.message.keepalive.PongMessage;
+import org.tron.p2p.exception.P2pException;
+
+public abstract class Message {
+
+ protected MessageType type;
+ protected byte[] data;
+
+ public Message(MessageType type, byte[] data) {
+ this.type = type;
+ this.data = data;
+ }
+
+ public MessageType getType() {
+ return this.type;
+ }
+
+ public byte[] getData() {
+ return this.data;
+ }
+
+ public byte[] getSendData() {
+ return ArrayUtils.add(this.data, 0, type.getType());
+ }
+
+ public abstract boolean valid();
+
+ public boolean needToLog() {
+ return type.equals(MessageType.DISCONNECT) || type.equals(MessageType.HANDSHAKE_HELLO);
+ }
+
+ public static Message parse(byte[] encode) throws P2pException {
+ byte type = encode[0];
+ try {
+ byte[] data = ArrayUtils.subarray(encode, 1, encode.length);
+ Message message;
+ switch (MessageType.fromByte(type)) {
+ case KEEP_ALIVE_PING:
+ message = new PingMessage(data);
+ break;
+ case KEEP_ALIVE_PONG:
+ message = new PongMessage(data);
+ break;
+ case HANDSHAKE_HELLO:
+ message = new HelloMessage(data);
+ break;
+ case STATUS:
+ message = new StatusMessage(data);
+ break;
+ case DISCONNECT:
+ message = new P2pDisconnectMessage(data);
+ break;
+ default:
+ throw new P2pException(P2pException.TypeEnum.NO_SUCH_MESSAGE, "type=" + type);
+ }
+ if (!message.valid()) {
+ throw new P2pException(P2pException.TypeEnum.BAD_MESSAGE, "type=" + type);
+ }
+ return message;
+ } catch (P2pException p2pException) {
+ throw p2pException;
+ } catch (Exception e) {
+ throw new P2pException(P2pException.TypeEnum.BAD_MESSAGE, "type:" + type);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "type: " + getType() + ", ";
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/message/MessageType.java b/p2p/src/main/java/org/tron/p2p/connection/message/MessageType.java
new file mode 100644
index 00000000000..548bb34a76e
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/message/MessageType.java
@@ -0,0 +1,42 @@
+package org.tron.p2p.connection.message;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public enum MessageType {
+
+ KEEP_ALIVE_PING((byte) 0xff),
+
+ KEEP_ALIVE_PONG((byte) 0xfe),
+
+ HANDSHAKE_HELLO((byte) 0xfd),
+
+ STATUS((byte) 0xfc),
+
+ DISCONNECT((byte) 0xfb),
+
+ UNKNOWN((byte) 0x80);
+
+ private final byte type;
+
+ MessageType(byte type) {
+ this.type = type;
+ }
+
+ public byte getType() {
+ return type;
+ }
+
+ private static final Map map = new HashMap<>();
+
+ static {
+ for (MessageType value : values()) {
+ map.put(value.type, value);
+ }
+ }
+
+ public static MessageType fromByte(byte type) {
+ MessageType typeEnum = map.get(type);
+ return typeEnum == null ? UNKNOWN : typeEnum;
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/message/base/P2pDisconnectMessage.java b/p2p/src/main/java/org/tron/p2p/connection/message/base/P2pDisconnectMessage.java
new file mode 100644
index 00000000000..28460676dd4
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/message/base/P2pDisconnectMessage.java
@@ -0,0 +1,39 @@
+package org.tron.p2p.connection.message.base;
+
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.MessageType;
+import org.tron.p2p.protos.Connect;
+import org.tron.p2p.protos.Connect.DisconnectReason;
+
+
+public class P2pDisconnectMessage extends Message {
+
+ private Connect.P2pDisconnectMessage p2pDisconnectMessage;
+
+ public P2pDisconnectMessage(byte[] data) throws Exception {
+ super(MessageType.DISCONNECT, data);
+ this.p2pDisconnectMessage = Connect.P2pDisconnectMessage.parseFrom(data);
+ }
+
+ public P2pDisconnectMessage(DisconnectReason disconnectReason) {
+ super(MessageType.DISCONNECT, null);
+ this.p2pDisconnectMessage = Connect.P2pDisconnectMessage.newBuilder()
+ .setReason(disconnectReason).build();
+ this.data = p2pDisconnectMessage.toByteArray();
+ }
+
+ private DisconnectReason getReason() {
+ return p2pDisconnectMessage.getReason();
+ }
+
+ @Override
+ public boolean valid() {
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return new StringBuilder().append(super.toString()).append("reason: ")
+ .append(getReason()).toString();
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/message/detect/StatusMessage.java b/p2p/src/main/java/org/tron/p2p/connection/message/detect/StatusMessage.java
new file mode 100644
index 00000000000..3cadb4620dc
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/message/detect/StatusMessage.java
@@ -0,0 +1,61 @@
+package org.tron.p2p.connection.message.detect;
+
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.ChannelManager;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.MessageType;
+import org.tron.p2p.discover.Node;
+import org.tron.p2p.protos.Connect;
+import org.tron.p2p.protos.Discover;
+import org.tron.p2p.utils.NetUtil;
+
+public class StatusMessage extends Message {
+ private Connect.StatusMessage statusMessage;
+
+ public StatusMessage(byte[] data) throws Exception {
+ super(MessageType.STATUS, data);
+ this.statusMessage = Connect.StatusMessage.parseFrom(data);
+ }
+
+ public StatusMessage() {
+ super(MessageType.STATUS, null);
+ Discover.Endpoint endpoint = Parameter.getHomeNode();
+ this.statusMessage = Connect.StatusMessage.newBuilder()
+ .setFrom(endpoint)
+ .setMaxConnections(Parameter.p2pConfig.getMaxConnections())
+ .setCurrentConnections(ChannelManager.getChannels().size())
+ .setNetworkId(Parameter.p2pConfig.getNetworkId())
+ .setTimestamp(System.currentTimeMillis()).build();
+ this.data = statusMessage.toByteArray();
+ }
+
+ public int getNetworkId() {
+ return this.statusMessage.getNetworkId();
+ }
+
+ public int getVersion() {
+ return this.statusMessage.getVersion();
+ }
+
+ public int getRemainConnections() {
+ return this.statusMessage.getMaxConnections() - this.statusMessage.getCurrentConnections();
+ }
+
+ public long getTimestamp() {
+ return this.statusMessage.getTimestamp();
+ }
+
+ public Node getFrom() {
+ return NetUtil.getNode(statusMessage.getFrom());
+ }
+
+ @Override
+ public String toString() {
+ return "[StatusMessage: " + statusMessage;
+ }
+
+ @Override
+ public boolean valid() {
+ return NetUtil.validNode(getFrom());
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/message/handshake/HelloMessage.java b/p2p/src/main/java/org/tron/p2p/connection/message/handshake/HelloMessage.java
new file mode 100644
index 00000000000..726379dc139
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/message/handshake/HelloMessage.java
@@ -0,0 +1,64 @@
+package org.tron.p2p.connection.message.handshake;
+
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.business.handshake.DisconnectCode;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.MessageType;
+import org.tron.p2p.discover.Node;
+import org.tron.p2p.protos.Connect;
+import org.tron.p2p.protos.Discover;
+import org.tron.p2p.utils.NetUtil;
+
+public class HelloMessage extends Message {
+
+ private Connect.HelloMessage helloMessage;
+
+ public HelloMessage(byte[] data) throws Exception {
+ super(MessageType.HANDSHAKE_HELLO, data);
+ this.helloMessage = Connect.HelloMessage.parseFrom(data);
+ }
+
+ public HelloMessage(DisconnectCode code, long time) {
+ super(MessageType.HANDSHAKE_HELLO, null);
+ Discover.Endpoint endpoint = Parameter.getHomeNode();
+ this.helloMessage = Connect.HelloMessage.newBuilder()
+ .setFrom(endpoint)
+ .setNetworkId(Parameter.p2pConfig.getNetworkId())
+ .setCode(code.getValue())
+ .setVersion(Parameter.version)
+ .setTimestamp(time).build();
+ this.data = helloMessage.toByteArray();
+ }
+
+ public int getNetworkId() {
+ return this.helloMessage.getNetworkId();
+ }
+
+ public int getVersion() {
+ return this.helloMessage.getVersion();
+ }
+
+ public int getCode() {
+ return this.helloMessage.getCode();
+ }
+
+ public long getTimestamp() {
+ return this.helloMessage.getTimestamp();
+ }
+
+ public Node getFrom() {
+ return NetUtil.getNode(helloMessage.getFrom());
+ }
+
+ @Override
+ public String toString() {
+ return "HelloMessage networkId: " + getNetworkId()
+ + ", version: " + getVersion()
+ + ", code: " + getCode();
+ }
+
+ @Override
+ public boolean valid() {
+ return NetUtil.validNode(getFrom());
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/message/keepalive/PingMessage.java b/p2p/src/main/java/org/tron/p2p/connection/message/keepalive/PingMessage.java
new file mode 100644
index 00000000000..8191b5e72f1
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/message/keepalive/PingMessage.java
@@ -0,0 +1,33 @@
+package org.tron.p2p.connection.message.keepalive;
+
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.MessageType;
+import org.tron.p2p.protos.Connect;
+
+public class PingMessage extends Message {
+
+ private Connect.KeepAliveMessage keepAliveMessage;
+
+ public PingMessage(byte[] data) throws Exception {
+ super(MessageType.KEEP_ALIVE_PING, data);
+ this.keepAliveMessage = Connect.KeepAliveMessage.parseFrom(data);
+ }
+
+ public PingMessage() {
+ super(MessageType.KEEP_ALIVE_PING, null);
+ this.keepAliveMessage = Connect.KeepAliveMessage.newBuilder()
+ .setTimestamp(System.currentTimeMillis()).build();
+ this.data = this.keepAliveMessage.toByteArray();
+ }
+
+ public long getTimeStamp() {
+ return this.keepAliveMessage.getTimestamp();
+ }
+
+ @Override
+ public boolean valid() {
+ return getTimeStamp() > 0
+ && getTimeStamp() <= System.currentTimeMillis() + Parameter.NETWORK_TIME_DIFF;
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/message/keepalive/PongMessage.java b/p2p/src/main/java/org/tron/p2p/connection/message/keepalive/PongMessage.java
new file mode 100644
index 00000000000..b3689bea93b
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/message/keepalive/PongMessage.java
@@ -0,0 +1,33 @@
+package org.tron.p2p.connection.message.keepalive;
+
+import org.tron.p2p.base.Parameter;
+import org.tron.p2p.connection.message.Message;
+import org.tron.p2p.connection.message.MessageType;
+import org.tron.p2p.protos.Connect;
+
+public class PongMessage extends Message {
+
+ private Connect.KeepAliveMessage keepAliveMessage;
+
+ public PongMessage(byte[] data) throws Exception {
+ super(MessageType.KEEP_ALIVE_PONG, data);
+ this.keepAliveMessage = Connect.KeepAliveMessage.parseFrom(data);
+ }
+
+ public PongMessage() {
+ super(MessageType.KEEP_ALIVE_PONG, null);
+ this.keepAliveMessage = Connect.KeepAliveMessage.newBuilder()
+ .setTimestamp(System.currentTimeMillis()).build();
+ this.data = this.keepAliveMessage.toByteArray();
+ }
+
+ public long getTimeStamp() {
+ return this.keepAliveMessage.getTimestamp();
+ }
+
+ @Override
+ public boolean valid() {
+ return getTimeStamp() > 0
+ && getTimeStamp() <= System.currentTimeMillis() + Parameter.NETWORK_TIME_DIFF;
+ }
+}
diff --git a/p2p/src/main/java/org/tron/p2p/connection/socket/MessageHandler.java b/p2p/src/main/java/org/tron/p2p/connection/socket/MessageHandler.java
new file mode 100644
index 00000000000..797197f0fad
--- /dev/null
+++ b/p2p/src/main/java/org/tron/p2p/connection/socket/MessageHandler.java
@@ -0,0 +1,90 @@
+package org.tron.p2p.connection.socket;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.connection.ChannelManager;
+import org.tron.p2p.connection.business.upgrade.UpgradeController;
+import org.tron.p2p.connection.message.base.P2pDisconnectMessage;
+import org.tron.p2p.connection.message.detect.StatusMessage;
+import org.tron.p2p.exception.P2pException;
+import org.tron.p2p.protos.Connect.DisconnectReason;
+import org.tron.p2p.utils.ByteArray;
+
+@Slf4j(topic = "net")
+public class MessageHandler extends ByteToMessageDecoder {
+
+ private final Channel channel;
+
+ public MessageHandler(Channel channel) {
+ this.channel = channel;
+ }
+
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx) {
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) {
+ logger.debug("Channel active, {}", ctx.channel().remoteAddress());
+ channel.setChannelHandlerContext(ctx);
+ if (channel.isActive()) {
+ if (channel.isDiscoveryMode()) {
+ channel.send(new StatusMessage());
+ } else {
+ ChannelManager.getHandshakeService().startHandshake(channel);
+ }
+ }
+ }
+
+ @Override
+ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List