initialize RocketMQ5

.gitignore

+# Created by .ignore support plugin (hsz.mobi)
+.classpath
+.project
+.settings/
+target/
+*.log*
+*.iml
+.idea/
+*.versionsBackup
+*.DS_Store

CONTRIBUTING.md

+## How To Contribute
+
+We are always very happy to have contributions, whether for trivial cleanups or big new features. We want to have high quality, well documented codes for each programming language. Nor is code the only way to contribute to the project. We strongly value documentation and gladly accept improvements to the documentation.
+Contributing code
+
+To submit a change for inclusion, please do the following:
+If the change is non-trivial please include some unit tests that cover the new functionality.
+
+If you are introducing a completely new feature or API it is a good idea to start a wiki and get consensus on the basic design first.
+
+It is our job to follow up on patches in a timely fashion. Nag us if we aren't doing our job (sometimes we drop things).
+
+## Becoming a Committer
+
+We are always interested in adding new contributors. What we look for are series of contributions, good taste and ongoing interest in the project. If you are interested in becoming a committer, please let one of the existing committers know and they can help you walk through the process.
\ No newline at end of file

README.md

+## Angelia
+Angelia provides a single API for most network related service that uses pluggable transports and codecs. The Angelia API provides the ability for making synchronous, asynchronous, oneway remote calls, push and pull callbacks. The intention is to allow for the use of different transports to fit different needs, yet still maintain the same API for making the remote invocations and only requiring configuration changes, not code changes.
+
+Angelia is a standalone project, separate from the Alibaba RocketMQ and Jukola project, but will be the framework used for many of the projects and components when making remote calls. Angelia is included in the recent releases of the Alibaba Jukola and can be run as a service within the container as well.
+
+### Features
+The features available with Angelia are:
+
+#### 1. Pluggable transports – can use different protocol transports the same remoting API.
+Provided transports:
+
+##### **MVP, a custom-build Minimum Viable Protocol**
+##### **HTTP2**
+     
+#### 2. Pluggable codecs – can use different codecs to convert the invocation payloads into desired data format for wire transfer.
+
+#### 3. Pluggable serialization - can use different serialization implementations for data streams.
+
+Provided serialization implementations:
+
+##### **MessagePack**
+##### **Kryo**
+##### **Fastjson**
+
+#### 4. Data Compression - can use compression codec for compresssion of large payloads.
+
+All the features within Angelia were created with ease of use and extensibility in mind. If you have a suggestion for a new feature or an improvement to a current feature, please let me know.
+
+
+
+
+
+
+
+
+## Documentation
+### [Quick Start]()
+### [User Guide]()
+### [Developer Guide]()
+### [Release Note]()
+
+
+
+----------
+
+## Contributing
+We are always very happy to have contributions, whether for trivial cleanups,big new features or other material rewards. more details see [here](CONTRIBUTING.md) 
+ 
+----------
+## License
+[Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0.html) Copyright (C) 2010-2016 Alibaba Group Holding Limited
\ No newline at end of file

example/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>rpc-all</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>example</artifactId>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rpc-impl</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+
+</project>
\ No newline at end of file

example/src/main/java/org/apache/rocketmq/rpc/example/model/TradeRequest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.rpc.example.model;
+
+import org.msgpack.annotation.Message;
+
+@Message
+public class TradeRequest {
+    private long orderId;
+    private String orderDesc;
+
+    public long getOrderId() {
+        return orderId;
+    }
+
+    public void setOrderId(long orderId) {
+        this.orderId = orderId;
+    }
+
+    public String getOrderDesc() {
+        return orderDesc;
+    }
+
+    public void setOrderDesc(String orderDesc) {
+        this.orderDesc = orderDesc;
+    }
+
+    @Override
+    public String toString() {
+        return "TradeRequest [orderId=" + orderId + ", orderDesc=" + orderDesc + "]";
+    }
+}

example/src/main/java/org/apache/rocketmq/rpc/example/model/TradeResponse.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.rpc.example.model;
+
+import org.msgpack.annotation.Message;
+
+@Message
+public class TradeResponse {
+    private long storeTimestamp;
+
+    public long getStoreTimestamp() {
+        return storeTimestamp;
+    }
+
+    public void setStoreTimestamp(long storeTimestamp) {
+        this.storeTimestamp = storeTimestamp;
+    }
+
+    @Override
+    public String toString() {
+        return "TradeResponse [storeTimestamp=" + storeTimestamp + "]";
+    }
+}

example/src/main/java/org/apache/rocketmq/rpc/example/model/TradeServiceAPI.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.rpc.example.model;
+
+import java.util.ArrayList;
+import org.apache.rocketmq.rpc.annotation.RemoteMethod;
+import org.apache.rocketmq.rpc.annotation.RemoteService;
+
+@RemoteService(name = "TradeServiceAPI")
+public interface TradeServiceAPI {
+    @RemoteMethod(name = "commitOrder")
+    TradeResponse commitOrder(TradeRequest request) throws InterruptedException;
+
+    @RemoteMethod(name = "deleteOrder")
+    void deleteOrder(TradeRequest request) throws InterruptedException;
+
+    @RemoteMethod(name = "throwUserException")
+    void throwUserException(TradeRequest request) throws Exception;
+
+    @RemoteMethod(name = "throwRuntimeException")
+    void throwRuntimeException(TradeRequest request);
+
+    @RemoteMethod(name = "getOrderList")
+    ArrayList<TradeResponse> getOrderList(TradeRequest request, int count);
+
+    @RemoteMethod(name = "getOrderListSize")
+    int getOrderListSize(ArrayList<TradeRequest> request, String obj);
+
+    @RemoteMethod(name = "setOrder")
+    String setOrder();
+
+}
\ No newline at end of file

example/src/main/java/org/apache/rocketmq/rpc/example/model/TradeServiceAPIGen.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.rpc.example.model;
+
+import java.util.ArrayList;
+import org.apache.rocketmq.rpc.annotation.MethodType;
+import org.apache.rocketmq.rpc.annotation.RemoteMethod;
+import org.apache.rocketmq.rpc.annotation.RemoteService;
+import org.apache.rocketmq.rpc.api.Promise;
+
+@RemoteService(name = "TradeServiceAPI")
+public interface TradeServiceAPIGen extends TradeServiceAPI {
+    @RemoteMethod(name = "commitOrder", type = MethodType.ASYNC)
+    Promise<TradeResponse> commitOrderAsync(TradeRequest request);
+
+    @RemoteMethod(name = "commitOrder", type = MethodType.ONEWAY)
+    void commitOrderOneway(final TradeRequest request);
+
+    @RemoteMethod(name = "deleteOrder", type = MethodType.ASYNC)
+    Promise<TradeResponse> deleteOrderAsync(TradeRequest request);
+
+    @RemoteMethod(name = "deleteOrder", type = MethodType.ONEWAY)
+    void deleteOrderOneway(TradeRequest request);
+
+    @RemoteMethod(name = "throwUserException", type = MethodType.ASYNC)
+    void throwUserExceptionAsync(TradeRequest request);
+
+    @RemoteMethod(name = "throwUserException", type = MethodType.ONEWAY)
+    void throwUserExceptionOneway(TradeRequest request);
+
+    @RemoteMethod(name = "throwRuntimeException", type = MethodType.ASYNC)
+    void throwRuntimeExceptionAsync(TradeRequest request);
+
+    @RemoteMethod(name = "throwRuntimeException", type = MethodType.ONEWAY)
+    void throwRuntimeExceptionOneway(TradeRequest request);
+
+    @RemoteMethod(name = "getOrderList", type = MethodType.ASYNC)
+    Promise<ArrayList<TradeResponse>> getOrderListAsync(TradeRequest request, int count);
+
+    @RemoteMethod(name = "getOrderListSize", type = MethodType.ASYNC)
+    Promise<Integer> getOrderListSizeAsync(ArrayList<TradeRequest> request, String obj);
+
+    @RemoteMethod(name = "setOrder", type = MethodType.ASYNC)
+    Promise<String> setOrderAsync();
+}

example/src/main/java/org/apache/rocketmq/rpc/example/model/TradeServiceAPIImpl.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.rpc.example.model;
+
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class TradeServiceAPIImpl implements TradeServiceAPI {
+
+    private final AtomicLong count = new AtomicLong();
+
+    public TradeServiceAPIImpl() {
+        super();
+
+    }
+
+    @Override
+    public TradeResponse commitOrder(TradeRequest request) throws InterruptedException {
+        TradeResponse response = new TradeResponse();
+        response.setStoreTimestamp(System.currentTimeMillis());
+        return response;
+    }
+
+    @Override
+    public void deleteOrder(TradeRequest request) throws InterruptedException {
+        System.out.println("deleteOrder: " + request);
+    }
+
+    @Override
+    public void throwUserException(TradeRequest request) throws Exception {
+        throw new Exception("User Exception detail message");
+    }
+
+    @Override
+    public void throwRuntimeException(TradeRequest request) {
+        throw new RuntimeException("runtime");
+    }
+
+    @Override
+    public ArrayList<TradeResponse> getOrderList(final TradeRequest request, final int count) {
+        return null;
+    }
+
+    @Override
+    public int getOrderListSize(final ArrayList<TradeRequest> request, final String obj) {
+        return 0;
+    }
+
+    @Override
+    public String setOrder() {
+        return null;
+    }
+}

example/src/main/java/org/apache/rocketmq/rpc/example/quickstart/Client.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.rpc.example.quickstart;
+
+import java.util.Properties;
+import org.apache.rocketmq.rpc.RpcBootstrapFactory;
+import org.apache.rocketmq.rpc.api.SimpleClient;
+import org.apache.rocketmq.rpc.example.model.TradeRequest;
+import org.apache.rocketmq.rpc.example.model.TradeResponse;
+import org.apache.rocketmq.rpc.example.model.TradeServiceAPI;
+
+public class Client {
+    public static void main(String[] args) throws InterruptedException {
+        SimpleClient client = RpcBootstrapFactory.createClientBootstrap(new Properties());
+        TradeServiceAPI tradeService = client.bind(TradeServiceAPI.class, "127.0.0.1:8888", new Properties());
+        client.start();
+
+        TradeResponse tradeResponse = tradeService.commitOrder(new TradeRequest());
+        System.out.println(tradeResponse);
+    }
+}

example/src/main/java/org/apache/rocketmq/rpc/example/quickstart/Server.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.rpc.example.quickstart;
+
+import java.util.Properties;
+import org.apache.rocketmq.rpc.RpcBootstrapFactory;
+import org.apache.rocketmq.rpc.api.SimpleServer;
+import org.apache.rocketmq.rpc.example.model.TradeServiceAPIImpl;
+
+public class Server {
+    public static void main(String[] args) {
+        SimpleServer server = RpcBootstrapFactory.createServerBootstrap(new Properties());
+        server.publish(new TradeServiceAPIImpl());
+        server.start();
+    }
+}

pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+    <inceptionYear>2012</inceptionYear>
+    <groupId>org.apache.rocketmq</groupId>
+    <artifactId>rpc-all</artifactId>
+    <version>2.0.0-SNAPSHOT</version>
+    <packaging>pom</packaging>
+
+    <modules>
+        <module>remoting-core</module>
+        <module>example</module>
+    </modules>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <maven.test.skip>false</maven.test.skip>
+        <maven.javadoc.skip>true</maven.javadoc.skip>
+        <!-- compiler settings properties -->
+        <maven.compiler.source>1.7</maven.compiler.source>
+        <maven.compiler.target>1.7</maven.compiler.target>
+        <metrics.version>3.1.2</metrics.version>
+    </properties>
+
+    <build>
+
+        <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>versions-maven-plugin</artifactId>
+                <version>2.2</version>
+            </plugin>
+            <plugin>
+                <artifactId>maven-enforcer-plugin</artifactId>
+                <version>1.4.1</version>
+                <executions>
+                    <execution>
+                        <id>enforce-ban-circular-dependencies</id>
+                        <goals>
+                            <goal>enforce</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <rules>
+                        <banCircularDependencies/>
+                    </rules>
+                    <fail>true</fail>
+                </configuration>
+                <dependencies>
+                    <dependency>
+                        <groupId>org.codehaus.mojo</groupId>
+                        <artifactId>extra-enforcer-rules</artifactId>
+                        <version>1.0-beta-4</version>
+                    </dependency>
+                </dependencies>
+            </plugin>
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.5.1</version>
+                <configuration>
+                    <source>${maven.compiler.source}</source>
+                    <target>${maven.compiler.target}</target>
+                    <compilerVersion>${maven.compiler.source}</compilerVersion>
+                    <showDeprecation>true</showDeprecation>
+                    <showWarnings>true</showWarnings>
+                </configuration>
+            </plugin>
+            <plugin>
+                <artifactId>maven-javadoc-plugin</artifactId>
+                <version>2.10.4</version>
+                <executions>
+                    <execution>
+                        <id>attach-javadocs</id>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <artifactId>maven-source-plugin</artifactId>
+                <version>3.0.1</version>
+                <executions>
+                    <execution>
+                        <id>attach-sources</id>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <artifactId>maven-checkstyle-plugin</artifactId>
+                <version>2.17</version>
+                <executions>
+                    <execution>
+                        <id>verify</id>
+                        <phase>verify</phase>
+                        <configuration>
+                            <configLocation>style/checkstyle.xml</configLocation>
+                            <encoding>${project.build.sourceEncoding}</encoding>
+                            <consoleOutput>true</consoleOutput>
+                            <failsOnError>true</failsOnError>
+                        </configuration>
+                        <goals>
+                            <goal>check</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <sourceDirectory>${project.build.sourceDirectory}</sourceDirectory>
+                </configuration>
+            </plugin>
+        </plugins>
+
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <groupId>org.codehaus.mojo</groupId>
+                    <artifactId>clirr-maven-plugin</artifactId>
+                    <version>2.7</version>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <filtering>false</filtering>
+            </resource>
+        </resources>
+    </build>
+
+    <distributionManagement>
+        <repository>
+            <id>releases</id>
+            <url>http://mvnrepo.alibaba-inc.com/mvn/releases</url>
+        </repository>
+        <snapshotRepository>
+            <id>snapshots</id>
+            <url>http://mvnrepo.alibaba-inc.com/mvn/snapshots</url>
+        </snapshotRepository>
+    </distributionManagement>
+
+</project>
\ No newline at end of file

