提交 703ac00b 编写于 作者: D dongeforever 提交者: yukon

[ROCKETMQ-220] Add IT test for Filter By SQL 92, closes

apache/incubator-rocketmq#114
上级 7374914b
...@@ -421,7 +421,7 @@ public class BrokerController { ...@@ -421,7 +421,7 @@ public class BrokerController {
this.fastRemotingServer.registerProcessor(RequestCode.HEART_BEAT, clientProcessor, this.clientManageExecutor); this.fastRemotingServer.registerProcessor(RequestCode.HEART_BEAT, clientProcessor, this.clientManageExecutor);
this.fastRemotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientProcessor, this.clientManageExecutor); this.fastRemotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientProcessor, this.clientManageExecutor);
this.remotingServer.registerProcessor(RequestCode.CHECK_CLIENT_CONFIG, clientProcessor, this.clientManageExecutor); this.fastRemotingServer.registerProcessor(RequestCode.CHECK_CLIENT_CONFIG, clientProcessor, this.clientManageExecutor);
/** /**
* ConsumerManageProcessor * ConsumerManageProcessor
......
...@@ -31,7 +31,6 @@ public class SqlConsumer { ...@@ -31,7 +31,6 @@ public class SqlConsumer {
public static void main(String[] args) { public static void main(String[] args) {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4"); DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
try { try {
consumer.subscribe("TopicTest", consumer.subscribe("TopicTest",
MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" + MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
......
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.test.client.rmq;
import org.apache.log4j.Logger;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.test.listener.AbstractListener;
public class RMQSqlConsumer extends RMQNormalConsumer {
private static Logger logger = Logger.getLogger(RMQSqlConsumer.class);
private MessageSelector selector;
public RMQSqlConsumer(String nsAddr, String topic, MessageSelector selector,
String consumerGroup, AbstractListener listener) {
super(nsAddr, topic, "*", consumerGroup, listener);
this.selector = selector;
}
@Override
public void create() {
super.create();
try {
consumer.subscribe(topic, selector);
} catch (Exception e) {
logger.error("Subscribe Sql Errored", e);
}
}
}
...@@ -17,8 +17,10 @@ ...@@ -17,8 +17,10 @@
package org.apache.rocketmq.test.factory; package org.apache.rocketmq.test.factory;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer; import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer; import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer;
import org.apache.rocketmq.test.client.rmq.RMQSqlConsumer;
import org.apache.rocketmq.test.listener.AbstractListener; import org.apache.rocketmq.test.listener.AbstractListener;
public class ConsumerFactory { public class ConsumerFactory {
...@@ -42,4 +44,14 @@ public class ConsumerFactory { ...@@ -42,4 +44,14 @@ public class ConsumerFactory {
consumer.start(); consumer.start();
return consumer; return consumer;
} }
public static RMQSqlConsumer getRMQSqlConsumer(String nsAddr, String consumerGroup,
String topic, MessageSelector selector,
AbstractListener listner) {
RMQSqlConsumer consumer = new RMQSqlConsumer(nsAddr, topic, selector,
consumerGroup, listner);
consumer.create();
consumer.start();
return consumer;
}
} }
...@@ -95,6 +95,28 @@ public class AbstractListener extends MQCollector implements MessageListener { ...@@ -95,6 +95,28 @@ public class AbstractListener extends MQCollector implements MessageListener {
return sendMsgs; return sendMsgs;
} }
public long waitForMessageConsume(int size,
int timeoutMills) {
long curTime = System.currentTimeMillis();
while (true) {
if (msgBodys.getDataSize() >= size) {
break;
}
if (System.currentTimeMillis() - curTime >= timeoutMills) {
logger.error(String.format("timeout but [%s] not recv all send messages!",
listnerName));
break;
} else {
logger.info(String.format("[%s] still [%s] msg not recv!", listnerName,
size - msgBodys.getDataSize()));
TestUtil.waitForMonment(500);
}
}
return msgBodys.getDataSize();
}
public void waitForMessageConsume(Map<Object, Object> sendMsgIndex, int timeoutMills) { public void waitForMessageConsume(Map<Object, Object> sendMsgIndex, int timeoutMills) {
Collection<Object> notRecvMsgs = waitForMessageConsume(sendMsgIndex.keySet(), timeoutMills); Collection<Object> notRecvMsgs = waitForMessageConsume(sendMsgIndex.keySet(), timeoutMills);
for (Object object : notRecvMsgs) { for (Object object : notRecvMsgs) {
......
...@@ -127,6 +127,7 @@ public class IntegrationTestBase { ...@@ -127,6 +127,7 @@ public class IntegrationTestBase {
brokerConfig.setBrokerName(BROKER_NAME_PREFIX + BROKER_INDEX.getAndIncrement()); brokerConfig.setBrokerName(BROKER_NAME_PREFIX + BROKER_INDEX.getAndIncrement());
brokerConfig.setBrokerIP1("127.0.0.1"); brokerConfig.setBrokerIP1("127.0.0.1");
brokerConfig.setNamesrvAddr(nsAddr); brokerConfig.setNamesrvAddr(nsAddr);
brokerConfig.setEnablePropertyFilter(true);
storeConfig.setStorePathRootDir(baseDir); storeConfig.setStorePathRootDir(baseDir);
storeConfig.setStorePathCommitLog(baseDir + SEP + "commitlog"); storeConfig.setStorePathCommitLog(baseDir + SEP + "commitlog");
storeConfig.setHaListenPort(8000 + random.nextInt(1000)); storeConfig.setHaListenPort(8000 + random.nextInt(1000));
......
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.test.client.consumer.filter;
import org.apache.log4j.Logger;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.test.base.BaseConf;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.normal.NormalMsgTwoSameGroupConsumerIT;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.client.rmq.RMQSqlConsumer;
import org.apache.rocketmq.test.factory.ConsumerFactory;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListner;
import org.apache.rocketmq.test.util.VerifyUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class SqlFilterIT extends BaseConf {
private static Logger logger = Logger.getLogger(SqlFilterIT.class);
private RMQNormalProducer producer = null;
private String topic = null;
@Before
public void setUp() {
topic = initTopic();
logger.info(String.format("use topic: %s;", topic));
producer = getProducer(nsAddr, topic);
}
@After
public void tearDown() {
super.shutDown();
}
@Test
public void testFilterConsumer() throws Exception {
int msgSize = 16;
String group = initConsumerGroup();
MessageSelector selector = MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))");
RMQSqlConsumer consumer = ConsumerFactory.getRMQSqlConsumer(nsAddr, group, topic, selector, new RMQNormalListner(group + "_1"));
Thread.sleep(3000);
producer.send("TagA", msgSize);
producer.send("TagB", msgSize);
producer.send("TagC", msgSize);
Assert.assertEquals("Not all sent succeeded", msgSize * 3, producer.getAllUndupMsgBody().size());
consumer.getListner().waitForMessageConsume(msgSize * 2, consumeTime);
assertThat(producer.getAllMsgBody())
.containsAllIn(VerifyUtils.getFilterdMessage(producer.getAllMsgBody(),
consumer.getListner().getAllMsgBody()));
assertThat(consumer.getListner().getAllMsgBody().size()).isEqualTo(msgSize * 2);
}
}
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册