diff --git a/common/src/test/java/org/apache/rocketmq/common/protocol/body/ConsumerConnectionTest.java b/common/src/test/java/org/apache/rocketmq/common/protocol/body/ConsumerConnectionTest.java new file mode 100644 index 0000000000000000000000000000000000000000..be1460ebcad353530dbdac7f8fa330812dfde937 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/protocol/body/ConsumerConnectionTest.java @@ -0,0 +1,83 @@ +/* + * 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.common.protocol.body; + +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +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.remoting.protocol.RemotingSerializable; +import org.junit.Test; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import static org.assertj.core.api.Assertions.assertThat; + +public class ConsumerConnectionTest { + + @Test + public void testFromJson() { + ConsumerConnection consumerConnection = new ConsumerConnection(); + HashSet connections = new HashSet(); + Connection conn = new Connection(); + connections.add(conn); + + ConcurrentHashMap subscriptionTable = new ConcurrentHashMap(); + SubscriptionData subscriptionData = new SubscriptionData(); + subscriptionTable.put("topicA", subscriptionData); + + ConsumeType consumeType = ConsumeType.CONSUME_ACTIVELY; + MessageModel messageModel = MessageModel.CLUSTERING; + ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET; + + consumerConnection.setConnectionSet(connections); + consumerConnection.setSubscriptionTable(subscriptionTable); + consumerConnection.setConsumeType(consumeType); + consumerConnection.setMessageModel(messageModel); + consumerConnection.setConsumeFromWhere(consumeFromWhere); + + String json = RemotingSerializable.toJson(consumerConnection, true); + ConsumerConnection fromJson = RemotingSerializable.fromJson(json, ConsumerConnection.class); + assertThat(fromJson.getConsumeType()).isEqualTo(ConsumeType.CONSUME_ACTIVELY); + assertThat(fromJson.getMessageModel()).isEqualTo(MessageModel.CLUSTERING); + + HashSet connectionSet = fromJson.getConnectionSet(); + assertThat(connectionSet).isInstanceOf(Set.class); + + SubscriptionData data = fromJson.getSubscriptionTable().get("topicA"); + assertThat(data).isExactlyInstanceOf(SubscriptionData.class); + } + + @Test + public void testComputeMinVersion() { + ConsumerConnection consumerConnection = new ConsumerConnection(); + HashSet connections = new HashSet(); + Connection conn1 = new Connection(); + conn1.setVersion(1); + connections.add(conn1); + Connection conn2 = new Connection(); + conn2.setVersion(10); + connections.add(conn2); + consumerConnection.setConnectionSet(connections); + + int version = consumerConnection.computeMinVersion(); + assertThat(version).isEqualTo(1); + } +} diff --git a/docs/en/Example_Orderly.md b/docs/en/Example_Orderly.md index 875e3de0bca0571424941a4e088a02435b84fb56..74012a662366478ef9ac45669a5ec2331cc17da8 100644 --- a/docs/en/Example_Orderly.md +++ b/docs/en/Example_Orderly.md @@ -4,7 +4,7 @@ RocketMQ provides ordered messages using FIFO order. All related messages need t The following demonstrates ordered messages by ensuring order of create, pay, send and finish steps of sales order process. -## 2.1 produce ordered messages +## 1 produce ordered messages ``` package org.apache.rocketmq.example.order2 diff --git a/store/src/test/java/org/apache/rocketmq/store/schedule/ScheduleMessageServiceTest.java b/store/src/test/java/org/apache/rocketmq/store/schedule/ScheduleMessageServiceTest.java index dc05c3b2976accb696fcef247b28d74f3773988a..c436aebdc01481213ab2ac1880eefe8536cde6f5 100644 --- a/store/src/test/java/org/apache/rocketmq/store/schedule/ScheduleMessageServiceTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/schedule/ScheduleMessageServiceTest.java @@ -16,6 +16,7 @@ */ package org.apache.rocketmq.store.schedule; + import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.UtilAll; import org.apache.rocketmq.common.message.MessageDecoder; @@ -45,17 +46,17 @@ import static org.assertj.core.api.Assertions.assertThat; public class ScheduleMessageServiceTest { - /**t + /** + * t * defaultMessageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" */ - String testMessageDelayLevel = "5s 10s"; + String testMessageDelayLevel = "2s 3s"; /** * choose delay level - * 1 = 5s */ - int delayLevel = 1; + int delayLevel = 2; - private static final String storePath = System.getProperty("user.home") + File.separator + "schedule_test"+ UUID.randomUUID(); + private static final String storePath = System.getProperty("user.home") + File.separator + "schedule_test#" + UUID.randomUUID(); private static final int commitLogFileSize = 1024; private static final int cqFileSize = 10; private static final int cqExtFileSize = 10 * (ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE + 64); @@ -67,7 +68,9 @@ public class ScheduleMessageServiceTest { BrokerConfig brokerConfig; ScheduleMessageService scheduleMessageService; - static String sendMessage = " ------- schedule message test -------"; + static String sendMessage = " ------- schedule message test -------"; + static String topic = "schedule_topic_test"; + static String messageGroup = "delayGroupTest"; static { @@ -107,67 +110,39 @@ public class ScheduleMessageServiceTest { } - - @Test - public void buildRunningStatsTest() throws InterruptedException { - MessageExtBrokerInner msg = buildMessage(); - msg.setDelayTimeLevel(delayLevel); - messageStore.putMessage(msg); - // wait offsetTable - TimeUnit.SECONDS.sleep(1); - scheduleMessageService.buildRunningStats(new HashMap() ); - } - - - @Test - public void computeDeliverTimestampTest() { - // testMessageDelayLevel just "5s 10s" - long storeTime = System.currentTimeMillis(); - long time1 = scheduleMessageService.computeDeliverTimestamp(1, storeTime); - assertThat(time1).isEqualTo(storeTime + 5 * 1000); - - long time2 = scheduleMessageService.computeDeliverTimestamp(2, storeTime); - assertThat(time2).isEqualTo(storeTime + 10 * 1000); - - } - - - @Test - public void delayLevel2QueueIdTest() { - int queueId = ScheduleMessageService.delayLevel2QueueId(delayLevel); - assertThat(queueId).isEqualTo(delayLevel - 1); - queueId = ScheduleMessageService.queueId2DelayLevel(delayLevel); - assertThat(queueId).isEqualTo(delayLevel + 1); - } - @Test public void deliverDelayedMessageTimerTaskTest() throws InterruptedException { MessageExtBrokerInner msg = buildMessage(); + int realQueueId = msg.getQueueId(); // set delayLevel,and send delay message msg.setDelayTimeLevel(delayLevel); PutMessageResult result = messageStore.putMessage(msg); assertThat(result.isOk()).isTrue(); + // consumer message + int delayQueueId = ScheduleMessageService.delayLevel2QueueId(delayLevel); + assertThat(delayQueueId).isEqualTo(delayLevel - 1); + Long offset = result.getAppendMessageResult().getLogicsOffset(); - String messageGroup = "delayGroupTest"; - GetMessageResult messageResult = messageStore.getMessage(messageGroup,msg.getTopic(), - msg.getQueueId(),offset,1,null); - // now, no message in queue,must wait > 5 seconds + // now, no message in queue,must wait > delayTime + GetMessageResult messageResult = getMessage(realQueueId, offset); assertThat(messageResult.getStatus()).isEqualTo(GetMessageStatus.NO_MESSAGE_IN_QUEUE); + // timer run maybe delay, then consumer message again + // and wait offsetTable + TimeUnit.SECONDS.sleep(3); + scheduleMessageService.buildRunningStats(new HashMap()); - TimeUnit.SECONDS.sleep(6); - messageResult = messageStore.getMessage(messageGroup,msg.getTopic(), - msg.getQueueId(),offset,1,null); + messageResult = getMessage(realQueueId, offset); // now,found the message assertThat(messageResult.getStatus()).isEqualTo(GetMessageStatus.FOUND); // get the message body ByteBuffer byteBuffer = ByteBuffer.allocate(messageResult.getBufferTotalSize()); - List byteBufferList = messageResult.getMessageBufferList(); + List byteBufferList = messageResult.getMessageBufferList(); for (ByteBuffer bb : byteBufferList) { byteBuffer.put(bb); } @@ -178,22 +153,42 @@ public class ScheduleMessageServiceTest { String retryMsg = new String(msgList.get(0).getBody()); assertThat(sendMessage).isEqualTo(retryMsg); + // method will wait 10s,so I run it by myself + scheduleMessageService.persist(); + // add mapFile release messageResult.release(); } + /** + * add some [error/no use] code test + */ @Test - public void persist(){ - // because of the method will wait 10s - scheduleMessageService.persist(); + public void otherTest() { + // the method no use ,why need ? + int queueId = ScheduleMessageService.queueId2DelayLevel(delayLevel); + assertThat(queueId).isEqualTo(delayLevel + 1); + + // error delayLevelTest + Long time = scheduleMessageService.computeDeliverTimestamp(999, 0); + assertThat(time).isEqualTo(1000); + + // just decode + scheduleMessageService.decode(new DelayOffsetSerializeWrapper().toJson()); + } + + + private GetMessageResult getMessage(int queueId, Long offset) { + return messageStore.getMessage(messageGroup, topic, + queueId, offset, 1, null); + } @After public void shutdown() throws InterruptedException { TimeUnit.SECONDS.sleep(1); - scheduleMessageService.shutdown(); messageStore.shutdown(); messageStore.destroy(); File file = new File(messageStoreConfig.getStorePathRootDir()); @@ -205,11 +200,10 @@ public class ScheduleMessageServiceTest { byte[] msgBody = sendMessage.getBytes(); MessageExtBrokerInner msg = new MessageExtBrokerInner(); - msg.setTopic("schedule_topic_test"); + msg.setTopic(topic); msg.setTags("schedule_tag"); msg.setKeys("schedule_key"); msg.setBody(msgBody); - msg.setQueueId(0); msg.setSysFlag(0); msg.setBornTimestamp(System.currentTimeMillis()); msg.setStoreHost(storeHost); @@ -226,6 +220,4 @@ public class ScheduleMessageServiceTest { } - - }