remoting-core/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>rpc-all</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>remoting-core</artifactId>
+    <packaging>pom</packaging>
+
+    <modules>
+        <module>remoting-api</module>
+        <module>remoting-impl</module>
+        <module>rpc-api</module>
+        <module>rpc-impl</module>
+    </modules>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.11</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>2.6.0</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>2.6.3</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.slf4j</groupId>
+                <artifactId>slf4j-api</artifactId>
+                <version>1.7.6</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-lang3</artifactId>
+                <version>3.4</version>
+            </dependency>
+            <dependency>
+                <groupId>org.jetbrains</groupId>
+                <artifactId>annotations</artifactId>
+                <version>15.0</version>
+            </dependency>
+            <dependency>
+                <groupId>io.netty</groupId>
+                <artifactId>netty-all</artifactId>
+                <version>4.1.6.Final</version>
+            </dependency>
+            <dependency>
+                <groupId>com.alibaba</groupId>
+                <artifactId>fastjson</artifactId>
+                <version>1.2.29</version>
+            </dependency>
+            <dependency>
+                <groupId>org.msgpack</groupId>
+                <artifactId>msgpack</artifactId>
+                <version>0.6.12</version>
+            </dependency>
+            <dependency>
+                <groupId>com.esotericsoftware</groupId>
+                <artifactId>kryo</artifactId>
+                <version>3.0.3</version>
+            </dependency>
+            <dependency>
+                <groupId>org.javassist</groupId>
+                <artifactId>javassist</artifactId>
+                <version>3.20.0-GA</version>
+            </dependency>
+            <!-- http://netty.io/wiki/requirements-for-4.x.html#wiki-h3-8 -->
+            <dependency>
+                <groupId>io.netty</groupId>
+                <artifactId>netty-tcnative-boringssl-static</artifactId>
+                <version>1.1.33.Fork22</version>
+            </dependency>
+            <dependency>
+                <groupId>io.dropwizard.metrics</groupId>
+                <artifactId>metrics-core</artifactId>
+                <version>${metrics.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.dropwizard.metrics</groupId>
+                <artifactId>metrics-healthchecks</artifactId>
+                <version>${metrics.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>19.0</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>remoting-api</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>rpc-api</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>remoting-impl</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>rpc-impl</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+</project>
\ No newline at end of file

remoting-core/remoting-api/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>remoting-core</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>remoting-api</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains</groupId>
+            <artifactId>annotations</artifactId>
+        </dependency>
+    </dependencies>
+</project>
\ No newline at end of file

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/AsyncHandler.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+/**
+ * The AsyncHandler interface is implemented when wishing to receive callback notification of the completion of
+ * service invoked asynchronously.
+ *
+ * @since 1.0.0
+ */
+public interface AsyncHandler {
+    void onFailure(RemotingCommand command);
+
+    void onSuccess(RemotingCommand command);
+
+    void onTimeout(long costTimeMillis, long timeoutMillis);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/ConnectionService.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import org.apache.rocketmq.remoting.api.channel.ChannelEventListener;
+
+public interface ConnectionService {
+    void registerChannelEventListener(ChannelEventListener listener);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/ObjectLifecycle.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+
+public interface ObjectLifecycle {
+    @PostConstruct
+    void start();
+
+    @PreDestroy
+    void stop();
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/RemotingClient.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public interface RemotingClient extends RemotingService {
+    RemotingCommand invoke(String address, RemotingCommand request, long timeoutMillis);
+
+    void invokeAsync(String address, RemotingCommand request, AsyncHandler asyncHandler, long timeoutMillis);
+
+    void invokeOneWay(String address, RemotingCommand request, long timeoutMillis);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/RemotingEndPoint.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+public enum RemotingEndPoint {
+    REQUEST,
+    RESPONSE
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/RemotingMarshaller.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import org.apache.rocketmq.remoting.api.protocol.ProtocolFactory;
+import org.apache.rocketmq.remoting.api.serializable.SerializerFactory;
+
+public interface RemotingMarshaller {
+    ProtocolFactory protocolFactory();
+
+    SerializerFactory serializerFactory();
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/RemotingServer.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import org.apache.rocketmq.remoting.api.channel.RemotingChannel;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public interface RemotingServer extends RemotingService {
+    int localListenPort();
+
+    RemotingCommand invoke(RemotingChannel remotingChannel, RemotingCommand request, long timeoutMillis);
+
+    void invokeAsync(RemotingChannel remotingChannel, RemotingCommand request, AsyncHandler asyncHandler,
+        long timeoutMillis);
+
+    void invokeOneWay(RemotingChannel remotingChannel, RemotingCommand request, long timeoutMillis);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/RemotingService.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import java.util.concurrent.ExecutorService;
+import org.apache.rocketmq.remoting.api.command.RemotingCommandFactory;
+import org.apache.rocketmq.remoting.api.interceptor.Interceptor;
+import org.apache.rocketmq.remoting.common.Pair;
+
+public interface RemotingService extends RemotingMarshaller, ConnectionService, ObjectLifecycle {
+    void registerInterceptor(Interceptor interceptor);
+
+    void registerRequestProcessor(final String requestCode, final RequestProcessor processor,
+        final ExecutorService executor);
+
+    void registerRequestProcessor(final String requestCode, final RequestProcessor processor);
+
+    void unregisterRequestProcessor(final String requestCode);
+
+    Pair<RequestProcessor, ExecutorService> processor(final String requestCode);
+
+    String remotingInstanceId();
+
+    RemotingCommandFactory commandFactory();
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/RequestProcessor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api;
+
+import org.apache.rocketmq.remoting.api.channel.RemotingChannel;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public interface RequestProcessor {
+    RemotingCommand processRequest(RemotingChannel channel, RemotingCommand request);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/buffer/ByteBufferWrapper.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.buffer;
+
+import java.nio.ByteBuffer;
+
+public interface ByteBufferWrapper {
+    void writeByte(int index, byte data);
+
+    void writeByte(byte data);
+
+    byte readByte();
+
+    void writeInt(int data);
+
+    void writeBytes(byte[] data);
+
+    void writeBytes(ByteBuffer data);
+
+    int readableBytes();
+
+    int readInt();
+
+    void readBytes(byte[] dst);
+
+    void readBytes(ByteBuffer dst);
+
+    int readerIndex();
+
+    void setReaderIndex(int readerIndex);
+
+    void writeLong(long id);
+
+    long readLong();
+
+    void ensureCapacity(int capacity);
+
+    short readShort();
+
+    void writeShort(short value);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/channel/ChannelEventListener.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.channel;
+
+public interface ChannelEventListener {
+    void onChannelConnect(final RemotingChannel channel);
+
+    void onChannelClose(final RemotingChannel channel);
+
+    void onChannelException(final RemotingChannel channel);
+
+    void onChannelIdle(final RemotingChannel channel);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/channel/ChannelHandlerContextWrapper.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.remoting.api.channel;
+
+public interface ChannelHandlerContextWrapper<T> {
+    T getContext();
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/channel/ChunkRegion.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.channel;
+
+import java.io.IOException;
+import java.nio.channels.WritableByteChannel;
+
+public interface ChunkRegion {
+    void release();
+
+    /**
+     * @return Returns the offset in the file where the transfer began.
+     */
+    long position();
+
+    /**
+     * @return Return the bytes which was transferred already
+     */
+    long transferred();
+
+    /**
+     * @return Returns the number of bytes to transfer.
+     */
+    long count();
+
+    /**
+     * Transfers the content of this file region to the specified channel.
+     *
+     * @param target the destination of the transfer
+     * @param position the relative offset of the file where the transfer begins
+     * from. For example, <tt>0</tt> will make the transfer start
+     * from {@link #position()}th byte and
+     * <tt>{@link #count()} - 1</tt> will make the last byte of the
+     * region transferred.
+     * @return the length of the transferred file region
+     * @throws IOException IOException
+     */
+    long transferTo(WritableByteChannel target, long position) throws IOException;
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/channel/RemotingChannel.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.channel;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public interface RemotingChannel {
+    /**
+     * Returns the local address where this {@code RemotingChannel} is bound to.  The returned
+     * {@link SocketAddress} is supposed to be down-cast into more concrete
+     * type such as {@link InetSocketAddress} to retrieve the detailed
+     * information.
+     *
+     * @return the local address of this channel.
+     * {@code null} if this channel is not bound.
+     */
+    SocketAddress localAddress();
+
+    /**
+     * Returns the remote address where this {@code RemotingChannel} is connected to.  The
+     * returned {@link SocketAddress} is supposed to be down-cast into more
+     * concrete type such as {@link InetSocketAddress} to retrieve the detailed
+     * information.
+     *
+     * @return the remote address of this channel.
+     * {@code null} if this channel is not connected.
+     */
+    SocketAddress remoteAddress();
+
+    /**
+     * Returns {@code true} if and only if the I/O thread will perform the
+     * requested write operation immediately.  Any write requests made when
+     * this method returns {@code false} are queued until the I/O thread is
+     * ready to process the queued write requests.
+     */
+    boolean isWritable();
+
+    /**
+     * Returns {@code true} if the {@code RemotingChannel} is active and so connected.
+     */
+    boolean isActive();
+
+    /**
+     * Requests to close the {@code RemotingChannel} immediately.
+     */
+    void close();
+
+    /**
+     * Writes a response {@code RemotingCommand} to remote.
+     *
+     * @param command the response command
+     */
+    void reply(RemotingCommand command);
+
+    /**
+     * Writes a response {@code ChunkRegion} to remote.
+     *
+     * @param fileRegion the response chunk file region
+     */
+    void reply(ChunkRegion fileRegion);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/command/RemotingCommand.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.command;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+import org.apache.rocketmq.remoting.api.serializable.SerializerFactory;
+import org.apache.rocketmq.remoting.common.TypePresentation;
+
+public interface RemotingCommand {
+    byte protocolType();
+
+    void protocolType(byte value);
+
+    int requestID();
+
+    void requestID(int value);
+
+    byte serializerType();
+
+    void serializerType(byte value);
+
+    TrafficType trafficType();
+
+    void trafficType(TrafficType value);
+
+    String opCode();
+
+    void opCode(String value);
+
+    String remark();
+
+    void remark(String value);
+
+    Map<String, String> properties();
+
+    void properties(Map<String, String> value);
+
+    String property(String key);
+
+    void property(String key, String value);
+
+    Object parameter();
+
+    void parameter(Object value);
+
+    byte[] parameterBytes();
+
+    void parameterBytes(byte[] value);
+
+    byte[] extraPayload();
+
+    void extraPayload(byte[] value);
+
+    <T> T parameter(final SerializerFactory serializerFactory, Class<T> c);
+
+    <T> T parameter(final SerializerFactory serializerFactory, final TypePresentation<T> typePresentation);
+
+    <T> T parameter(final SerializerFactory serializerFactory, final Type type);
+
+    enum CommandFlag {
+        SUCCESS("0"),
+        ERROR("-1");
+
+        private String flag;
+
+        CommandFlag(final String flag) {
+            this.flag = flag;
+        }
+
+        public String flag() {
+            return flag;
+        }
+    }
+}
\ No newline at end of file

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/command/RemotingCommandFactory.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.command;
+
+public interface RemotingCommandFactory {
+    RemotingCommand createRequest();
+
+    RemotingCommand createResponse(RemotingCommand command);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/command/TrafficType.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.command;
+
+public enum TrafficType {
+    REQUEST_SYNC,
+    REQUEST_ASYNC,
+    REQUEST_ONEWAY,
+    RESPONSE;
+
+    public static TrafficType parse(int index) {
+        switch (index) {
+            case 0:
+                return REQUEST_SYNC;
+            case 1:
+                return REQUEST_ASYNC;
+            case 2:
+                return REQUEST_ONEWAY;
+            case 3:
+                return RESPONSE;
+            default:
+                throw new IllegalArgumentException("Not supported " + index);
+        }
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compressable/Compressor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.compressable;
+
+public interface Compressor {
+    String name();
+
+    byte type();
+
+    byte[] compress(final byte[] content) throws Exception;
+
+    byte[] deCompress(final byte[] content) throws Exception;
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compressable/CompressorFactory.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.compressable;
+
+public interface CompressorFactory {
+    void register(Compressor compressor);
+
+    byte type(String compressionName);
+
+    Compressor get(byte type);
+
+    void clearAll();
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/exception/NestedRuntimeException.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.remoting.api.exception;
+
+/**
+ * Handy class for wrapping runtime {@code Exceptions} with a root cause.
+ *
+ * <p>This class is {@code abstract} to force the programmer to extend
+ * the class. {@code getMessage} will include nested exception
+ * information; {@code getRootCause} will include the innermost cause of
+ * this exception, if any; {@code printStackTrace} and other like methods will
+ * delegate to the wrapped exception, if any.
+ *
+ * @since 1.0.0
+ */
+public abstract class NestedRuntimeException extends RuntimeException {
+    private static final long serialVersionUID = -8371779880133933367L;
+
+    /**
+     * Construct a {@code NestedRuntimeException} with the specified detail message.
+     *
+     * @param msg the detail message
+     */
+    public NestedRuntimeException(String msg) {
+        super(msg);
+    }
+
+    /**
+     * Construct a {@code NestedRuntimeException} with the specified detail message
+     * and nested exception.
+     *
+     * @param msg the detail message
+     * @param cause the nested exception
+     */
+    public NestedRuntimeException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+
+    /**
+     * Build a message for the given base message and root cause.
+     *
+     * @param message the base message
+     * @param cause the root cause
+     * @return the full exception message
+     */
+    private static String getMessageWithCause(String message, Throwable cause) {
+        if (cause != null) {
+            StringBuilder sb = new StringBuilder();
+            if (message != null) {
+                sb.append(message).append("; ");
+            }
+            sb.append("nested exception is ").append(cause);
+            return sb.toString();
+        } else {
+            return message;
+        }
+    }
+
+    /**
+     * Return the detail message, including the message from the nested exception
+     * if there is one.
+     */
+    @Override
+    public String getMessage() {
+        return getMessageWithCause(super.getMessage(), getCause());
+    }
+
+    /**
+     * Retrieve the innermost cause of this exception, if any.
+     *
+     * @return the innermost exception, or {@code null} if none
+     */
+    public Throwable getRootCause() {
+        Throwable rootCause = null;
+        Throwable cause = getCause();
+        while (cause != null && cause != rootCause) {
+            rootCause = cause;
+            cause = cause.getCause();
+        }
+        return rootCause;
+    }
+
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/exception/RemoteAccessException.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.exception;
+
+/**
+ * Generic remote access exception. A service proxy for any remoting
+ * protocol should throw this exception or subclasses of it, in order
+ * to transparently expose a plain Java business interface.
+ *
+ * <p>A client may catch RemoteAccessException if it wants to, but as
+ * remote access errors are typically unrecoverable, it will probably let
+ * such exceptions propagate to a higher level that handles them generically.
+ * In this case, the client opCode doesn't show any signs of being involved in
+ * remote access, as there aren't any remoting-specific dependencies.
+ *
+ * @since 1.0.0
+ */
+public class RemoteAccessException extends NestedRuntimeException {
+    private static final long serialVersionUID = 6280428909532427263L;
+
+    /**
+     * Constructor for RemoteAccessException with the specified detail message.
+     *
+     * @param msg the detail message
+     */
+    public RemoteAccessException(String msg) {
+        super(msg);
+    }
+
+    /**
+     * Constructor for RemoteAccessException with the specified detail message
+     * and nested exception.
+     *
+     * @param msg the detail message
+     * @param cause the root cause (usually from using an underlying
+     * remoting API such as RMI)
+     */
+    public RemoteAccessException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/exception/RemoteCodecException.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.exception;
+
+/**
+ * @since 1.0.0
+ */
+public class RemoteCodecException extends RemoteAccessException {
+    private static final long serialVersionUID = -7597014042746200543L;
+
+    public RemoteCodecException(String msg) {
+        super(msg);
+    }
+
+    public RemoteCodecException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/exception/RemoteConnectFailureException.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.exception;
+
+/**
+ * RemoteConnectFailureException will be thrown when connection
+ * could not be established with a remote service.
+ *
+ * @since 1.0.0
+ */
+public class RemoteConnectFailureException extends RemoteAccessException {
+    private static final long serialVersionUID = -5565366231695911316L;
+
+    /**
+     * Constructor for RemoteConnectFailureException with the specified detail message
+     * and nested exception.
+     *
+     * @param msg the detail message
+     * @param cause the root cause from the remoting API in use
+     */
+    public RemoteConnectFailureException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+
+    /**
+     * Constructor for RemoteConnectFailureException with the specified detail message.
+     *
+     * @param msg the detail message
+     */
+    public RemoteConnectFailureException(String msg) {
+        super(msg);
+    }
+
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/exception/RemoteTimeoutException.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.remoting.api.exception;
+
+/**
+ * RemoteTimeoutException will be thrown when the execution
+ * of the target method did not complete before a configurable
+ * timeout, for example when a reply message was not received.
+ *
+ * @since 1.0.0
+ */
+public class RemoteTimeoutException extends RemoteAccessException {
+    private static final long serialVersionUID = 8710772392914461626L;
+
+    /**
+     * Constructor for RemoteTimeoutException with the specified detail message,configurable timeout.
+     *
+     * @param msg the detail message
+     * @param timeoutMillis configurable timeout
+     */
+    public RemoteTimeoutException(String msg, long timeoutMillis) {
+        this(msg, timeoutMillis, null);
+    }
+
+    /**
+     * Constructor for RemoteTimeoutException with the specified detail message,configurable timeout
+     * and nested exception..
+     *
+     * @param msg the detail message
+     * @param timeoutMillis configurable timeout
+     * @param cause Exception cause
+     */
+    public RemoteTimeoutException(String msg, long timeoutMillis, Throwable cause) {
+        super(String.format("%s, waiting for %s ms", msg, timeoutMillis), cause);
+    }
+
+    /**
+     * Constructor for RemoteTimeoutException with the specified detail message.
+     *
+     * @param msg the detail message
+     */
+    public RemoteTimeoutException(String msg) {
+        super(msg);
+    }
+
+    /**
+     * Constructor for RemoteTimeoutException with the specified detail message
+     * and nested exception.
+     *
+     * @param msg the detail message
+     * @param cause the root cause from the remoting API in use
+     */
+    public RemoteTimeoutException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/ExceptionContext.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.interceptor;
+
+import org.apache.rocketmq.remoting.api.RemotingEndPoint;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public class ExceptionContext extends RequestContext {
+    private Throwable exception;
+    private String remark;
+
+    public ExceptionContext(RemotingEndPoint remotingEndPoint, String remoteAddr, RemotingCommand request,
+        Throwable exception, String remark) {
+        super(remotingEndPoint, remoteAddr, request);
+        this.remotingEndPoint = remotingEndPoint;
+        this.remoteAddr = remoteAddr;
+        this.request = request;
+        this.exception = exception;
+        this.remark = remark;
+    }
+
+    public RemotingEndPoint getRemotingEndPoint() {
+        return remotingEndPoint;
+    }
+
+    public void setRemotingEndPoint(RemotingEndPoint remotingEndPoint) {
+        this.remotingEndPoint = remotingEndPoint;
+    }
+
+    public String getRemoteAddr() {
+        return remoteAddr;
+    }
+
+    public void setRemoteAddr(String remoteAddr) {
+        this.remoteAddr = remoteAddr;
+    }
+
+    public RemotingCommand getRequest() {
+        return request;
+    }
+
+    public void setRequest(RemotingCommand request) {
+        this.request = request;
+    }
+
+    public Throwable getException() {
+        return exception;
+    }
+
+    public void setException(Throwable exception) {
+        this.exception = exception;
+    }
+
+    public String getRemark() {
+        return remark;
+    }
+
+    public void setRemark(String remark) {
+        this.remark = remark;
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/Interceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.interceptor;
+
+public interface Interceptor {
+    void beforeRequest(final RequestContext context);
+
+    void afterResponseReceived(final ResponseContext context);
+
+    void onException(final ExceptionContext context);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/InterceptorGroup.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.interceptor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class InterceptorGroup {
+    private final List<Interceptor> interceptors = new ArrayList<Interceptor>();
+
+    public void registerInterceptor(final Interceptor interceptor) {
+        if (interceptor != null) {
+            this.interceptors.add(interceptor);
+        }
+    }
+
+    public void beforeRequest(final RequestContext context) {
+        for (Interceptor interceptor : interceptors) {
+            interceptor.beforeRequest(context);
+        }
+    }
+
+    public void afterResponseReceived(final ResponseContext context) {
+        for (Interceptor interceptor : interceptors) {
+            interceptor.afterResponseReceived(context);
+        }
+    }
+
+    public void onException(final ExceptionContext context) {
+        for (Interceptor interceptor : interceptors) {
+            interceptor.onException(context);
+        }
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/RequestContext.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.interceptor;
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+import org.apache.rocketmq.remoting.api.RemotingEndPoint;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public class RequestContext {
+    protected RemotingEndPoint remotingEndPoint;
+    protected String remoteAddr;
+    protected RemotingCommand request;
+
+    public RequestContext(RemotingEndPoint remotingEndPoint, String remoteAddr, RemotingCommand request) {
+        super();
+        this.remotingEndPoint = remotingEndPoint;
+        this.remoteAddr = remoteAddr;
+        this.request = request;
+    }
+
+    public RemotingEndPoint getRemotingEndPoint() {
+        return remotingEndPoint;
+    }
+
+    public void setRemotingEndPoint(RemotingEndPoint remotingEndPoint) {
+        this.remotingEndPoint = remotingEndPoint;
+    }
+
+    public String getRemoteAddr() {
+        return remoteAddr;
+    }
+
+    public void setRemoteAddr(String remoteAddr) {
+        this.remoteAddr = remoteAddr;
+    }
+
+    public RemotingCommand getRequest() {
+        return request;
+    }
+
+    public void setRequest(RemotingCommand request) {
+        this.request = request;
+    }
+
+    @Override
+    public String toString() {
+        return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/ResponseContext.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.interceptor;
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+import org.apache.rocketmq.remoting.api.RemotingEndPoint;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public class ResponseContext extends RequestContext {
+    private RemotingCommand response;
+
+    public ResponseContext(RemotingEndPoint remotingEndPoint, String remoteAddr, RemotingCommand request,
+        RemotingCommand response) {
+        super(remotingEndPoint, remoteAddr, request);
+        this.remotingEndPoint = remotingEndPoint;
+        this.remoteAddr = remoteAddr;
+        this.request = request;
+        this.response = response;
+    }
+
+    public RemotingEndPoint getRemotingEndPoint() {
+        return remotingEndPoint;
+    }
+
+    public void setRemotingEndPoint(RemotingEndPoint remotingEndPoint) {
+        this.remotingEndPoint = remotingEndPoint;
+    }
+
+    public String getRemoteAddr() {
+        return remoteAddr;
+    }
+
+    public void setRemoteAddr(String remoteAddr) {
+        this.remoteAddr = remoteAddr;
+    }
+
+    public RemotingCommand getRequest() {
+        return request;
+    }
+
+    public void setRequest(RemotingCommand request) {
+        this.request = request;
+    }
+
+    public RemotingCommand getResponse() {
+        return response;
+    }
+
+    public void setResponse(RemotingCommand response) {
+        this.response = response;
+    }
+
+    @Override
+    public String toString() {
+        return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/protocol/Protocol.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.protocol;
+
+import org.apache.rocketmq.remoting.api.channel.ChannelHandlerContextWrapper;
+
+public interface Protocol {
+    /**
+     * Minimum Viable Protocol
+     */
+    String MVP = "mvp";
+    String HTTP2 = "http2";
+    String WEBSOCKET = "websocket";
+
+    byte MVP_MAGIC = 0x14;
+    byte WEBSOCKET_MAGIC = 0x15;
+    byte HTTP_2_MAGIC = 0x16;
+
+    String name();
+
+    byte type();
+
+    void assembleHandler(ChannelHandlerContextWrapper ctx);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/protocol/ProtocolFactory.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.protocol;
+
+public interface ProtocolFactory {
+    void register(Protocol protocol);
+
+    void resetAll(Protocol protocol);
+
+    byte type(String protocolName);
+
+    Protocol get(byte type);
+
+    void clearAll();
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/serializable/Serializer.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.serializable;
+
+import java.lang.reflect.Type;
+import java.nio.ByteBuffer;
+import org.apache.rocketmq.remoting.common.TypePresentation;
+
+public interface Serializer {
+    String name();
+
+    byte type();
+
+    <T> T decode(final byte[] content, final Class<T> c);
+
+    <T> T decode(final byte[] content, final TypePresentation<T> typePresentation);
+
+    <T> T decode(final byte[] content, final Type type);
+
+    ByteBuffer encode(final Object object);
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/serializable/SerializerFactory.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.api.serializable;
+
+public interface SerializerFactory {
+    void register(Serializer serialization);
+
+    byte type(String serializationName);
+
+    Serializer get(byte type);
+
+    void clearAll();
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/common/Pair.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.common;
+
+public class Pair<L, R> {
+    private L left;
+    private R right;
+
+    public Pair(L left, R right) {
+        this.left = left;
+        this.right = right;
+    }
+
+    public L getLeft() {
+        return left;
+    }
+
+    public void setLeft(L left) {
+        this.left = left;
+    }
+
+    public R getRight() {
+        return right;
+    }
+
+    public void setRight(R right) {
+        this.right = right;
+    }
+}

remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/common/TypePresentation.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.common;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Represents a generic type {@code T}. Java doesn't yet provide a way to
+ * represent generic types, so this class does. Forces clients to create a
+ * subclass of this class which enables retrieval the type information even at
+ * runtime.
+ *
+ * <p>For example, to create a type literal for {@code List<String>}, you can
+ * create an empty anonymous inner class:
+ *
+ * <pre>
+ * TypePresentation&lt;List&lt;String&gt;&gt; list = new TypePresentation&lt;List&lt;String&gt;&gt;() {};
+ * </pre>
+ *
+ * To create a type literal for {@code Map<String, Integer>}:
+ *
+ * <pre>
+ * TypePresentation&lt;Map&lt;String, Integer&gt;&gt; map = new TypePresentation&lt;Map&lt;String, Integer&gt;&gt;() {};
+ * </pre>
+ *
+ * This syntax cannot be used to create type literals that have wildcard
+ * parameters, such as {@code Class<?>} or {@code List<? extends CharSequence>}.
+ *
+ * @since 1.0.0
+ */
+public class TypePresentation<T> {
+    static ConcurrentMap<Class<?>, ConcurrentMap<Type, ConcurrentMap<Type, Type>>> classTypeCache
+        = new ConcurrentHashMap<Class<?>, ConcurrentMap<Type, ConcurrentMap<Type, Type>>>(16, 0.75f, 1);
+    protected final Type type;
+
+    /**
+     * Constructs a new type literal. Derives represented class from type
+     * parameter.
+     *
+     * <p>Clients create an empty anonymous subclass. Doing so embeds the type
+     * parameter in the anonymous class's type hierarchy so we can reconstitute it
+     * at runtime despite erasure.
+     */
+    protected TypePresentation() {
+        Type superClass = getClass().getGenericSuperclass();
+        type = ((ParameterizedType) superClass).getActualTypeArguments()[0];
+    }
+
+    /**
+     * @return underlying {@code Type} instance.
+     */
+    public Type getType() {
+        return type;
+    }
+}

remoting-core/remoting-impl/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>remoting-core</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>remoting-impl</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.msgpack</groupId>
+            <artifactId>msgpack</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.esotericsoftware</groupId>
+            <artifactId>kryo</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-tcnative-boringssl-static</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>remoting-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>clirr-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
\ No newline at end of file

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/ChannelEventListenerGroup.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.common;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.remoting.api.channel.ChannelEventListener;
+import org.apache.rocketmq.remoting.api.channel.RemotingChannel;
+
+public class ChannelEventListenerGroup {
+    private final List<ChannelEventListener> listenerList = new ArrayList<ChannelEventListener>();
+
+    public int size() {
+        return this.listenerList.size();
+    }
+
+    public void registerChannelEventListener(final ChannelEventListener listener) {
+        if (listener != null) {
+            this.listenerList.add(listener);
+        }
+    }
+
+    public void onChannelConnect(final RemotingChannel channel) {
+        for (ChannelEventListener listener : listenerList) {
+            listener.onChannelConnect(channel);
+        }
+    }
+
+    public void onChannelClose(final RemotingChannel channel) {
+        for (ChannelEventListener listener : listenerList) {
+            listener.onChannelClose(channel);
+        }
+    }
+
+    public void onChannelException(final RemotingChannel channel) {
+        for (ChannelEventListener listener : listenerList) {
+            listener.onChannelException(channel);
+        }
+    }
+
+    public void onChannelIdle(final RemotingChannel channel) {
+        for (ChannelEventListener listener : listenerList) {
+            listener.onChannelIdle(channel);
+        }
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/RemotingCommandFactoryMeta.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.common;
+
+import org.apache.rocketmq.remoting.api.protocol.ProtocolFactory;
+import org.apache.rocketmq.remoting.api.serializable.SerializerFactory;
+import org.apache.rocketmq.remoting.impl.protocol.Httpv2Protocol;
+import org.apache.rocketmq.remoting.impl.protocol.ProtocolFactoryImpl;
+import org.apache.rocketmq.remoting.impl.protocol.serializer.MsgPackSerializer;
+import org.apache.rocketmq.remoting.impl.protocol.serializer.SerializerFactoryImpl;
+
+public class RemotingCommandFactoryMeta {
+    private final ProtocolFactory protocolFactory = new ProtocolFactoryImpl();
+    private final SerializerFactory serializerFactory = new SerializerFactoryImpl();
+    private byte protocolType = Httpv2Protocol.MVP_MAGIC;
+    private byte serializeType = MsgPackSerializer.SERIALIZER_TYPE;
+
+    public RemotingCommandFactoryMeta() {
+    }
+
+    public RemotingCommandFactoryMeta(String protocolName, String serializeName) {
+        this.protocolType = protocolFactory.type(protocolName);
+        this.serializeType = serializerFactory.type(serializeName);
+    }
+
+    public byte getSerializeType() {
+        return serializeType;
+    }
+
+    public byte getProtocolType() {
+        return protocolType;
+    }
+
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/ResponseResult.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.common;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+import org.apache.rocketmq.remoting.api.AsyncHandler;
+import org.apache.rocketmq.remoting.api.RemotingEndPoint;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+import org.apache.rocketmq.remoting.api.interceptor.ExceptionContext;
+import org.apache.rocketmq.remoting.api.interceptor.InterceptorGroup;
+import org.apache.rocketmq.remoting.api.interceptor.ResponseContext;
+
+public class ResponseResult {
+    private final long beginTimestamp = System.currentTimeMillis();
+    private final CountDownLatch countDownLatch = new CountDownLatch(1);
+    private final AtomicBoolean interceptorExecuted = new AtomicBoolean(false);
+
+    private int requestId;
+    private long timeoutMillis;
+    private AsyncHandler asyncHandler;
+
+    private volatile RemotingCommand responseCommand;
+    private volatile boolean sendRequestOK = true;
+    private volatile Throwable cause;
+    private SemaphoreReleaseOnlyOnce once;
+
+    private RemotingCommand requestCommand;
+    private InterceptorGroup interceptorGroup;
+    private String remoteAddr;
+
+    public ResponseResult(int requestId, long timeoutMillis, AsyncHandler asyncHandler, SemaphoreReleaseOnlyOnce once) {
+        this.requestId = requestId;
+        this.timeoutMillis = timeoutMillis;
+        this.asyncHandler = asyncHandler;
+        this.once = once;
+    }
+
+    public ResponseResult(int requestId, long timeoutMillis) {
+        this.requestId = requestId;
+        this.timeoutMillis = timeoutMillis;
+    }
+
+    public void executeRequestSendFailed() {
+        if (this.interceptorExecuted.compareAndSet(false, true)) {
+            try {
+                interceptorGroup.onException(new ExceptionContext(RemotingEndPoint.REQUEST, this.remoteAddr, this.requestCommand,
+                    cause, "REQUEST_SEND_FAILED"));
+            } catch (Throwable e) {
+            }
+            //Sync call
+            if (null != asyncHandler) {
+                asyncHandler.onFailure(requestCommand);
+            }
+        }
+    }
+
+    public void executeCallbackArrived(final RemotingCommand response) {
+        if (this.interceptorExecuted.compareAndSet(false, true)) {
+            try {
+                interceptorGroup.afterResponseReceived(new ResponseContext(RemotingEndPoint.REQUEST, this.remoteAddr,
+                    this.requestCommand, response));
+            } catch (Throwable e) {
+            }
+            if (null != asyncHandler) {
+                asyncHandler.onSuccess(response);
+            }
+        }
+    }
+
+    public void onTimeout(long costTimeMillis, long timoutMillis) {
+        if (this.interceptorExecuted.compareAndSet(false, true)) {
+            try {
+                interceptorGroup.onException(new ExceptionContext(RemotingEndPoint.REQUEST, this.remoteAddr, this.requestCommand,
+                    null, "CALLBACK_TIMEOUT"));
+            } catch (Throwable e) {
+            }
+            if (null != asyncHandler) {
+                asyncHandler.onTimeout(costTimeMillis, timoutMillis);
+            }
+        }
+    }
+
+    public void release() {
+        if (this.once != null) {
+            this.once.release();
+        }
+    }
+
+    public RemotingCommand waitResponse(final long timeoutMillis) {
+        try {
+            this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        return this.responseCommand;
+    }
+
+    public void putResponse(final RemotingCommand responseCommand) {
+        this.responseCommand = responseCommand;
+        this.countDownLatch.countDown();
+    }
+
+    public long getBeginTimestamp() {
+        return beginTimestamp;
+    }
+
+    public boolean isSendRequestOK() {
+        return sendRequestOK;
+    }
+
+    public void setSendRequestOK(boolean sendRequestOK) {
+        this.sendRequestOK = sendRequestOK;
+    }
+
+    public long getTimeoutMillis() {
+        return timeoutMillis;
+    }
+
+    public AsyncHandler getAsyncHandler() {
+        return asyncHandler;
+    }
+
+    public Throwable getCause() {
+        return cause;
+    }
+
+    public void setCause(Throwable cause) {
+        this.cause = cause;
+    }
+
+    public RemotingCommand getResponseCommand() {
+        return responseCommand;
+    }
+
+    public void setResponseCommand(RemotingCommand responseCommand) {
+        this.responseCommand = responseCommand;
+    }
+
+    public int getRequestId() {
+        return requestId;
+    }
+
+    public RemotingCommand getRequestCommand() {
+        return requestCommand;
+    }
+
+    public void setRequestCommand(RemotingCommand requestCommand) {
+        this.requestCommand = requestCommand;
+    }
+
+    public InterceptorGroup getInterceptorGroup() {
+        return interceptorGroup;
+    }
+
+    public void setInterceptorGroup(InterceptorGroup interceptorGroup) {
+        this.interceptorGroup = interceptorGroup;
+    }
+
+    public String getRemoteAddr() {
+        return remoteAddr;
+    }
+
+    public void setRemoteAddr(String remoteAddr) {
+        this.remoteAddr = remoteAddr;
+    }
+
+    @Override
+    public String toString() {
+        return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/SemaphoreReleaseOnlyOnce.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.common;
+
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class SemaphoreReleaseOnlyOnce {
+    private final AtomicBoolean released = new AtomicBoolean(false);
+    private final Semaphore semaphore;
+
+    public SemaphoreReleaseOnlyOnce(Semaphore semaphore) {
+        this.semaphore = semaphore;
+    }
+
+    public void release() {
+        if (this.released.compareAndSet(false, true)) {
+            this.semaphore.release();
+        }
+    }
+
+    public Semaphore getSemaphore() {
+        return semaphore;
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/metrics/ChannelMetrics.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.common.metrics;
+
+import io.netty.channel.group.ChannelGroup;
+
+public interface ChannelMetrics {
+
+    Integer getChannelCount();
+
+    ChannelGroup getChannels();
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/config/RemotingConfig.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.config;
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+import org.apache.rocketmq.remoting.api.protocol.Protocol;
+import org.apache.rocketmq.remoting.impl.protocol.compression.GZipCompressor;
+import org.apache.rocketmq.remoting.impl.protocol.serializer.MsgPackSerializer;
+
+public class RemotingConfig extends TcpSocketConfig {
+    private int connectionMaxRetries = 3;
+    private int connectionChannelReaderIdleSeconds = 0;
+    private int connectionChannelWriterIdleSeconds = 0;
+    /**
+     * IdleStateEvent will be triggered when neither read nor write was
+     * performed for the specified period of this time. Specify {@code 0} to
+     * disable
+     */
+    private int connectionChannelIdleSeconds = 120;
+    private int writeBufLowWaterMark = 32 * 10240;
+    private int writeBufHighWaterMark = 64 * 10240;
+    private int threadTaskLowWaterMark = 30000;
+    private int threadTaskHighWaterMark = 50000;
+    private int connectionRetryBackoffMillis = 3000;
+    private String protocolName = Protocol.MVP;
+    private String serializerName = MsgPackSerializer.SERIALIZER_NAME;
+    private String compressorName = GZipCompressor.COMPRESSOR_NAME;
+    private int serviceThreadBlockQueueSize = 50000;
+    private boolean clientNativeEpollEnable = false;
+    private int clientWorkerThreads = 16 + Runtime.getRuntime().availableProcessors() * 2;
+    private int clientConnectionFutureAwaitTimeoutMillis = 30000;
+    private int clientAsyncCallbackExecutorThreads = 16 + Runtime.getRuntime().availableProcessors() * 2;
+    private int clientOnewayInvokeSemaphore = 20480;
+
+    //=============Server configuration==================
+    private int clientAsyncInvokeSemaphore = 20480;
+    private boolean clientPooledBytebufAllocatorEnable = false;
+    private boolean clientCloseSocketIfTimeout = true;
+    private boolean clientShortConnectionEnable = false;
+    private long clientPublishServiceTimeout = 10000;
+    private long clientConsumerServiceTimeout = 10000;
+    private long clientInvokeServiceTimeout = 10000;
+    private int clientMaxRetryCount = 10;
+    private int clientSleepBeforeRetry = 100;
+    private int serverListenPort = 8888;
+    /**
+     * If server only listened 1 port,recommend to set the value to 1
+     */
+    private int serverAcceptorThreads = 1;
+    private int serverIoThreads = 16 + Runtime.getRuntime().availableProcessors() * 2;
+    private int serverWorkerThreads = 16 + Runtime.getRuntime().availableProcessors() * 2;
+    private int serverOnewayInvokeSemaphore = 256;
+    private int serverAsyncInvokeSemaphore = 6400;
+    private boolean serverNativeEpollEnable = false;
+    private int serverAsyncCallbackExecutorThreads = Runtime.getRuntime().availableProcessors() * 2;
+    private boolean serverPooledBytebufAllocatorEnable = true;
+    private boolean serverAuthOpenEnable = true;
+
+    @Override
+    public String toString() {
+        return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
+    }
+
+    public int getConnectionMaxRetries() {
+        return connectionMaxRetries;
+    }
+
+    public void setConnectionMaxRetries(final int connectionMaxRetries) {
+        this.connectionMaxRetries = connectionMaxRetries;
+    }
+
+    public int getConnectionChannelReaderIdleSeconds() {
+        return connectionChannelReaderIdleSeconds;
+    }
+
+    public void setConnectionChannelReaderIdleSeconds(final int connectionChannelReaderIdleSeconds) {
+        this.connectionChannelReaderIdleSeconds = connectionChannelReaderIdleSeconds;
+    }
+
+    public int getConnectionChannelWriterIdleSeconds() {
+        return connectionChannelWriterIdleSeconds;
+    }
+
+    public void setConnectionChannelWriterIdleSeconds(final int connectionChannelWriterIdleSeconds) {
+        this.connectionChannelWriterIdleSeconds = connectionChannelWriterIdleSeconds;
+    }
+
+    public int getConnectionChannelIdleSeconds() {
+        return connectionChannelIdleSeconds;
+    }
+
+    public void setConnectionChannelIdleSeconds(final int connectionChannelIdleSeconds) {
+        this.connectionChannelIdleSeconds = connectionChannelIdleSeconds;
+    }
+
+    public int getWriteBufLowWaterMark() {
+        return writeBufLowWaterMark;
+    }
+
+    public void setWriteBufLowWaterMark(final int writeBufLowWaterMark) {
+        this.writeBufLowWaterMark = writeBufLowWaterMark;
+    }
+
+    public int getWriteBufHighWaterMark() {
+        return writeBufHighWaterMark;
+    }
+
+    public void setWriteBufHighWaterMark(final int writeBufHighWaterMark) {
+        this.writeBufHighWaterMark = writeBufHighWaterMark;
+    }
+
+    public int getThreadTaskLowWaterMark() {
+        return threadTaskLowWaterMark;
+    }
+
+    public void setThreadTaskLowWaterMark(final int threadTaskLowWaterMark) {
+        this.threadTaskLowWaterMark = threadTaskLowWaterMark;
+    }
+
+    public int getThreadTaskHighWaterMark() {
+        return threadTaskHighWaterMark;
+    }
+
+    public void setThreadTaskHighWaterMark(final int threadTaskHighWaterMark) {
+        this.threadTaskHighWaterMark = threadTaskHighWaterMark;
+    }
+
+    public int getConnectionRetryBackoffMillis() {
+        return connectionRetryBackoffMillis;
+    }
+
+    public void setConnectionRetryBackoffMillis(final int connectionRetryBackoffMillis) {
+        this.connectionRetryBackoffMillis = connectionRetryBackoffMillis;
+    }
+
+    public String getProtocolName() {
+        return protocolName;
+    }
+
+    public void setProtocolName(final String protocolName) {
+        this.protocolName = protocolName;
+    }
+
+    public String getSerializerName() {
+        return serializerName;
+    }
+
+    public void setSerializerName(final String serializerName) {
+        this.serializerName = serializerName;
+    }
+
+    public String getCompressorName() {
+        return compressorName;
+    }
+
+    public void setCompressorName(final String compressorName) {
+        this.compressorName = compressorName;
+    }
+
+    public int getServiceThreadBlockQueueSize() {
+        return serviceThreadBlockQueueSize;
+    }
+
+    public void setServiceThreadBlockQueueSize(final int serviceThreadBlockQueueSize) {
+        this.serviceThreadBlockQueueSize = serviceThreadBlockQueueSize;
+    }
+
+    public boolean isClientNativeEpollEnable() {
+        return clientNativeEpollEnable;
+    }
+
+    public void setClientNativeEpollEnable(final boolean clientNativeEpollEnable) {
+        this.clientNativeEpollEnable = clientNativeEpollEnable;
+    }
+
+    public int getClientWorkerThreads() {
+        return clientWorkerThreads;
+    }
+
+    public void setClientWorkerThreads(final int clientWorkerThreads) {
+        this.clientWorkerThreads = clientWorkerThreads;
+    }
+
+    public int getClientConnectionFutureAwaitTimeoutMillis() {
+        return clientConnectionFutureAwaitTimeoutMillis;
+    }
+
+    public void setClientConnectionFutureAwaitTimeoutMillis(final int clientConnectionFutureAwaitTimeoutMillis) {
+        this.clientConnectionFutureAwaitTimeoutMillis = clientConnectionFutureAwaitTimeoutMillis;
+    }
+
+    public int getClientAsyncCallbackExecutorThreads() {
+        return clientAsyncCallbackExecutorThreads;
+    }
+
+    public void setClientAsyncCallbackExecutorThreads(final int clientAsyncCallbackExecutorThreads) {
+        this.clientAsyncCallbackExecutorThreads = clientAsyncCallbackExecutorThreads;
+    }
+
+    public int getClientOnewayInvokeSemaphore() {
+        return clientOnewayInvokeSemaphore;
+    }
+
+    public void setClientOnewayInvokeSemaphore(final int clientOnewayInvokeSemaphore) {
+        this.clientOnewayInvokeSemaphore = clientOnewayInvokeSemaphore;
+    }
+
+    public int getClientAsyncInvokeSemaphore() {
+        return clientAsyncInvokeSemaphore;
+    }
+
+    public void setClientAsyncInvokeSemaphore(final int clientAsyncInvokeSemaphore) {
+        this.clientAsyncInvokeSemaphore = clientAsyncInvokeSemaphore;
+    }
+
+    public boolean isClientPooledBytebufAllocatorEnable() {
+        return clientPooledBytebufAllocatorEnable;
+    }
+
+    public void setClientPooledBytebufAllocatorEnable(final boolean clientPooledBytebufAllocatorEnable) {
+        this.clientPooledBytebufAllocatorEnable = clientPooledBytebufAllocatorEnable;
+    }
+
+    public boolean isClientCloseSocketIfTimeout() {
+        return clientCloseSocketIfTimeout;
+    }
+
+    public void setClientCloseSocketIfTimeout(final boolean clientCloseSocketIfTimeout) {
+        this.clientCloseSocketIfTimeout = clientCloseSocketIfTimeout;
+    }
+
+    public boolean isClientShortConnectionEnable() {
+        return clientShortConnectionEnable;
+    }
+
+    public void setClientShortConnectionEnable(final boolean clientShortConnectionEnable) {
+        this.clientShortConnectionEnable = clientShortConnectionEnable;
+    }
+
+    public long getClientPublishServiceTimeout() {
+        return clientPublishServiceTimeout;
+    }
+
+    public void setClientPublishServiceTimeout(final long clientPublishServiceTimeout) {
+        this.clientPublishServiceTimeout = clientPublishServiceTimeout;
+    }
+
+    public long getClientConsumerServiceTimeout() {
+        return clientConsumerServiceTimeout;
+    }
+
+    public void setClientConsumerServiceTimeout(final long clientConsumerServiceTimeout) {
+        this.clientConsumerServiceTimeout = clientConsumerServiceTimeout;
+    }
+
+    public long getClientInvokeServiceTimeout() {
+        return clientInvokeServiceTimeout;
+    }
+
+    public void setClientInvokeServiceTimeout(final long clientInvokeServiceTimeout) {
+        this.clientInvokeServiceTimeout = clientInvokeServiceTimeout;
+    }
+
+    public int getClientMaxRetryCount() {
+        return clientMaxRetryCount;
+    }
+
+    public void setClientMaxRetryCount(final int clientMaxRetryCount) {
+        this.clientMaxRetryCount = clientMaxRetryCount;
+    }
+
+    public int getClientSleepBeforeRetry() {
+        return clientSleepBeforeRetry;
+    }
+
+    public void setClientSleepBeforeRetry(final int clientSleepBeforeRetry) {
+        this.clientSleepBeforeRetry = clientSleepBeforeRetry;
+    }
+
+    public int getServerListenPort() {
+        return serverListenPort;
+    }
+
+    public void setServerListenPort(final int serverListenPort) {
+        this.serverListenPort = serverListenPort;
+    }
+
+    public int getServerAcceptorThreads() {
+        return serverAcceptorThreads;
+    }
+
+    public void setServerAcceptorThreads(final int serverAcceptorThreads) {
+        this.serverAcceptorThreads = serverAcceptorThreads;
+    }
+
+    public int getServerIoThreads() {
+        return serverIoThreads;
+    }
+
+    public void setServerIoThreads(final int serverIoThreads) {
+        this.serverIoThreads = serverIoThreads;
+    }
+
+    public int getServerWorkerThreads() {
+        return serverWorkerThreads;
+    }
+
+    public void setServerWorkerThreads(final int serverWorkerThreads) {
+        this.serverWorkerThreads = serverWorkerThreads;
+    }
+
+    public int getServerOnewayInvokeSemaphore() {
+        return serverOnewayInvokeSemaphore;
+    }
+
+    public void setServerOnewayInvokeSemaphore(final int serverOnewayInvokeSemaphore) {
+        this.serverOnewayInvokeSemaphore = serverOnewayInvokeSemaphore;
+    }
+
+    public int getServerAsyncInvokeSemaphore() {
+        return serverAsyncInvokeSemaphore;
+    }
+
+    public void setServerAsyncInvokeSemaphore(final int serverAsyncInvokeSemaphore) {
+        this.serverAsyncInvokeSemaphore = serverAsyncInvokeSemaphore;
+    }
+
+    public boolean isServerNativeEpollEnable() {
+        return serverNativeEpollEnable;
+    }
+
+    public void setServerNativeEpollEnable(final boolean serverNativeEpollEnable) {
+        this.serverNativeEpollEnable = serverNativeEpollEnable;
+    }
+
+    public int getServerAsyncCallbackExecutorThreads() {
+        return serverAsyncCallbackExecutorThreads;
+    }
+
+    public void setServerAsyncCallbackExecutorThreads(final int serverAsyncCallbackExecutorThreads) {
+        this.serverAsyncCallbackExecutorThreads = serverAsyncCallbackExecutorThreads;
+    }
+
+    public boolean isServerPooledBytebufAllocatorEnable() {
+        return serverPooledBytebufAllocatorEnable;
+    }
+
+    public void setServerPooledBytebufAllocatorEnable(final boolean serverPooledBytebufAllocatorEnable) {
+        this.serverPooledBytebufAllocatorEnable = serverPooledBytebufAllocatorEnable;
+    }
+
+    public boolean isServerAuthOpenEnable() {
+        return serverAuthOpenEnable;
+    }
+
+    public void setServerAuthOpenEnable(final boolean serverAuthOpenEnable) {
+        this.serverAuthOpenEnable = serverAuthOpenEnable;
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/config/TcpSocketConfig.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.config;
+
+/**
+ * TCP socket configuration
+ *
+ * @see java.net.SocketOptions
+ */
+public class TcpSocketConfig {
+    private boolean tcpSoReuseAddress;
+    private boolean tcpSoKeepAlive;
+    private boolean tcpSoNoDelay;
+    private int tcpSoSndBufSize;  // see /proc/sys/net/ipv4/tcp_rmem
+    private int tcpSoRcvBufSize;  // see /proc/sys/net/ipv4/tcp_wmem
+    private int tcpSoBacklogSize;
+    private int tcpSoLinger;
+    private int tcpSoTimeout;
+
+    public boolean isTcpSoReuseAddress() {
+        return tcpSoReuseAddress;
+    }
+
+    public void setTcpSoReuseAddress(final boolean tcpSoReuseAddress) {
+        this.tcpSoReuseAddress = tcpSoReuseAddress;
+    }
+
+    public boolean isTcpSoKeepAlive() {
+        return tcpSoKeepAlive;
+    }
+
+    public void setTcpSoKeepAlive(final boolean tcpSoKeepAlive) {
+        this.tcpSoKeepAlive = tcpSoKeepAlive;
+    }
+
+    public boolean isTcpSoNoDelay() {
+        return tcpSoNoDelay;
+    }
+
+    public void setTcpSoNoDelay(final boolean tcpSoNoDelay) {
+        this.tcpSoNoDelay = tcpSoNoDelay;
+    }
+
+    public int getTcpSoSndBufSize() {
+        return tcpSoSndBufSize;
+    }
+
+    public void setTcpSoSndBufSize(final int tcpSoSndBufSize) {
+        this.tcpSoSndBufSize = tcpSoSndBufSize;
+    }
+
+    public int getTcpSoRcvBufSize() {
+        return tcpSoRcvBufSize;
+    }
+
+    public void setTcpSoRcvBufSize(final int tcpSoRcvBufSize) {
+        this.tcpSoRcvBufSize = tcpSoRcvBufSize;
+    }
+
+    public int getTcpSoBacklogSize() {
+        return tcpSoBacklogSize;
+    }
+
+    public void setTcpSoBacklogSize(final int tcpSoBacklogSize) {
+        this.tcpSoBacklogSize = tcpSoBacklogSize;
+    }
+
+    public int getTcpSoLinger() {
+        return tcpSoLinger;
+    }
+
+    public void setTcpSoLinger(final int tcpSoLinger) {
+        this.tcpSoLinger = tcpSoLinger;
+    }
+
+    public int getTcpSoTimeout() {
+        return tcpSoTimeout;
+    }
+
+    public void setTcpSoTimeout(final int tcpSoTimeout) {
+        this.tcpSoTimeout = tcpSoTimeout;
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/external/ThreadUtils.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.external;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class ThreadUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(ThreadUtils.class);
+
+    /**
+     * A constructor to stop this class being constructed.
+     */
+    private ThreadUtils() {
+        // Unused
+
+    }
+
+    public static ExecutorService newThreadPoolExecutor(int corePoolSize,
+        int maximumPoolSize,
+        long keepAliveTime,
+        TimeUnit unit,
+        BlockingQueue<Runnable> workQueue, String processName, boolean isDaemon) {
+        return new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, newThreadFactory(processName, isDaemon));
+    }
+
+    public static ExecutorService newSingleThreadExecutor(String processName, boolean isDaemon) {
+        return Executors.newSingleThreadExecutor(newThreadFactory(processName, isDaemon));
+    }
+
+    public static ScheduledExecutorService newSingleThreadScheduledExecutor(String processName, boolean isDaemon) {
+        return Executors.newSingleThreadScheduledExecutor(newThreadFactory(processName, isDaemon));
+    }
+
+    public static ScheduledExecutorService newFixedThreadScheduledPool(int nThreads, String processName,
+        boolean isDaemon) {
+        return Executors.newScheduledThreadPool(nThreads, newThreadFactory(processName, isDaemon));
+    }
+
+    public static ThreadFactory newThreadFactory(String processName, boolean isDaemon) {
+        return newGenericThreadFactory("Remoting-" + processName, isDaemon);
+    }
+
+    public static ThreadFactory newGenericThreadFactory(String processName) {
+        return newGenericThreadFactory(processName, false);
+    }
+
+    public static ThreadFactory newGenericThreadFactory(String processName, int threads) {
+        return newGenericThreadFactory(processName, threads, false);
+    }
+
+    public static ThreadFactory newGenericThreadFactory(final String processName, final boolean isDaemon) {
+        return new ThreadFactory() {
+            private AtomicInteger threadIndex = new AtomicInteger(0);
+
+            @Override
+            public Thread newThread(Runnable r) {
+                Thread thread = new Thread(r, String.format("%s_%d", processName, this.threadIndex.incrementAndGet()));
+                thread.setDaemon(isDaemon);
+                return thread;
+            }
+        };
+    }
+
+    public static ThreadFactory newGenericThreadFactory(final String processName, final int threads,
+        final boolean isDaemon) {
+        return new ThreadFactory() {
+            private AtomicInteger threadIndex = new AtomicInteger(0);
+
+            @Override
+            public Thread newThread(Runnable r) {
+                Thread thread = new Thread(r, String.format("%s_%d_%d", processName, threads, this.threadIndex.incrementAndGet()));
+                thread.setDaemon(isDaemon);
+                return thread;
+            }
+        };
+    }
+
+    /**
+     * Create a new thread
+     *
+     * @param name The name of the thread
+     * @param runnable The work for the thread to do
+     * @param daemon Should the thread block JVM stop?
+     * @return The unstarted thread
+     */
+    public static Thread newThread(String name, Runnable runnable, boolean daemon) {
+        Thread thread = new Thread(runnable, name);
+        thread.setDaemon(daemon);
+        thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+            public void uncaughtException(Thread t, Throwable e) {
+                LOG.error("Uncaught exception in thread '" + t.getName() + "':", e);
+            }
+        });
+        return thread;
+    }
+
+    /**
+     * Shutdown passed thread using isAlive and join.
+     *
+     * @param t Thread to stop
+     */
+    public static void shutdownGracefully(final Thread t) {
+        shutdownGracefully(t, 0);
+    }
+
+    /**
+     * Shutdown passed thread using isAlive and join.
+     *
+     * @param millis Pass 0 if we're to wait forever.
+     * @param t Thread to stop
+     */
+    public static void shutdownGracefully(final Thread t, final long millis) {
+        if (t == null)
+            return;
+        while (t.isAlive()) {
+            try {
+                t.interrupt();
+                t.join(millis);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
+    /**
+     * An implementation of the graceful stop sequence recommended by
+     * {@link ExecutorService}.
+     *
+     * @param executor executor
+     * @param timeout timeout
+     * @param timeUnit timeUnit
+     */
+    public static void shutdownGracefully(ExecutorService executor, long timeout, TimeUnit timeUnit) {
+        // Disable new tasks from being submitted.
+        executor.shutdown();
+        try {
+            // Wait a while for existing tasks to terminate.
+            if (!executor
+                .awaitTermination(timeout, timeUnit)) {
+                executor.shutdownNow();
+                // Wait a while for tasks to respond to being cancelled.
+                if (!executor.awaitTermination(timeout, timeUnit)) {
+                    LOG.warn(String.format("%s didn't terminate!", executor));
+                }
+            }
+        } catch (InterruptedException ie) {
+            // (Re-)Cancel if current thread also interrupted.
+            executor.shutdownNow();
+            // Preserve interrupt status.
+            Thread.currentThread().interrupt();
+        }
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/buffer/NettyByteBufferWrapper.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.buffer;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import java.nio.ByteBuffer;
+import org.apache.rocketmq.remoting.api.buffer.ByteBufferWrapper;
+
+public class NettyByteBufferWrapper implements ByteBufferWrapper {
+    private final ByteBuf buffer;
+    private final Channel channel;
+
+    public NettyByteBufferWrapper(ByteBuf buffer) {
+        this(buffer, null);
+    }
+
+    public NettyByteBufferWrapper(ByteBuf buffer, Channel channel) {
+        this.channel = channel;
+        this.buffer = buffer;
+    }
+
+    public void writeByte(int index, byte data) {
+        buffer.writeByte(data);
+    }
+
+    public void writeByte(byte data) {
+        buffer.writeByte(data);
+    }
+
+    public byte readByte() {
+        return buffer.readByte();
+    }
+
+    public void writeInt(int data) {
+        buffer.writeInt(data);
+    }
+
+    public void writeBytes(byte[] data) {
+        buffer.writeBytes(data);
+    }
+
+    @Override
+    public void writeBytes(final ByteBuffer data) {
+        buffer.writeBytes(data);
+    }
+
+    public int readableBytes() {
+        return buffer.readableBytes();
+    }
+
+    public int readInt() {
+        return buffer.readInt();
+    }
+
+    public void readBytes(byte[] dst) {
+        buffer.readBytes(dst);
+    }
+
+    @Override
+    public void readBytes(final ByteBuffer dst) {
+        buffer.readBytes(dst);
+    }
+
+    public int readerIndex() {
+        return buffer.readerIndex();
+    }
+
+    public void setReaderIndex(int index) {
+        buffer.setIndex(index, buffer.writerIndex());
+    }
+
+    @Override
+    public void writeLong(long value) {
+        buffer.writeLong(value);
+    }
+
+    @Override
+    public long readLong() {
+        return buffer.readLong();
+    }
+
+    @Override
+    public void ensureCapacity(int capacity) {
+        buffer.capacity(capacity);
+    }
+
+    @Override
+    public short readShort() {
+        return buffer.readShort();
+    }
+
+    @Override
+    public void writeShort(final short value) {
+        buffer.writeShort(value);
+    }
+}
+
+

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/channel/ChannelHandlerContextWrapperImpl.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.channel;
+
+import org.apache.rocketmq.remoting.api.channel.ChannelHandlerContextWrapper;
+
+public class ChannelHandlerContextWrapperImpl<ChannelHandlerContext> implements ChannelHandlerContextWrapper {
+
+    private io.netty.channel.ChannelHandlerContext context;
+
+    public ChannelHandlerContextWrapperImpl(io.netty.channel.ChannelHandlerContext context) {
+        this.context = context;
+    }
+
+    public io.netty.channel.ChannelHandlerContext getContext() {
+        return context;
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/channel/FileRegionImpl.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.channel;
+
+import io.netty.channel.FileRegion;
+import io.netty.util.AbstractReferenceCounted;
+import java.io.IOException;
+import java.nio.channels.WritableByteChannel;
+import org.apache.rocketmq.remoting.api.channel.ChunkRegion;
+
+public class FileRegionImpl extends AbstractReferenceCounted implements FileRegion {
+    private final ChunkRegion chunkRegion;
+
+    public FileRegionImpl(ChunkRegion chunkRegion) {
+        this.chunkRegion = chunkRegion;
+    }
+
+    @Override
+    public long position() {
+        return chunkRegion.position();
+    }
+
+    @Override
+    public long transfered() {
+        return chunkRegion.transferred();
+    }
+
+    @Override
+    public long transferred() {
+        return chunkRegion.transferred();
+    }
+
+    @Override
+    public long count() {
+        return chunkRegion.count();
+    }
+
+    @Override
+    public long transferTo(WritableByteChannel target, long position) throws IOException {
+        return chunkRegion.transferTo(target, position);
+    }
+
+    @Override
+    protected void deallocate() {
+        chunkRegion.release();
+    }
+
+    @Override
+    public FileRegion retain() {
+        super.retain();
+        return this;
+    }
+
+    @Override
+    public FileRegion retain(int increment) {
+        super.retain(increment);
+        return this;
+    }
+
+    @Override
+    public FileRegion touch() {
+        return this;
+    }
+
+    @Override
+    public FileRegion touch(Object hint) {
+        return this;
+    }
+
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/channel/NettyChannelImpl.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.channel;
+
+import io.netty.channel.Channel;
+import java.net.SocketAddress;
+import org.apache.rocketmq.remoting.api.channel.ChunkRegion;
+import org.apache.rocketmq.remoting.api.channel.RemotingChannel;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+
+public class NettyChannelImpl implements RemotingChannel {
+    private final io.netty.channel.Channel channel;
+
+    public NettyChannelImpl(Channel channel) {
+        this.channel = channel;
+    }
+
+    @Override
+    public SocketAddress localAddress() {
+        return channel.localAddress();
+    }
+
+    @Override
+    public SocketAddress remoteAddress() {
+        return channel.remoteAddress();
+    }
+
+    @Override
+    public boolean isWritable() {
+        return channel.isWritable();
+    }
+
+    @Override
+    public boolean isActive() {
+        return channel.isActive();
+    }
+
+    @Override
+    public void close() {
+        channel.close();
+    }
+
+    @Override
+    public void reply(final RemotingCommand command) {
+        channel.writeAndFlush(command);
+    }
+
+    @Override
+    public void reply(final ChunkRegion fileRegion) {
+        channel.writeAndFlush(fileRegion);
+    }
+
+    public io.netty.channel.Channel getChannel() {
+        return channel;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o)
+            return true;
+        if (o == null || getClass() != o.getClass())
+            return false;
+
+        final NettyChannelImpl that = (NettyChannelImpl) o;
+
+        return channel != null ? channel.equals(that.channel) : that.channel == null;
+
+    }
+
+    @Override
+    public int hashCode() {
+        return channel != null ? channel.hashCode() : 0;
+    }
+
+    @Override
+    public String toString() {
+        return "NettyChannelImpl [channel=" + channel + "]";
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/command/CodecHelper.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.command;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Map.Entry;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+import org.apache.rocketmq.remoting.api.command.TrafficType;
+import org.apache.rocketmq.remoting.api.exception.RemoteCodecException;
+
+public class CodecHelper {
+    //ProtocolType + TotalLength + RequestId + SerializeType + TrafficType + CodeLength + RemarkLength + PropertiesSize + ParameterLength
+    public final static int MIN_PROTOCOL_LEN = 1 + 4 + 4 + 1 + 1 + 2 + 2 + 2 + 4;
+    public final static char PROPERTY_SEPARATOR = '\n';
+    public final static Charset REMOTING_CHARSET = Charset.forName("UTF-8");
+
+    public final static int CODE_MAX_LEN = 512;
+    public final static int PARAMETER_MAX_LEN = 33554432;
+    public final static int BODY_MAX_LEN = 33554432;
+    public final static int PACKET_MAX_LEN = 33554432;
+
+    public static ByteBuffer encodeHeader(final RemotingCommand command, final int parameterLength,
+        final int extraPayload) {
+        byte[] code = command.opCode().getBytes(REMOTING_CHARSET);
+        int codeLength = code.length;
+
+        byte[] remark = command.remark().getBytes(REMOTING_CHARSET);
+        int remarkLength = remark.length;
+
+        byte[][] props = null;
+        int propsLength = 0;
+        StringBuilder sb = new StringBuilder();
+        if (!command.properties().isEmpty()) {
+            props = new byte[command.properties().size()][];
+            int i = 0;
+            for (Entry<String, String> next : command.properties().entrySet()) {
+                sb.setLength(0);
+                sb.append(next.getKey());
+                sb.append(PROPERTY_SEPARATOR);
+                sb.append(next.getValue());
+
+                props[i] = sb.toString().getBytes(REMOTING_CHARSET);
+
+                propsLength += 2;
+                propsLength += props[i].length;
+                i++;
+            }
+        }
+
+        int totalLength = MIN_PROTOCOL_LEN - 1 - 4
+            + codeLength
+            + remarkLength
+            + propsLength
+            + parameterLength
+            + extraPayload;
+
+        int headerLength = 1 + 4 + totalLength - parameterLength - extraPayload;
+
+        ByteBuffer buf = ByteBuffer.allocate(headerLength);
+        buf.put(command.protocolType());
+        buf.putInt(totalLength);
+        buf.putInt(command.requestID());
+        buf.put(command.serializerType());
+        buf.put((byte) command.trafficType().ordinal());
+
+        buf.putShort((short) codeLength);
+        if (codeLength > 0) {
+            buf.put(code);
+        }
+        buf.putShort((short) remarkLength);
+        if (remarkLength > 0) {
+            buf.put(remark);
+        }
+        if (props != null) {
+            buf.putShort((short) props.length);
+            for (byte[] prop : props) {
+                buf.putShort((short) prop.length);
+                buf.put(prop);
+            }
+        } else {
+            buf.putShort((short) 0);
+        }
+
+        buf.putInt(parameterLength);
+
+        buf.flip();
+
+        return buf;
+    }
+
+    public static RemotingCommand decode(final ByteBuffer byteBuffer) {
+        RemotingCommandImpl cmd = new RemotingCommandImpl();
+        int totalLength = byteBuffer.limit();
+        cmd.requestID(byteBuffer.getInt());
+        cmd.serializerType(byteBuffer.get());
+        cmd.trafficType(TrafficType.parse(byteBuffer.get()));
+
+        {
+            short size = byteBuffer.getShort();
+            if (size > 0 && size <= CODE_MAX_LEN) {
+                byte[] bytes = new byte[size];
+                byteBuffer.get(bytes);
+                String str = new String(bytes, REMOTING_CHARSET);
+                cmd.opCode(str);
+            } else {
+                throw new RemoteCodecException(String.format("Code length: %d over max limit: %d", size, CODE_MAX_LEN));
+            }
+        }
+
+        {
+            short size = byteBuffer.getShort();
+            if (size > 0) {
+                byte[] bytes = new byte[size];
+                byteBuffer.get(bytes);
+                String str = new String(bytes, REMOTING_CHARSET);
+                cmd.remark(str);
+            }
+        }
+
+        {
+            short size = byteBuffer.getShort();
+            if (size > 0) {
+                for (int i = 0; i < size; i++) {
+                    short length = byteBuffer.getShort();
+                    if (length > 0) {
+                        byte[] bytes = new byte[length];
+                        byteBuffer.get(bytes);
+                        String str = new String(bytes, REMOTING_CHARSET);
+                        int index = str.indexOf(PROPERTY_SEPARATOR);
+                        if (index > 0) {
+                            String key = str.substring(0, index);
+                            String value = str.substring(index + 1);
+                            cmd.property(key, value);
+                        }
+                    }
+                }
+            }
+        }
+
+        {
+            int size = byteBuffer.getInt();
+            if (size > 0 && size <= PARAMETER_MAX_LEN) {
+                byte[] bytes = new byte[size];
+                byteBuffer.get(bytes);
+                cmd.parameterBytes(bytes);
+            } else if (size != 0) {
+                throw new RemoteCodecException(String.format("Parameter size: %d over max limit: %d", size, PARAMETER_MAX_LEN));
+            }
+        }
+
+        {
+            int size = totalLength - byteBuffer.position();
+            if (size > 0 && size <= BODY_MAX_LEN) {
+                byte[] bytes = new byte[size];
+                byteBuffer.get(bytes);
+                cmd.extraPayload(bytes);
+            } else if (size != 0) {
+                throw new RemoteCodecException(String.format("Body size: %d over max limit: %d", size, BODY_MAX_LEN));
+            }
+        }
+
+        return cmd;
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/command/RemotingCommandFactoryImpl.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.command;
+
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+import org.apache.rocketmq.remoting.api.command.RemotingCommandFactory;
+import org.apache.rocketmq.remoting.api.command.TrafficType;
+import org.apache.rocketmq.remoting.common.RemotingCommandFactoryMeta;
+
+public class RemotingCommandFactoryImpl implements RemotingCommandFactory {
+    private RemotingCommandFactoryMeta remotingCommandFactoryMeta;
+
+    public RemotingCommandFactoryImpl() {
+        this(new RemotingCommandFactoryMeta());
+    }
+
+    public RemotingCommandFactoryImpl(final RemotingCommandFactoryMeta remotingCommandFactoryMeta) {
+        this.remotingCommandFactoryMeta = remotingCommandFactoryMeta;
+    }
+
+    @Override
+    public RemotingCommand createRequest() {
+        RemotingCommand request = new RemotingCommandImpl();
+        request.protocolType(this.remotingCommandFactoryMeta.getProtocolType());
+        request.serializerType(this.remotingCommandFactoryMeta.getSerializeType());
+        return request;
+    }
+
+    @Override
+    public RemotingCommand createResponse(final RemotingCommand command) {
+        RemotingCommand response = new RemotingCommandImpl();
+        response.requestID(command.requestID());
+        response.protocolType(command.protocolType());
+        response.serializerType(command.serializerType());
+        response.trafficType(TrafficType.RESPONSE);
+        return response;
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/command/RemotingCommandImpl.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.command;
+
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+import org.apache.rocketmq.remoting.api.command.RemotingCommand;
+import org.apache.rocketmq.remoting.api.command.TrafficType;
+import org.apache.rocketmq.remoting.api.serializable.SerializerFactory;
+import org.apache.rocketmq.remoting.common.TypePresentation;
+
+public class RemotingCommandImpl implements RemotingCommand {
+    public final static RequestIdGenerator REQUEST_ID_GENERATOR = RequestIdGenerator.inst;
+
+    private byte protocolType;
+    private byte serializeType;
+
+    private volatile int requestId = REQUEST_ID_GENERATOR.incrementAndGet();
+    private TrafficType trafficType = TrafficType.REQUEST_SYNC;
+    private String code = CommandFlag.SUCCESS.flag();
+    private String remark = "";
+    private Map<String, String> properties = new HashMap<String, String>();
+    private Object parameter;
+    private byte[] extraPayload;
+
+    private byte[] parameterByte;
+
+    protected RemotingCommandImpl() {
+    }
+
+    @Override
+    public byte protocolType() {
+        return this.protocolType;
+    }
+
+    @Override
+    public void protocolType(byte value) {
+        this.protocolType = value;
+    }
+
+    @Override
+    public int requestID() {
+        return requestId;
+    }
+
+    @Override
+    public void requestID(int value) {
+        this.requestId = value;
+    }
+
+    @Override
+    public byte serializerType() {
+        return this.serializeType;
+    }
+
+    @Override
+    public void serializerType(byte value) {
+        this.serializeType = value;
+    }
+
+    @Override
+    public TrafficType trafficType() {
+        return this.trafficType;
+    }
+
+    @Override
+    public void trafficType(TrafficType value) {
+        this.trafficType = value;
+    }
+
+    @Override
+    public String opCode() {
+        return this.code;
+    }
+
+    @Override
+    public void opCode(String value) {
+        this.code = value;
+    }
+
+    @Override
+    public String remark() {
+        return this.remark;
+    }
+
+    @Override
+    public void remark(String value) {
+        this.remark = value;
+    }
+
+    @Override
+    public Map<String, String> properties() {
+        return this.properties;
+    }
+
+    @Override
+    public void properties(Map<String, String> value) {
+        this.properties = value;
+    }
+
+    @Override
+    public String property(String key) {
+        return this.properties.get(key);
+    }
+
+    @Override
+    public void property(String key, String value) {
+        this.properties.put(key, value);
+    }
+
+    @Override
+    public Object parameter() {
+        return this.parameter;
+    }
+
+    @Override
+    public void parameter(Object value) {
+        this.parameter = value;
+    }
+
+    @Override
+    public byte[] parameterBytes() {
+        return this.getParameterByte();
+    }
+
+    public byte[] getParameterByte() {
+        return parameterByte;
+    }
+
+    public void setParameterByte(byte[] parameterByte) {
+        this.parameterByte = parameterByte;
+    }
+
+    @Override
+    public void parameterBytes(byte[] value) {
+        this.setParameterByte(value);
+    }
+
+    @Override
+    public byte[] extraPayload() {
+        return this.extraPayload;
+    }
+
+    @Override
+    public void extraPayload(byte[] value) {
+        this.extraPayload = value;
+    }
+
+    @Override
+    public <T> T parameter(SerializerFactory serializerFactory, Class<T> c) {
+        if (this.parameter() != null)
+            return (T) this.parameter();
+        final T decode = serializerFactory.get(this.serializerType()).decode(this.parameterBytes(), c);
+        this.parameter(decode);
+        return decode;
+    }
+
+    @Override
+    public <T> T parameter(SerializerFactory serializerFactory, TypePresentation<T> typePresentation) {
+        if (this.parameter() != null)
+            return (T) this.parameter();
+        final T decode = serializerFactory.get(this.serializerType()).decode(this.parameterBytes(), typePresentation);
+        this.parameter(decode);
+        return decode;
+    }
+
+    @Override
+    public <T> T parameter(SerializerFactory serializerFactory, Type type) {
+        if (this.parameter() != null)
+            return (T) this.parameter();
+        final T decode = serializerFactory.get(this.serializerType()).decode(this.parameterBytes(), type);
+        this.parameter(decode);
+        return decode;
+    }
+
+    @Override
+    public int hashCode() {
+        return HashCodeBuilder.reflectionHashCode(this);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        return EqualsBuilder.reflectionEquals(this, o);
+    }
+
+    @Override
+    public String toString() {
+        return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/command/RequestIdGenerator.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.command;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class RequestIdGenerator {
+    public static RequestIdGenerator inst = new RequestIdGenerator();
+
+    private AtomicInteger generator = new AtomicInteger(0);
+
+    private RequestIdGenerator() {
+
+    }
+
+    public int incrementAndGet() {
+        return generator.incrementAndGet();
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/netty/NettyChannelEvent.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.netty;
+
+import io.netty.channel.Channel;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+
+public class NettyChannelEvent {
+    private final Channel channel;
+    private final NettyChannelEventType type;
+    private final Throwable cause;
+
+    public NettyChannelEvent(NettyChannelEventType type, Channel channel) {
+        this(type, channel, null);
+    }
+
+    public NettyChannelEvent(NettyChannelEventType type, Channel channel, Throwable cause) {
+        this.type = type;
+        this.channel = channel;
+        this.cause = cause;
+    }
+
+    public NettyChannelEventType getType() {
+        return type;
+    }
+
+    public Channel getChannel() {
+        return channel;
+    }
+
+    public Throwable getCause() {
+        return cause;
+    }
+
+    @Override
+    public String toString() {
+        return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
+    }
+}

remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/netty/NettyChannelEventType.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.impl.netty;
+
+public enum NettyChannelEventType {
+    ACTIVE,
+    INACTIVE,
+    IDLE,
+    EXCEPTION
+}