You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by yu...@apache.org on 2019/06/28 11:20:10 UTC
[rocketmq-remoting] 35/39: Do some minor polish to pass unit tests
This is an automated email from the ASF dual-hosted git repository.
yukon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-remoting.git
commit 866abacb6797038ad7b9fba557c4d82f3dc206a5
Author: yukon <yu...@apache.org>
AuthorDate: Tue Jun 11 11:55:50 2019 +0800
Do some minor polish to pass unit tests
---
.../remoting/api/interceptor/RequestContext.java | 2 +-
.../remoting/api/interceptor/ResponseContext.java | 2 +-
remoting-core/remoting-impl/pom.xml | 1 +
.../rocketmq/remoting/common/ResponseFuture.java | 2 +-
.../remoting/impl/netty/ClientChannelManager.java | 22 +++++++++++-----------
.../remoting/common/ResponseFutureTest.java | 2 +-
.../impl/netty/NettyRemotingAbstractTest.java | 20 ++++++++++----------
7 files changed, 26 insertions(+), 25 deletions(-)
diff --git a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/RequestContext.java b/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/RequestContext.java
index a93e71c..11814b7 100644
--- a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/RequestContext.java
+++ b/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/RequestContext.java
@@ -60,6 +60,6 @@ public class RequestContext {
@Override
public String toString() {
- return ToStringBuilder.reflectionToString(this, ToStringStyle.DEFAULT_STYLE);
+ return ToStringBuilder.reflectionToString(this, ToStringStyle.NO_CLASS_NAME_STYLE);
}
}
diff --git a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/ResponseContext.java b/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/ResponseContext.java
index f076d8f..e4c9b90 100644
--- a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/ResponseContext.java
+++ b/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/interceptor/ResponseContext.java
@@ -36,7 +36,7 @@ public class ResponseContext extends RequestContext {
@Override
public String toString() {
- return ToStringBuilder.reflectionToString(this, ToStringStyle.DEFAULT_STYLE);
+ return ToStringBuilder.reflectionToString(this, ToStringStyle.NO_CLASS_NAME_STYLE);
}
public RemotingCommand getResponse() {
diff --git a/remoting-core/remoting-impl/pom.xml b/remoting-core/remoting-impl/pom.xml
index 0af7895..4805004 100644
--- a/remoting-core/remoting-impl/pom.xml
+++ b/remoting-core/remoting-impl/pom.xml
@@ -31,6 +31,7 @@
<artifactId>netty-transport-native-epoll</artifactId>
<version>4.1.26.Final</version>
<classifier>${os.detected.name}-${os.detected.arch}</classifier>
+ <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
diff --git a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/ResponseFuture.java b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/ResponseFuture.java
index 705a0d4..cdff0d1 100644
--- a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/ResponseFuture.java
+++ b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/common/ResponseFuture.java
@@ -155,6 +155,6 @@ public class ResponseFuture {
@Override
public String toString() {
- return ToStringBuilder.reflectionToString(this, ToStringStyle.DEFAULT_STYLE);
+ return ToStringBuilder.reflectionToString(this, ToStringStyle.NO_CLASS_NAME_STYLE);
}
}
diff --git a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/netty/ClientChannelManager.java b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/netty/ClientChannelManager.java
index 4f4ec5c..44a1792 100644
--- a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/netty/ClientChannelManager.java
+++ b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/netty/ClientChannelManager.java
@@ -38,7 +38,7 @@ public class ClientChannelManager {
protected static final Logger LOG = LoggerFactory.getLogger(ClientChannelManager.class);
private static final long LOCK_TIMEOUT_MILLIS = 3000;
- final ConcurrentHashMap<String, ChannelWrapper> channelTables = new ConcurrentHashMap<>();
+ final ConcurrentHashMap<String, RemotingChannelFuture> channelTables = new ConcurrentHashMap<>();
private final Lock lockChannelTables = new ReentrantLock();
private final Bootstrap clientBootstrap;
private final RemotingClientConfig clientConfig;
@@ -50,7 +50,7 @@ public class ClientChannelManager {
}
void clear() {
- for (ChannelWrapper cw : this.channelTables.values()) {
+ for (RemotingChannelFuture cw : this.channelTables.values()) {
this.closeChannel(null, cw.getChannel());
}
@@ -58,7 +58,7 @@ public class ClientChannelManager {
}
Channel createIfAbsent(final String addr) {
- ChannelWrapper cw = this.channelTables.get(addr);
+ RemotingChannelFuture cw = this.channelTables.get(addr);
if (cw != null && cw.isActive()) {
return cw.getChannel();
}
@@ -66,7 +66,7 @@ public class ClientChannelManager {
}
private Channel createChannel(final String addr) {
- ChannelWrapper cw = null;
+ RemotingChannelFuture cw = null;
try {
if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
@@ -90,7 +90,7 @@ public class ClientChannelManager {
SocketAddress socketAddress = new InetSocketAddress(s[0], Integer.valueOf(s[1]));
ChannelFuture channelFuture = this.clientBootstrap.connect(socketAddress);
LOG.info("createChannel: begin to connect remote host[{}] asynchronously", addr);
- cw = new ChannelWrapper(channelFuture);
+ cw = new RemotingChannelFuture(channelFuture);
this.channelTables.put(addr, cw);
}
} catch (Exception e) {
@@ -129,7 +129,7 @@ public class ClientChannelManager {
if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
boolean removeItemFromTable = true;
- ChannelWrapper prevCW = this.channelTables.get(addrRemote);
+ RemotingChannelFuture prevCW = this.channelTables.get(addrRemote);
//Workaround for null
if (null == prevCW) {
return;
@@ -171,11 +171,11 @@ public class ClientChannelManager {
if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
boolean removeItemFromTable = true;
- ChannelWrapper prevCW = null;
+ RemotingChannelFuture prevCW = null;
String addrRemote = null;
- for (Map.Entry<String, ChannelWrapper> entry : channelTables.entrySet()) {
- ChannelWrapper prev = entry.getValue();
+ for (Map.Entry<String, RemotingChannelFuture> entry : channelTables.entrySet()) {
+ RemotingChannelFuture prev = entry.getValue();
if (prev.getChannel() != null) {
if (prev.getChannel() == channel) {
prevCW = prev;
@@ -208,10 +208,10 @@ public class ClientChannelManager {
}
}
- private class ChannelWrapper {
+ private class RemotingChannelFuture {
private final ChannelFuture channelFuture;
- ChannelWrapper(ChannelFuture channelFuture) {
+ RemotingChannelFuture(ChannelFuture channelFuture) {
this.channelFuture = channelFuture;
}
diff --git a/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/common/ResponseFutureTest.java b/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/common/ResponseFutureTest.java
index 302771d..69c7a66 100644
--- a/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/common/ResponseFutureTest.java
+++ b/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/common/ResponseFutureTest.java
@@ -101,7 +101,7 @@ public class ResponseFutureTest extends BaseTest {
@Test
public void waitResponse_Timeout() {
future = new ResponseFuture(1, 1000, null, null);
- RemotingCommand response = future.waitResponse(10);
+ RemotingCommand response = future.waitResponse(100);
assertNull(response);
}
}
\ No newline at end of file
diff --git a/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/impl/netty/NettyRemotingAbstractTest.java b/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/impl/netty/NettyRemotingAbstractTest.java
index 510dfa3..31f0393 100644
--- a/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/impl/netty/NettyRemotingAbstractTest.java
+++ b/remoting-core/remoting-impl/src/test/java/org/apache/rocketmq/remoting/impl/netty/NettyRemotingAbstractTest.java
@@ -240,7 +240,7 @@ public class NettyRemotingAbstractTest extends BaseTest {
@Test
public void scanResponseTable_RemoveTimeoutRequest() throws InterruptedException {
- final ObjectFuture<Throwable> objectFuture = newObjectFuture(1, 10);
+ final ObjectFuture<Throwable> objectFuture = newObjectFuture(1, 100);
remotingAbstract.invokeAsyncWithInterceptor(new EmbeddedChannel(),
remotingAbstract.commandFactory().createRequest(),
@@ -255,7 +255,7 @@ public class NettyRemotingAbstractTest extends BaseTest {
public void onSuccess(final RemotingCommand response) {
}
- }, 10);
+ }, 100);
TimeUnit.MILLISECONDS.sleep(15);
remotingAbstract.scanResponseTable();
@@ -274,10 +274,10 @@ public class NettyRemotingAbstractTest extends BaseTest {
@Test
public void invokeWithInterceptor_Timeout() {
- registerTimeoutProcessor(20);
+ registerTimeoutProcessor(200);
try {
- RemotingCommand response = remotingAbstract.invokeWithInterceptor(clientChannel, remotingRequest, 10);
+ RemotingCommand response = remotingAbstract.invokeWithInterceptor(clientChannel, remotingRequest, 100);
failBecauseExceptionWasNotThrown(RemotingTimeoutException.class);
} catch (Exception e) {
assertThat(e).isInstanceOf(RemotingTimeoutException.class);
@@ -292,7 +292,7 @@ public class NettyRemotingAbstractTest extends BaseTest {
channelPromise.setFailure(new UnitTestException());
try {
- RemotingCommand response = remotingAbstract.invokeWithInterceptor(mockedClientChannel, remotingRequest, 10);
+ RemotingCommand response = remotingAbstract.invokeWithInterceptor(mockedClientChannel, remotingRequest, 100);
failBecauseExceptionWasNotThrown(RemotingAccessException.class);
} catch (Exception e) {
assertThat(e.getCause()).isInstanceOf(UnitTestException.class);
@@ -304,7 +304,7 @@ public class NettyRemotingAbstractTest extends BaseTest {
public void invokeAsyncWithInterceptor_Success() {
registerNormalProcessor();
- final ObjectFuture<RemotingCommand> objectFuture = newObjectFuture(1, 10);
+ final ObjectFuture<RemotingCommand> objectFuture = newObjectFuture(1, 100);
remotingAbstract.invokeAsyncWithInterceptor(clientChannel, remotingRequest, new AsyncHandler() {
@Override
@@ -326,7 +326,7 @@ public class NettyRemotingAbstractTest extends BaseTest {
public void invokeAsyncWithInterceptor_SemaphoreExhausted() {
registerTimeoutProcessor(1000);
- final ObjectFuture<Throwable> objectFuture = newObjectFuture(1, 10);
+ final ObjectFuture<Throwable> objectFuture = newObjectFuture(1, 100);
for (int i = 0; i < semaphoreNum; i++) {
remotingAbstract.invokeAsyncWithInterceptor(clientChannel, remotingRequest, null, 100);
@@ -355,7 +355,7 @@ public class NettyRemotingAbstractTest extends BaseTest {
when(mockedClientChannel.writeAndFlush(any(Object.class))).thenReturn(channelPromise);
channelPromise.setFailure(new UnitTestException());
- final ObjectFuture<Throwable> objectFuture = newObjectFuture(1, 10);
+ final ObjectFuture<Throwable> objectFuture = newObjectFuture(1, 100);
remotingAbstract.invokeAsyncWithInterceptor(mockedClientChannel, remotingRequest, new AsyncHandler() {
@Override
@@ -368,14 +368,14 @@ public class NettyRemotingAbstractTest extends BaseTest {
public void onSuccess(final RemotingCommand response) {
}
- }, 10);
+ }, 100);
assertThat(objectFuture.getObject().getCause()).isInstanceOf(UnitTestException.class);
}
@Test
public void invokeOnewayWithInterceptor_Success() {
- ObjectFuture<RemotingCommand> objectFuture = newObjectFuture(1, 10);
+ ObjectFuture<RemotingCommand> objectFuture = newObjectFuture(1, 100);
registerOnewayProcessor(objectFuture);
remotingAbstract.invokeOnewayWithInterceptor(clientChannel, remotingRequest);