Skip to content

R
Rocketmq

小五666\n哈哈 / Rocketmq
与 Fork 源项目一致

Fork自 Apache RocketMQ / Rocketmq

通知 1
Star 0
Fork 0
R
Rocketmq
提交 f7c55b9f 编写于 作者: D duhenglucky
浏览文件
操作

Fix unit test ensure mvn install pass

上级 ac25bccc
隐藏空白更改
内联 并排
Showing with 251 addition and 289 deletion
broker/src/main/java/org/apache/rocketmq/broker/client/ClientChannelInfo.java
浏览文件 @ f7c55b9f
......@@ -16,12 +16,11 @@
*/
package org.apache.rocketmq.broker.client;
import io.netty.channel.Channel;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
public class ClientChannelInfo {
private final RemotingChannel channel;
private final RemotingChannel remotingChannel;
private final String clientId;
private final LanguageCode language;
private final int version;
......@@ -31,15 +30,15 @@ public class ClientChannelInfo {
this(channel, null, null, 0);
}
public ClientChannelInfo(RemotingChannel channel, String clientId, LanguageCode language, int version) {
this.channel = channel;
public ClientChannelInfo(RemotingChannel remotingChannel, String clientId, LanguageCode language, int version) {
this.remotingChannel = remotingChannel;
this.clientId = clientId;
this.language = language;
this.version = version;
}
public RemotingChannel getChannel() {
return channel;
public RemotingChannel getRemotingChannel() {
return remotingChannel;
}
public String getClientId() {
......@@ -66,7 +65,7 @@ public class ClientChannelInfo {
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((channel == null) ? 0 : channel.hashCode());
result = prime * result + ((remotingChannel == null) ? 0 : remotingChannel.hashCode());
result = prime * result + ((clientId == null) ? 0 : clientId.hashCode());
result = prime * result + ((language == null) ? 0 : language.hashCode());
result = prime * result + (int) (lastUpdateTimestamp ^ (lastUpdateTimestamp >>> 32));
......@@ -83,10 +82,10 @@ public class ClientChannelInfo {
if (getClass() != obj.getClass())
return false;
ClientChannelInfo other = (ClientChannelInfo) obj;
if (channel == null) {
if (other.channel != null)
if (remotingChannel == null) {
if (other.remotingChannel != null)
return false;
} else if (this.channel != other.channel) {
} else if (this.remotingChannel != other.remotingChannel) {
return false;
}
......@@ -95,7 +94,7 @@ public class ClientChannelInfo {
@Override
public String toString() {
return "ClientChannelInfo [channel=" + channel + ", clientId=" + clientId + ", language=" + language
return "ClientChannelInfo [remotingChannel=" + remotingChannel + ", clientId=" + clientId + ", language=" + language
+ ", version=" + version + ", lastUpdateTimestamp=" + lastUpdateTimestamp + "]";
}
}
broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java
浏览文件 @ f7c55b9f
......@@ -74,28 +74,28 @@ public class ClientHousekeepingService implements ChannelEventListener {
log.info("Remoting channel closed: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
}
@Override
public void onChannelException(String remoteAddr, RemotingChannel remotingChannel) {
log.info("Remoting channel exception: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
log.info("Remoting channel exception: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
}
@Override
public void onChannelIdle(String remoteAddr, RemotingChannel remotingChannel) {
log.info("Remoting channel idle: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
log.info("Remoting channel idle: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
}
}
broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerGroupInfo.java
浏览文件 @ f7c55b9f
......@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.broker.client;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
......@@ -26,11 +25,11 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
public class ConsumerGroupInfo {
......@@ -99,13 +98,13 @@ public class ConsumerGroupInfo {
}
public void unregisterChannel(final ClientChannelInfo clientChannelInfo) {
ClientChannelInfo old = this.channelInfoTable.remove(clientChannelInfo.getChannel());
ClientChannelInfo old = this.channelInfoTable.remove(clientChannelInfo.getRemotingChannel());
if (old != null) {
log.info("unregister a consumer[{}] from consumerGroupInfo {}", this.groupName, old.toString());
}
}
public boolean doChannelCloseEvent(final String remoteAddr, final Channel channel) {
public boolean doChannelCloseEvent(final String remoteAddr, final RemotingChannel channel) {
final ClientChannelInfo info = this.channelInfoTable.remove(channel);
if (info != null) {
log.warn(
......@@ -124,9 +123,9 @@ public class ConsumerGroupInfo {
this.messageModel = messageModel;
this.consumeFromWhere = consumeFromWhere;
ClientChannelInfo infoOld = this.channelInfoTable.get(infoNew.getChannel());
ClientChannelInfo infoOld = this.channelInfoTable.get(infoNew.getRemotingChannel());
if (null == infoOld) {
ClientChannelInfo prev = this.channelInfoTable.put(infoNew.getChannel(), infoNew);
ClientChannelInfo prev = this.channelInfoTable.put(infoNew.getRemotingChannel(), infoNew);
if (null == prev) {
log.info("new consumer connected, group: {} {} {} channel: {}", this.groupName, consumeType,
messageModel, infoNew.toString());
......@@ -140,7 +139,7 @@ public class ConsumerGroupInfo {
this.groupName,
infoOld.toString(),
infoNew.toString());
this.channelInfoTable.put(infoNew.getChannel(), infoNew);
this.channelInfoTable.put(infoNew.getRemotingChannel(), infoNew);
}
}
......
broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java
浏览文件 @ f7c55b9f
......@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.broker.client;
import io.netty.channel.Channel;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map.Entry;
......@@ -25,14 +24,13 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.common.RemotingUtil;
public class ConsumerManager {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
......@@ -75,7 +73,7 @@ public class ConsumerManager {
return 0;
}
public void doChannelCloseEvent(final String remoteAddr, final Channel channel) {
public void doChannelCloseEvent(final String remoteAddr, final RemotingChannel channel) {
Iterator<Entry<String, ConsumerGroupInfo>> it = this.consumerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, ConsumerGroupInfo> next = it.next();
......@@ -159,9 +157,9 @@ public class ConsumerManager {
if (diff > CHANNEL_EXPIRED_TIMEOUT) {
log.warn(
"SCAN: remove expired channel from ConsumerManager consumerTable. channel={}, consumerGroup={}",
RemotingHelper.parseChannelRemoteAddr(clientChannelInfo.getChannel().remoteAddress()), group);
RemotingHelper.parseChannelRemoteAddr(clientChannelInfo.getRemotingChannel().remoteAddress()), group);
clientChannelInfo.getChannel().close();
clientChannelInfo.getRemotingChannel().close();
itChannel.remove();
}
}
......
broker/src/main/java/org/apache/rocketmq/broker/client/DefaultConsumerIdsChangeListener.java
浏览文件 @ f7c55b9f
......@@ -16,11 +16,8 @@
*/
package org.apache.rocketmq.broker.client;
import io.netty.channel.Channel;
import java.util.Collection;
import java.util.List;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.RemotingChannel;
......
broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java
浏览文件 @ f7c55b9f
......@@ -16,8 +16,6 @@
*/
package org.apache.rocketmq.broker.client;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
......@@ -27,14 +25,12 @@ import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.rocketmq.broker.util.PositiveAtomicCounter;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.common.RemotingUtil;
public class ProducerManager {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
......@@ -86,8 +82,8 @@ public class ProducerManager {
it.remove();
log.warn(
"SCAN: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}",
RemotingHelper.parseChannelRemoteAddr(info.getChannel().remoteAddress()), group);
info.getChannel().close();
RemotingHelper.parseChannelRemoteAddr(info.getRemotingChannel().remoteAddress()), group);
info.getRemotingChannel().close();
}
}
}
......@@ -102,7 +98,7 @@ public class ProducerManager {
}
}
public void doChannelCloseEvent(final String remoteAddr, final Channel channel) {
public void doChannelCloseEvent(final String remoteAddr, final RemotingChannel channel) {
if (channel != null) {
try {
if (this.groupChannelLock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
......@@ -110,16 +106,12 @@ public class ProducerManager {
for (final Map.Entry<String, HashMap<RemotingChannel, ClientChannelInfo>> entry : this.groupChannelTable
.entrySet()) {
final String group = entry.getKey();
final HashMap<RemotingChannel, ClientChannelInfo> clientChannelInfoTable =
entry.getValue();
final ClientChannelInfo clientChannelInfo =
clientChannelInfoTable.remove(channel);
final HashMap<RemotingChannel, ClientChannelInfo> clientChannelInfoTable = entry.getValue();
final ClientChannelInfo clientChannelInfo = clientChannelInfoTable.remove(channel);
if (clientChannelInfo != null) {
log.info(
"NETTY EVENT: remove channel[{}][{}] from ProducerManager groupChannelTable, producer group: {}",
log.info("NETTY EVENT: remove channel[{}][{}] from ProducerManager groupChannelTable, producer group: {}",
clientChannelInfo.toString(), remoteAddr, group);
}
}
} finally {
this.groupChannelLock.unlock();
......@@ -145,9 +137,9 @@ public class ProducerManager {
this.groupChannelTable.put(group, channelTable);
}
clientChannelInfoFound = channelTable.get(clientChannelInfo.getChannel());
clientChannelInfoFound = channelTable.get(clientChannelInfo.getRemotingChannel());
if (null == clientChannelInfoFound) {
channelTable.put(clientChannelInfo.getChannel(), clientChannelInfo);
channelTable.put(clientChannelInfo.getRemotingChannel(), clientChannelInfo);
log.info("new producer connected, group: {} channel: {}", group,
clientChannelInfo.toString());
}
......@@ -172,7 +164,7 @@ public class ProducerManager {
try {
HashMap<RemotingChannel, ClientChannelInfo> channelTable = this.groupChannelTable.get(group);
if (null != channelTable && !channelTable.isEmpty()) {
ClientChannelInfo old = channelTable.remove(clientChannelInfo.getChannel());
ClientChannelInfo old = channelTable.remove(clientChannelInfo.getRemotingChannel());
if (old != null) {
log.info("unregister a producer[{}] from groupChannelTable {}", group,
clientChannelInfo.toString());
......
broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java
浏览文件 @ f7c55b9f
......@@ -16,7 +16,12 @@
*/
package org.apache.rocketmq.broker.client.net;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
......@@ -45,13 +50,6 @@ import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
public class Broker2Client {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private final BrokerController brokerController;
......
broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
浏览文件 @ f7c55b9f
......@@ -47,6 +47,7 @@ import org.apache.rocketmq.common.protocol.header.namesrv.RegisterBrokerResponse
import org.apache.rocketmq.common.protocol.header.namesrv.UnRegisterBrokerRequestHeader;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.RemotingClientFactory;
......@@ -55,7 +56,6 @@ import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class BrokerOuterAPI {
......@@ -193,9 +193,6 @@ public class BrokerOuterAPI {
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
assert response != null;
if (response == null){
System.out.println("ssssssssssssss");
}
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
RegisterBrokerResponseHeader responseHeader =
......
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
浏览文件 @ f7c55b9f
......@@ -603,7 +603,7 @@ public class AdminBrokerProcessor implements RequestProcessor {
connection.setClientId(info.getClientId());
connection.setLanguage(info.getLanguage());
connection.setVersion(info.getVersion());
connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getChannel().remoteAddress()));
connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getRemotingChannel().remoteAddress()));
bodydata.getConnectionSet().add(connection);
}
......@@ -638,7 +638,7 @@ public class AdminBrokerProcessor implements RequestProcessor {
connection.setClientId(info.getClientId());
connection.setLanguage(info.getLanguage());
connection.setVersion(info.getVersion());
connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getChannel().remoteAddress()));
connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getRemotingChannel().remoteAddress()));
bodydata.getConnectionSet().add(connection);
}
......@@ -1278,7 +1278,7 @@ public class AdminBrokerProcessor implements RequestProcessor {
newRequest.setExtFields(request.getExtFields());
newRequest.setBody(request.getBody());
return this.brokerController.getBroker2Client().callClient(clientChannelInfo.getChannel(), newRequest);
return this.brokerController.getBroker2Client().callClient(clientChannelInfo.getRemotingChannel(), newRequest);
} catch (RemotingTimeoutException e) {
response.setCode(ResponseCode.CONSUME_MSG_TIMEOUT);
response
......
broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
浏览文件 @ f7c55b9f
......@@ -16,7 +16,11 @@
*/
package org.apache.rocketmq.broker.transaction;
import io.netty.channel.Channel;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
......@@ -24,12 +28,6 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.remoting.RemotingChannel;
public abstract class AbstractTransactionalMessageCheckListener {
......
broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java
浏览文件 @ f7c55b9f
......@@ -21,6 +21,7 @@ import io.netty.channel.ChannelFuture;
import java.lang.reflect.Field;
import java.util.HashMap;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
......@@ -37,35 +38,38 @@ public class ProducerManagerTest {
private String group = "FooBar";
private ClientChannelInfo clientInfo;
private RemotingChannel remotingChannel;
@Mock
private RemotingChannel channel;
private Channel channel;
@Before
public void init() {
producerManager = new ProducerManager();
clientInfo = new ClientChannelInfo(channel);
remotingChannel = new NettyChannelImpl(channel);
clientInfo = new ClientChannelInfo(remotingChannel);
}
@Test
public void scanNotActiveChannel() throws Exception {
producerManager.registerProducer(group, clientInfo);
assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNotNull();
assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNotNull();
Field field = ProducerManager.class.getDeclaredField("CHANNEL_EXPIRED_TIMEOUT");
field.setAccessible(true);
long CHANNEL_EXPIRED_TIMEOUT = field.getLong(producerManager);
clientInfo.setLastUpdateTimestamp(System.currentTimeMillis() - CHANNEL_EXPIRED_TIMEOUT - 10);
// when(channel.close()).thenReturn(mock(ChannelFuture.class));
when(channel.close()).thenReturn(mock(ChannelFuture.class));
producerManager.scanNotActiveChannel();
assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNull();
assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNull();
}
@Test
public void doChannelCloseEvent() throws Exception {
producerManager.registerProducer(group, clientInfo);
assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNotNull();
// producerManager.doChannelCloseEvent("127.0.0.1", channel);
assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNull();
assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNotNull();
producerManager.doChannelCloseEvent("127.0.0.1", remotingChannel);
assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNull();
}
@Test
......@@ -73,7 +77,7 @@ public class ProducerManagerTest {
producerManager.registerProducer(group, clientInfo);
HashMap<RemotingChannel, ClientChannelInfo> channelMap = producerManager.getGroupChannelTable().get(group);
assertThat(channelMap).isNotNull();
assertThat(channelMap.get(channel)).isEqualTo(clientInfo);
assertThat(channelMap.get(remotingChannel)).isEqualTo(clientInfo);
}
@Test
......@@ -81,7 +85,7 @@ public class ProducerManagerTest {
producerManager.registerProducer(group, clientInfo);
HashMap<RemotingChannel, ClientChannelInfo> channelMap = producerManager.getGroupChannelTable().get(group);
assertThat(channelMap).isNotNull();
assertThat(channelMap.get(channel)).isEqualTo(clientInfo);
assertThat(channelMap.get(remotingChannel)).isEqualTo(clientInfo);
producerManager.unregisterProducer(group, clientInfo);
channelMap = producerManager.getGroupChannelTable().get(group);
......
broker/src/test/java/org/apache/rocketmq/broker/processor/ClientManageProcessorTest.java
浏览文件 @ f7c55b9f
......@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.broker.processor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.util.HashMap;
import java.util.UUID;
......@@ -27,14 +28,15 @@ import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.UnregisterClientRequestHeader;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.CodecHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.Before;
import org.junit.Test;
......@@ -45,6 +47,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import static org.apache.rocketmq.broker.processor.PullMessageProcessorTest.createConsumerData;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class ClientManageProcessorTest {
......@@ -52,9 +55,15 @@ public class ClientManageProcessorTest {
@Spy
private BrokerController brokerController = new BrokerController(new BrokerConfig(), new ServerConfig(), new ClientConfig(), new MessageStoreConfig());
@Mock
private ChannelHandlerContext handlerContext;
private NettyChannelHandlerContextImpl handlerContext;
@Mock
private NettyChannelImpl channel;
private ChannelHandlerContext channelHandlerContext;
private RemotingChannel remotingChannel;
@Mock
private Channel channel;
private ClientChannelInfo clientChannelInfo;
private String clientId = UUID.randomUUID().toString();
......@@ -63,11 +72,12 @@ public class ClientManageProcessorTest {
@Before
public void init() {
// when(handlerContext.channel()).thenReturn(channel);
when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
when(channelHandlerContext.channel()).thenReturn(channel);
clientManageProcessor = new ClientManageProcessor(brokerController);
clientChannelInfo = new ClientChannelInfo(channel, clientId, LanguageCode.JAVA, 100);
remotingChannel = new NettyChannelImpl(channel);
clientChannelInfo = new ClientChannelInfo(remotingChannel, clientId, LanguageCode.JAVA, 100);
brokerController.getProducerManager().registerProducer(group, clientChannelInfo);
ConsumerData consumerData = createConsumerData(group, topic);
brokerController.getConsumerManager().registerConsumer(
consumerData.getGroupName(),
......@@ -84,11 +94,10 @@ public class ClientManageProcessorTest {
brokerController.getProducerManager().registerProducer(group, clientChannelInfo);
HashMap<RemotingChannel, ClientChannelInfo> channelMap = brokerController.getProducerManager().getGroupChannelTable().get(group);
assertThat(channelMap).isNotNull();
assertThat(channelMap.get(channel)).isEqualTo(clientChannelInfo);
assertThat(channelMap.get(remotingChannel)).isEqualTo(clientChannelInfo);
RemotingCommand request = createUnRegisterProducerCommand();
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = clientManageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = clientManageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
......@@ -102,9 +111,8 @@ public class ClientManageProcessorTest {
assertThat(consumerGroupInfo).isNotNull();
RemotingCommand request = createUnRegisterConsumerCommand();
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = clientManageProcessor.processRequest(nettyChannelHandlerContext, request); assertThat(response).isNotNull();
RemotingCommand response = clientManageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
consumerGroupInfo = brokerController.getConsumerManager().getConsumerGroupInfo(group);
......@@ -118,7 +126,7 @@ public class ClientManageProcessorTest {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, requestHeader);
request.setLanguage(LanguageCode.JAVA);
request.setVersion(100);
// request.makeCustomHeaderToNet();
CodecHelper.makeCustomHeaderToNet(request);
return request;
}
......@@ -129,7 +137,7 @@ public class ClientManageProcessorTest {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, requestHeader);
request.setLanguage(LanguageCode.JAVA);
request.setVersion(100);
// request.makeCustomHeaderToNet();
CodecHelper.makeCustomHeaderToNet(request);
return request;
}
}
\ No newline at end of file
broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
浏览文件 @ f7c55b9f
......@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.broker.processor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.transaction.OperationResult;
......@@ -28,10 +29,11 @@ import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.CodecHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.AppendMessageStatus;
......@@ -49,6 +51,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
......@@ -57,7 +60,10 @@ public class EndTransactionProcessorTest {
private EndTransactionProcessor endTransactionProcessor;
@Mock
private ChannelHandlerContext handlerContext;
private NettyChannelHandlerContextImpl handlerContext;
@Mock
private ChannelHandlerContext channelHandlerContext;
@Spy
private BrokerController
......@@ -74,10 +80,13 @@ public class EndTransactionProcessorTest {
public void init() {
brokerController.setMessageStore(messageStore);
brokerController.setTransactionalMessageService(transactionMsgService);
Channel mockChannel = mock(Channel.class);
when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
when(channelHandlerContext.channel()).thenReturn(mockChannel);
endTransactionProcessor = new EndTransactionProcessor(brokerController);
}
private OperationResult createResponse(int status) {
private OperationResult createResponse(int status){
OperationResult response = new OperationResult();
response.setPrepareMessage(createDefaultMessageExt());
response.setResponseCode(status);
......@@ -91,9 +100,7 @@ public class EndTransactionProcessorTest {
when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult
(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)));
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_COMMIT_TYPE, false);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
......@@ -103,16 +110,14 @@ public class EndTransactionProcessorTest {
when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult
(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)));
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_COMMIT_TYPE, true);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testProcessRequest_NotType() throws RemotingCommandException {
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_NOT_TYPE, true);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response).isNull();
}
......@@ -120,8 +125,7 @@ public class EndTransactionProcessorTest {
public void testProcessRequest_RollBack() throws RemotingCommandException {
when(transactionMsgService.rollbackMessage(any(EndTransactionRequestHeader.class))).thenReturn(createResponse(ResponseCode.SUCCESS));
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE, true);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
......@@ -152,7 +156,7 @@ public class EndTransactionProcessorTest {
private RemotingCommand createEndTransactionMsgCommand(int status, boolean isCheckMsg) {
EndTransactionRequestHeader header = createEndTransactionRequestHeader(status, isCheckMsg);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, header);
// request.makeCustomHeaderToNet();
CodecHelper.makeCustomHeaderToNet(request);
return request;
}
}
broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java
浏览文件 @ f7c55b9f
......@@ -38,11 +38,12 @@ import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.CodecHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.GetMessageStatus;
......@@ -69,9 +70,11 @@ public class PullMessageProcessorTest {
@Spy
private BrokerController brokerController = new BrokerController(new BrokerConfig(), new ServerConfig(), new ClientConfig(), new MessageStoreConfig());
@Mock
private ChannelHandlerContext handlerContext;
private NettyChannelHandlerContextImpl handlerContext;
@Mock
private Channel channel;
private ChannelHandlerContext channelHandlerContext;
@Mock
private MessageStore messageStore;
private ClientChannelInfo clientChannelInfo;
......@@ -82,11 +85,13 @@ public class PullMessageProcessorTest {
public void init() {
brokerController.setMessageStore(messageStore);
pullMessageProcessor = new PullMessageProcessor(brokerController);
RemotingChannel mockChannel = mock(RemotingChannel.class);
Channel mockChannel = mock(Channel.class);
RemotingChannel remotingChannel = mock(RemotingChannel.class);
when(mockChannel.remoteAddress()).thenReturn(new InetSocketAddress(1024));
// when(handlerContext.channel()).thenReturn(mockChannel);
when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
when(channelHandlerContext.channel()).thenReturn(mockChannel);
brokerController.getTopicConfigManager().getTopicConfigTable().put(topic, new TopicConfig());
clientChannelInfo = new ClientChannelInfo(mockChannel);
clientChannelInfo = new ClientChannelInfo(remotingChannel);
ConsumerData consumerData = createConsumerData(group, topic);
brokerController.getConsumerManager().registerConsumer(
consumerData.getGroupName(),
......@@ -102,8 +107,7 @@ public class PullMessageProcessorTest {
public void testProcessRequest_TopicNotExist() throws RemotingCommandException {
brokerController.getTopicConfigManager().getTopicConfigTable().remove(topic);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.TOPIC_NOT_EXIST);
assertThat(response.getRemark()).contains("topic[" + topic + "] not exist");
......@@ -113,9 +117,7 @@ public class PullMessageProcessorTest {
public void testProcessRequest_SubNotExist() throws RemotingCommandException {
brokerController.getConsumerManager().unregisterConsumer(group, clientChannelInfo, false);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_NOT_EXIST);
assertThat(response.getRemark()).contains("consumer's group info not exist");
......@@ -125,8 +127,7 @@ public class PullMessageProcessorTest {
public void testProcessRequest_SubNotLatest() throws RemotingCommandException {
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
request.addExtField("subVersion", String.valueOf(101));
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_NOT_LATEST);
assertThat(response.getRemark()).contains("subscription not latest");
......@@ -138,9 +139,7 @@ public class PullMessageProcessorTest {
when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
......@@ -169,9 +168,7 @@ public class PullMessageProcessorTest {
consumeMessageHookList.add(consumeMessageHook);
pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(messageContext[0]).isNotNull();
......@@ -187,9 +184,7 @@ public class PullMessageProcessorTest {
when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.PULL_RETRY_IMMEDIATELY);
}
......@@ -201,9 +196,7 @@ public class PullMessageProcessorTest {
when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.PULL_OFFSET_MOVED);
}
......@@ -220,7 +213,7 @@ public class PullMessageProcessorTest {
requestHeader.setSysFlag(0);
requestHeader.setSubVersion(100L);
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, requestHeader);
// request.makeCustomHeaderToNet();
CodecHelper.makeCustomHeaderToNet(request);
return request;
}
......
broker/src/test/java/org/apache/rocketmq/broker/processor/SendMessageProcessorTest.java
浏览文件 @ f7c55b9f
......@@ -18,6 +18,10 @@ package org.apache.rocketmq.broker.processor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.mqtrace.SendMessageContext;
import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
......@@ -32,10 +36,11 @@ import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.CodecHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.AppendMessageStatus;
......@@ -53,11 +58,6 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
......@@ -69,7 +69,10 @@ import static org.mockito.Mockito.when;
public class SendMessageProcessorTest {
private SendMessageProcessor sendMessageProcessor;
@Mock
private ChannelHandlerContext handlerContext;
private NettyChannelHandlerContextImpl handlerContext;
@Mock
private ChannelHandlerContext channelHandlerContext;
@Spy
private BrokerController brokerController = new BrokerController(new BrokerConfig(), new ServerConfig(), new ClientConfig(), new MessageStoreConfig());
@Mock
......@@ -87,7 +90,8 @@ public class SendMessageProcessorTest {
when(messageStore.now()).thenReturn(System.currentTimeMillis());
Channel mockChannel = mock(Channel.class);
when(mockChannel.remoteAddress()).thenReturn(new InetSocketAddress(1024));
when(handlerContext.channel()).thenReturn(mockChannel);
when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
when(channelHandlerContext.channel()).thenReturn(mockChannel);
when(messageStore.lookMessageByOffset(anyLong())).thenReturn(new MessageExt());
sendMessageProcessor = new SendMessageProcessor(brokerController);
}
......@@ -182,9 +186,7 @@ public class SendMessageProcessorTest {
final RemotingCommand request = createSendMsgBackCommand(RequestCode.CONSUMER_SEND_MSG_BACK);
sendMessageProcessor = new SendMessageProcessor(brokerController);
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand response = sendMessageProcessor.processRequest(nettyChannelHandlerContext, request);
final RemotingCommand response = sendMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
......@@ -201,11 +203,8 @@ public class SendMessageProcessorTest {
response[0] = invocation.getArgument(0);
return null;
}
}).when(handlerContext).writeAndFlush(any(Object.class));
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand responseToReturn = sendMessageProcessor.processRequest(nettyChannelHandlerContext, request);
}).when(channelHandlerContext).writeAndFlush(any(Object.class));
RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request);
if (responseToReturn != null) {
assertThat(response[0]).isNull();
response[0] = responseToReturn;
......@@ -213,7 +212,6 @@ public class SendMessageProcessorTest {
assertThat(response[0].getCode()).isEqualTo(ResponseCode.SUCCESS);
}
private RemotingCommand createSendTransactionMsgCommand(int requestCode) {
SendMessageRequestHeader header = createSendMsgRequestHeader();
int sysFlag = header.getSysFlag();
......@@ -224,7 +222,7 @@ public class SendMessageProcessorTest {
header.setSysFlag(sysFlag);
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, header);
request.setBody(new byte[] {'a'});
// request.makeCustomHeaderToNet();
CodecHelper.makeCustomHeaderToNet(request);
return request;
}
......@@ -247,7 +245,7 @@ public class SendMessageProcessorTest {
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, requestHeader);
request.setBody(new byte[] {'a'});
// request.makeCustomHeaderToNet();
CodecHelper.makeCustomHeaderToNet(request);
return request;
}
......@@ -260,7 +258,7 @@ public class SendMessageProcessorTest {
requestHeader.setOffset(123L);
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, requestHeader);
// request.makeCustomHeaderToNet();
CodecHelper.makeCustomHeaderToNet(request);
return request;
}
......@@ -273,10 +271,8 @@ public class SendMessageProcessorTest {
response[0] = invocation.getArgument(0);
return null;
}
}).when(handlerContext).writeAndFlush(any(Object.class));
NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
RemotingCommand responseToReturn = sendMessageProcessor.processRequest(nettyChannelHandlerContext, request);
}).when(channelHandlerContext).writeAndFlush(any(Object.class));
RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request);
if (responseToReturn != null) {
assertThat(response[0]).isNull();
response[0] = responseToReturn;
......
client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
浏览文件 @ f7c55b9f
......@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.client.consumer.store;
import com.sun.org.apache.regexp.internal.RE;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
......@@ -31,10 +30,10 @@ import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.exception.RemotingException;
/**
......
client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
浏览文件 @ f7c55b9f
......@@ -135,8 +135,8 @@ import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
......@@ -144,12 +144,10 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.netty.ResponseFuture;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.remoting.serialize.RemotingSerializable;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyRemotingClient;
......
client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
浏览文件 @ f7c55b9f
......@@ -62,7 +62,6 @@ import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.filter.ExpressionType;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
......@@ -75,11 +74,10 @@ import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.QueueData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
......
client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
浏览文件 @ f7c55b9f
......@@ -147,7 +147,6 @@ public class DefaultMQPushConsumerTest {
});
doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
messageQueueSet.add(createPullRequest().getMessageQueue());
pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
......
common/src/main/java/org/apache/rocketmq/common/UtilAll.java
浏览文件 @ f7c55b9f
......@@ -16,8 +16,6 @@
*/
package org.apache.rocketmq.common;
import com.alibaba.fastjson.parser.ParserConfig;
import com.alibaba.fastjson.util.TypeUtils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
......@@ -31,12 +29,10 @@ import java.net.NetworkInterface;
import java.text.NumberFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.zip.CRC32;
import java.util.zip.DeflaterOutputStream;
......
common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlConfig.java
浏览文件 @ f7c55b9f
......@@ -37,14 +37,11 @@ public class FlowControlConfig {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
private String flowControlFileHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
private static final String DEFAULT_FLOW_CONTROL_FILE = "conf/flowControl.yml";
private String flowControlFileName = System.getProperty("rocketmq.flow.control.file", DEFAULT_FLOW_CONTROL_FILE);
private List<FlowControlRule> rules;
public static final String defaultResourceName = "overallFlowControl";
private Map<String/*server name*/, Map<String/*flowControlType*/, List<FlowControlRule>>> plainFlowControlRules;
public FlowControlConfig() {
......
common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlRule.java
浏览文件 @ f7c55b9f
......@@ -16,8 +16,6 @@
*/
package org.apache.rocketmq.common.flowcontrol;
import org.apache.rocketmq.remoting.interceptor.Interceptor;
public class FlowControlRule {
private String flowControlResourceName;
private Integer flowControlGrade;
......
example/src/main/java/org/apache/rocketmq/example/benchmark/Producer.java
浏览文件 @ f7c55b9f
......@@ -24,7 +24,6 @@ import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
......@@ -33,8 +32,8 @@ import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.srvutil.ServerUtil;
......
example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java
浏览文件 @ f7c55b9f
......@@ -20,7 +20,6 @@ import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.remoting.common.RemotingHelper;
/**
......
namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
浏览文件 @ f7c55b9f
......@@ -295,10 +295,9 @@ public class DefaultRequestProcessor implements RequestProcessor {
public RemotingCommand registerBroker(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
final RegisterBrokerRequestHeader requestHeader =
(RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);
log.info("requestHeader： " + requestHeader);
final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
if (!checksum(ctx, request, requestHeader)) {
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark("crc32 not match");
......
namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java
浏览文件 @ f7c55b9f
......@@ -39,22 +39,28 @@ public class BrokerHousekeepingService implements ChannelEventListener {
@Override
public void onChannelClose(String remoteAddr, RemotingChannel remotingChannel) {
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
if (remotingChannel != null) {
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
}
}
@Override
public void onChannelException(String remoteAddr, RemotingChannel remotingChannel) {
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
if (remotingChannel != null) {
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
}
}
@Override
public void onChannelIdle(String remoteAddr, RemotingChannel remotingChannel) {
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
if (remotingChannel != null) {
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
}
}
}
namesrv/src/test/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessorTest.java
浏览文件 @ f7c55b9f
......@@ -25,7 +25,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.common.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.common.protocol.RequestCode;
......@@ -37,11 +36,12 @@ import org.apache.rocketmq.common.protocol.header.namesrv.GetKVConfigResponseHea
import org.apache.rocketmq.common.protocol.header.namesrv.PutKVConfigRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.RegisterBrokerRequestHeader;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.assertj.core.util.Maps;
import org.junit.Before;
......@@ -91,8 +91,10 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
request.addExtField("value", "value");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
......@@ -111,7 +113,10 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
......@@ -130,7 +135,10 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.QUERY_NOT_FOUND);
assertThat(response.getRemark()).isEqualTo("No config item, Namespace: namespace Key: key");
......@@ -151,8 +159,11 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
......
remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServer.java
浏览文件 @ f7c55b9f
......@@ -21,7 +21,6 @@ import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public interface RemotingServer extends RemotingService {
......
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelHandlerContextImpl.java
浏览文件 @ f7c55b9f
......@@ -99,7 +99,7 @@ public class NettyChannelHandlerContextImpl implements RemotingChannel {
final NettyChannelHandlerContextImpl that = (NettyChannelHandlerContextImpl) o;
return channelHandlerContext != null ? channelHandlerContext.equals(that.channelHandlerContext) : that.channelHandlerContext == null;
return channelHandlerContext.channel() != null ? channelHandlerContext.channel().equals(that.channelHandlerContext.channel()) : that.channelHandlerContext.channel() == null;
}
......
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelImpl.java
浏览文件 @ f7c55b9f
......@@ -32,7 +32,7 @@ public class NettyChannelImpl implements RemotingChannel {
private static final InternalLogger log = InternalLoggerFactory.getLogger(ROCKETMQ_REMOTING);
private final io.netty.channel.Channel channel;
private final Channel channel;
public NettyChannelImpl(Channel channel) {
this.channel = channel;
......
remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java
浏览文件 @ f7c55b9f
......@@ -17,7 +17,6 @@
package org.apache.rocketmq.remoting.transport.http2;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
......@@ -41,26 +40,21 @@ import javax.net.ssl.SSLException;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.interceptor.InterceptorInvoker;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.NettyRemotingClientAbstract;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
import org.apache.rocketmq.remoting.util.ThreadUtils;
public class Http2ClientImpl extends NettyRemotingClientAbstract implements RemotingClient {
......@@ -248,7 +242,7 @@ public class Http2ClientImpl extends NettyRemotingClientAbstract implements Remo
@Override
public void registerProcessor(int requestCode, RequestProcessor processor, ExecutorService executor) {
executor = (executor == null ? this.publicExecutor : executor);
executor = executor == null ? this.publicExecutor : executor;
registerNettyProcessor(requestCode, processor, executor);
}
......
remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java
浏览文件 @ f7c55b9f
......@@ -48,24 +48,22 @@ import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.netty.ChannelStatisticsHandler;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.NettyRemotingServerAbstract;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
import org.apache.rocketmq.remoting.util.JvmUtils;
import org.apache.rocketmq.remoting.util.ThreadUtils;
......@@ -139,7 +137,7 @@ public class Http2ServerImpl extends NettyRemotingServerAbstract implements Remo
@Override
public void registerProcessor(int requestCode, RequestProcessor processor, ExecutorService executor) {
executor = (executor == null ? this.publicExecutor : executor);
executor = executor == null ? this.publicExecutor : executor;
registerNettyProcessor(requestCode, processor, executor);
}
......@@ -204,25 +202,23 @@ public class Http2ServerImpl extends NettyRemotingServerAbstract implements Remo
public void start() {
super.start();
final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
this.serverBootstrap.group(this.bossGroup, this.ioGroup).
channel(socketChannelClass).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
channels.add(ch);
ChannelPipeline cp = ch.pipeline();
cp.addLast(ChannelStatisticsHandler.NAME, new ChannelStatisticsHandler(channels));
cp.addLast(workerGroup,
Http2Handler.newHandler(true),
new NettyEncoder(),
new NettyDecoder(),
new IdleStateHandler(serverConfig.getConnectionChannelReaderIdleSeconds(),
serverConfig.getConnectionChannelWriterIdleSeconds(),
serverConfig.getServerChannelMaxIdleTimeSeconds()),
new NettyConnectManageHandler(),
new NettyServerHandler());
}
});
this.serverBootstrap.group(this.bossGroup, this.ioGroup).channel(socketChannelClass).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
channels.add(ch);
ChannelPipeline cp = ch.pipeline();
cp.addLast(ChannelStatisticsHandler.NAME, new ChannelStatisticsHandler(channels));
cp.addLast(workerGroup,
Http2Handler.newHandler(true),
new NettyEncoder(),
new NettyDecoder(),
new IdleStateHandler(serverConfig.getConnectionChannelReaderIdleSeconds(),
serverConfig.getConnectionChannelWriterIdleSeconds(),
serverConfig.getServerChannelMaxIdleTimeSeconds()),
new NettyConnectManageHandler(),
new NettyServerHandler());
}
});
applyOptions(serverBootstrap);
ChannelFuture channelFuture = this.serverBootstrap.bind(this.port).syncUninterruptibly();
......
remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingClient.java
浏览文件 @ f7c55b9f
......@@ -36,10 +36,11 @@ import java.util.concurrent.ExecutorService;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
......@@ -47,14 +48,7 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.interceptor.ExceptionContext;
import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.interceptor.InterceptorInvoker;
import org.apache.rocketmq.remoting.interceptor.RequestContext;
import org.apache.rocketmq.remoting.interceptor.ResponseContext;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.netty.TlsHelper;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
......
remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingServer.java
浏览文件 @ f7c55b9f
......@@ -52,7 +52,6 @@ import org.apache.rocketmq.remoting.common.TlsMode;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.netty.FileRegionEncoder;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
......@@ -222,7 +221,7 @@ public class NettyRemotingServer extends NettyRemotingServerAbstract implements
@Override
public void registerProcessor(int requestCode, RequestProcessor processor, ExecutorService executor) {
executor = (executor == null ? this.publicExecutor : executor);
executor = executor == null ? this.publicExecutor : executor;
registerNettyProcessor(requestCode, processor, executor);
}
......
remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java
浏览文件 @ f7c55b9f
......@@ -31,7 +31,7 @@ public class NettyRemotingClientTest {
private NettyRemotingClient remotingClient = new NettyRemotingClient(new ClientConfig());
@Test
public void testSetCallbackExecutor() throws NoSuchFieldException, IllegalAccessException {
public void testSetCallbackExecutor() {
ExecutorService customized = Executors.newCachedThreadPool();
remotingClient.setCallbackExecutor(customized);
......
remoting/src/test/java/org/apache/rocketmq/remoting/protocol/RemotingCommandTest.java
浏览文件 @ f7c55b9f
......@@ -109,7 +109,7 @@ public class RemotingCommandTest {
@Test
public void testNotNullField() throws Exception {
RemotingCommand remotingCommand = new RemotingCommand();
Method method = RemotingCommand.class.getDeclaredMethod("isFieldNullable", Field.class);
Method method = CodecHelper.class.getDeclaredMethod("isFieldNullable", Field.class);
method.setAccessible(true);
Field nullString = FieldTestClass.class.getDeclaredField("nullString");
......
snode/src/main/java/org/apache/rocketmq/snode/client/DefaultConsumerIdsChangeListener.java
浏览文件 @ f7c55b9f
......@@ -20,7 +20,9 @@ import java.util.List;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.snode.SnodeController;
//TODO Filter implementation
/**
* TODO Filter
*/
public class DefaultConsumerIdsChangeListener implements ConsumerIdsChangeListener {
private final SnodeController snodeController;
......
snode/src/main/java/org/apache/rocketmq/snode/constant/SnodeConstant.java
浏览文件 @ f7c55b9f
......@@ -15,12 +15,13 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.constant;
public class SnodeConstant {
public static final long heartbeatTimeout = 3000;
public static final long HEARTBEAT_TIME_OUT = 3000;
public static final long oneWaytimeout = 10;
public static final long ONE_WAY_TIMEOUT = 10;
public static final long defaultTimeoutMills = 3000L;
public static final long DEFAULT_TIMEOUT_MILLS = 3000L;
public static final long CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND = 1000 * 30;
......
snode/src/main/java/org/apache/rocketmq/snode/flowcontrol/RequestSizeFlowControlServiceImpl.java
浏览文件 @ f7c55b9f
......@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.snode.flowcontrol;
import com.sun.org.apache.bcel.internal.generic.IF_ACMPEQ;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.flowcontrol.AbstractFlowControlService;
......
snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
浏览文件 @ f7c55b9f
......@@ -61,7 +61,7 @@ public class HeartbeatProcessor implements RequestProcessor {
private RemotingCommand heartbeat(RemotingChannel remotingChannel, RemotingCommand request) {
HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);
ClientChannelInfo clientChannelInfo = new ClientChannelInfo(
new NettyChannelImpl((((NettyChannelHandlerContextImpl) remotingChannel).getChannelHandlerContext().channel())),
new NettyChannelImpl(((NettyChannelHandlerContextImpl) remotingChannel).getChannelHandlerContext().channel()),
heartbeatData.getClientID(),
request.getLanguage(),
request.getVersion()
......
snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java
浏览文件 @ f7c55b9f
......@@ -54,7 +54,6 @@ public class SendMessageProcessor implements RequestProcessor {
boolean isSendBack = false;
if (request.getCode() == RequestCode.SEND_MESSAGE_V2) {
sendMessageRequestHeaderV2 = (SendMessageRequestHeaderV2) request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
System.out.println("sendMessageRequestHeaderV2: " + sendMessageRequestHeaderV2);
enodeName = sendMessageRequestHeaderV2.getN();
} else {
isSendBack = true;
......
snode/src/main/java/org/apache/rocketmq/snode/service/impl/EnodeServiceImpl.java
浏览文件 @ f7c55b9f
......@@ -29,13 +29,10 @@ import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
import org.apache.rocketmq.common.protocol.header.CreateTopicRequestHeader;
import org.apache.rocketmq.common.protocol.header.GetMinOffsetRequestHeader;
import org.apache.rocketmq.common.protocol.header.NotifyConsumerIdsChangedRequestHeader;
import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeaderV2;
import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.logging.InternalLogger;
......@@ -71,7 +68,7 @@ public class EnodeServiceImpl implements EnodeService {
String enodeAddr = entry.getValue().get(MixAll.MASTER_ID);
if (enodeAddr != null) {
try {
this.snodeController.getRemotingClient().invokeSync(enodeAddr, remotingCommand, SnodeConstant.defaultTimeoutMills);
this.snodeController.getRemotingClient().invokeSync(enodeAddr, remotingCommand, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
} catch (Exception ex) {
log.warn("Send heart beat faild:{} ,ex:{}", enodeAddr, ex);
}
......@@ -114,7 +111,7 @@ public class EnodeServiceImpl implements EnodeService {
CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
try {
String enodeAddress = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
this.snodeController.getRemotingClient().invokeAsync(enodeAddress, request, SnodeConstant.defaultTimeoutMills, (responseFuture) -> {
this.snodeController.getRemotingClient().invokeAsync(enodeAddress, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS, (responseFuture) -> {
future.complete(responseFuture.getResponseCommand());
});
} catch (Exception ex) {
......@@ -139,7 +136,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand.createRequestCommand(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, requestHeader);
try {
this.snodeController.getSnodeServer().invokeOneway(channel, request, SnodeConstant.oneWaytimeout);
this.snodeController.getSnodeServer().invokeOneway(channel, request, SnodeConstant.ONE_WAY_TIMEOUT);
} catch (Exception e) {
log.error("NotifyConsumerIdsChanged consumer group: {} exception ", consumerGroup, e);
}
......@@ -164,7 +161,7 @@ public class EnodeServiceImpl implements EnodeService {
public void updateEnodeAddr(String clusterName) throws InterruptedException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException, MQBrokerException {
synchronized (this) {
ClusterInfo clusterInfo = getBrokerClusterInfo(SnodeConstant.defaultTimeoutMills);
ClusterInfo clusterInfo = getBrokerClusterInfo(SnodeConstant.DEFAULT_TIMEOUT_MILLS);
if (clusterInfo != null) {
HashMap<String, Set<String>> enodeAddress = clusterInfo.getClusterAddrTable();
for (Map.Entry<String, Set<String>> entry : enodeAddress.entrySet()) {
......@@ -189,7 +186,7 @@ public class EnodeServiceImpl implements EnodeService {
String enodeAddress = entry.getValue().get(MixAll.MASTER_ID);
try {
RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(enodeAddress,
request, SnodeConstant.defaultTimeoutMills);
request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
if (response != null && response.getCode() == ResponseCode.SUCCESS) {
persist = true;
} else {
......@@ -215,7 +212,7 @@ public class EnodeServiceImpl implements EnodeService {
requestHeader.setCommitOffset(offset);
requestHeader.setEnodeName(enodeName);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_CONSUMER_OFFSET, requestHeader);
this.snodeController.getRemotingClient().invokeOneway(address, request, SnodeConstant.defaultTimeoutMills);
this.snodeController.getRemotingClient().invokeOneway(address, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
} catch (Exception ex) {
log.error("Persist offset to Enode error!");
}
......@@ -231,7 +228,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_MIN_OFFSET, requestHeader);
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
request, SnodeConstant.defaultTimeoutMills);
request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
......@@ -246,7 +243,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUMER_OFFSET, requestHeader);
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(this.snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
request, SnodeConstant.defaultTimeoutMills);
request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
......@@ -254,7 +251,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, RemotingCommandException {
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
request, SnodeConstant.defaultTimeoutMills);
request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
......@@ -262,7 +259,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
request, SnodeConstant.defaultTimeoutMills);
request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
......@@ -279,6 +276,6 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
String address = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(address,
request, SnodeConstant.defaultTimeoutMills);
request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
}
snode/src/main/java/org/apache/rocketmq/snode/service/impl/NnodeServiceImpl.java
浏览文件 @ f7c55b9f
......@@ -66,7 +66,7 @@ public class NnodeServiceImpl implements NnodeService {
if (nnodeAddressList != null && nnodeAddressList.size() > 0) {
for (String nodeAddress : nnodeAddressList) {
try {
this.snodeController.getRemotingClient().invokeSync(nodeAddress, remotingCommand, SnodeConstant.heartbeatTimeout);
this.snodeController.getRemotingClient().invokeSync(nodeAddress, remotingCommand, SnodeConstant.HEARTBEAT_TIME_OUT);
} catch (Exception ex) {
log.warn("Register Snode to Nnode addr: {} error, ex:{} ", nodeAddress, ex);
}
......@@ -93,7 +93,7 @@ public class NnodeServiceImpl implements NnodeService {
requestHeader.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINTO_BY_TOPIC, requestHeader);
RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.defaultTimeoutMills);
RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
log.info("GetTopicRouteInfoFromNameServer response: " + response);
assert response != null;
switch (response.getCode()) {
......@@ -167,7 +167,7 @@ public class NnodeServiceImpl implements NnodeService {
RemotingSendRequestException, RemotingConnectException {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);
RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.defaultTimeoutMills);
RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ClusterInfo clusterInfo = ClusterInfo.decode(response.getBody(), ClusterInfo.class);
......
snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
浏览文件 @ f7c55b9f
......@@ -90,7 +90,7 @@ public class PushServiceImpl implements PushService {
RemotingChannel remotingChannel = clientChannelInfoEntry.getValue().getChannel();
if (remotingChannel.isWritable()) {
log.warn("Push message to topic: {} queueId: {} consumer group:{}, message:{}", topic, queueId, clientChannelInfoEntry.getKey(), pushMessage);
snodeController.getSnodeServer().push(remotingChannel, pushMessage, SnodeConstant.defaultTimeoutMills);
snodeController.getSnodeServer().push(remotingChannel, pushMessage, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
}
} else {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或