Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Wheat_Liu
Rocketmq
提交
f619e451
R
Rocketmq
项目概览
Wheat_Liu
/
Rocketmq
与 Fork 源项目一致
Fork自
Apache RocketMQ / Rocketmq
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
Rocketmq
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
f619e451
编写于
6月 15, 2017
作者:
S
shroman
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Fixed typo.
上级
c4a3e0c1
变更
42
隐藏空白更改
内联
并排
Showing
42 changed file
with
272 addition
and
275 deletion
+272
-275
test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQNormalConsumer.java
...rg/apache/rocketmq/test/client/rmq/RMQNormalConsumer.java
+6
-6
test/src/main/java/org/apache/rocketmq/test/clientinterface/AbstractMQConsumer.java
...che/rocketmq/test/clientinterface/AbstractMQConsumer.java
+12
-12
test/src/main/java/org/apache/rocketmq/test/listener/AbstractListener.java
...a/org/apache/rocketmq/test/listener/AbstractListener.java
+7
-7
test/src/main/java/org/apache/rocketmq/test/listener/rmq/concurrent/RMQDelayListner.java
...ocketmq/test/listener/rmq/concurrent/RMQDelayListner.java
+1
-1
test/src/main/java/org/apache/rocketmq/test/listener/rmq/concurrent/RMQNormalListner.java
...cketmq/test/listener/rmq/concurrent/RMQNormalListner.java
+2
-2
test/src/main/java/org/apache/rocketmq/test/listener/rmq/order/RMQOrderListener.java
...he/rocketmq/test/listener/rmq/order/RMQOrderListener.java
+2
-2
test/src/test/java/org/apache/rocketmq/test/client/consumer/balance/NormalMsgDynamicBalanceIT.java
...st/client/consumer/balance/NormalMsgDynamicBalanceIT.java
+10
-10
test/src/test/java/org/apache/rocketmq/test/client/consumer/balance/NormalMsgStaticBalanceIT.java
...est/client/consumer/balance/NormalMsgStaticBalanceIT.java
+9
-9
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgNotRecvIT.java
...onsumer/broadcast/normal/BroadCastNormalMsgNotRecvIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgRecvCrashIT.java
...sumer/broadcast/normal/BroadCastNormalMsgRecvCrashIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgRecvFailIT.java
...nsumer/broadcast/normal/BroadCastNormalMsgRecvFailIT.java
+2
-2
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgRecvStartLaterIT.java
.../broadcast/normal/BroadCastNormalMsgRecvStartLaterIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgTwoDiffGroupRecvIT.java
...roadcast/normal/BroadCastNormalMsgTwoDiffGroupRecvIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/NormalMsgTwoSameGroupConsumerIT.java
...mer/broadcast/normal/NormalMsgTwoSameGroupConsumerIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/order/OrderMsgBroadCastIT.java
.../client/consumer/broadcast/order/OrderMsgBroadCastIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/tag/BroadCastTwoConsumerFilterIT.java
.../consumer/broadcast/tag/BroadCastTwoConsumerFilterIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/tag/BroadCastTwoConsumerSubDiffTagIT.java
...sumer/broadcast/tag/BroadCastTwoConsumerSubDiffTagIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/tag/BroadCastTwoConsumerSubTagIT.java
.../consumer/broadcast/tag/BroadCastTwoConsumerSubTagIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/consumer/cluster/DynamicAddAndCrashIT.java
...mq/test/client/consumer/cluster/DynamicAddAndCrashIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/consumer/cluster/DynamicAddConsumerIT.java
...mq/test/client/consumer/cluster/DynamicAddConsumerIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/consumer/cluster/DynamicCrashConsumerIT.java
.../test/client/consumer/cluster/DynamicCrashConsumerIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
...che/rocketmq/test/client/consumer/filter/SqlFilterIT.java
+3
-6
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/MulTagSubIT.java
...apache/rocketmq/test/client/consumer/tag/MulTagSubIT.java
+9
-9
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWith1ConsumerIT.java
...q/test/client/consumer/tag/TagMessageWith1ConsumerIT.java
+16
-16
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWithMulConsumerIT.java
...test/client/consumer/tag/TagMessageWithMulConsumerIT.java
+20
-20
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWithSameGroupConsumerIT.java
...lient/consumer/tag/TagMessageWithSameGroupConsumerIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/consumer/topic/MulConsumerMulTopicIT.java
...tmq/test/client/consumer/topic/MulConsumerMulTopicIT.java
+3
-3
test/src/test/java/org/apache/rocketmq/test/client/consumer/topic/OneConsumerMulTopicIT.java
...tmq/test/client/consumer/topic/OneConsumerMulTopicIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueIT.java
...st/client/producer/async/AsyncSendWithMessageQueueIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueSelectorIT.java
...t/producer/async/AsyncSendWithMessageQueueSelectorIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithOnlySendCallBackIT.java
...lient/producer/async/AsyncSendWithOnlySendCallBackIT.java
+2
-2
test/src/test/java/org/apache/rocketmq/test/client/producer/exception/msg/MessageUserPropIT.java
...test/client/producer/exception/msg/MessageUserPropIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/producer/oneway/OneWaySendIT.java
...he/rocketmq/test/client/producer/oneway/OneWaySendIT.java
+2
-2
test/src/test/java/org/apache/rocketmq/test/client/producer/oneway/OneWaySendWithMQIT.java
...ketmq/test/client/producer/oneway/OneWaySendWithMQIT.java
+4
-4
test/src/test/java/org/apache/rocketmq/test/client/producer/oneway/OneWaySendWithSelectorIT.java
...test/client/producer/oneway/OneWaySendWithSelectorIT.java
+6
-6
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgDynamicRebalanceIT.java
...est/client/producer/order/OrderMsgDynamicRebalanceIT.java
+11
-11
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgIT.java
...pache/rocketmq/test/client/producer/order/OrderMsgIT.java
+9
-9
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgRebalanceIT.java
...ketmq/test/client/producer/order/OrderMsgRebalanceIT.java
+19
-19
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgWithTagIT.java
...ocketmq/test/client/producer/order/OrderMsgWithTagIT.java
+14
-14
test/src/test/java/org/apache/rocketmq/test/client/producer/querymsg/QueryMsgByIdIT.java
...ocketmq/test/client/producer/querymsg/QueryMsgByIdIT.java
+3
-3
test/src/test/java/org/apache/rocketmq/test/delay/NormalMsgDelayIT.java
...java/org/apache/rocketmq/test/delay/NormalMsgDelayIT.java
+12
-12
test/src/test/java/org/apache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java
...pache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java
+2
-2
未找到文件。
test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQNormalConsumer.java
浏览文件 @
f619e451
...
...
@@ -33,12 +33,12 @@ public class RMQNormalConsumer extends AbstractMQConsumer {
super
(
nsAddr
,
topic
,
subExpression
,
consumerGroup
,
listner
);
}
public
AbstractListener
getListner
()
{
return
listner
;
public
AbstractListener
getList
e
ner
()
{
return
list
e
ner
;
}
public
void
setList
ner
(
AbstractListener
list
ner
)
{
this
.
list
ner
=
list
ner
;
public
void
setList
ener
(
AbstractListener
liste
ner
)
{
this
.
list
ener
=
liste
ner
;
}
public
void
create
()
{
...
...
@@ -51,7 +51,7 @@ public class RMQNormalConsumer extends AbstractMQConsumer {
logger
.
error
(
"consumer subscribe failed!"
);
e
.
printStackTrace
();
}
consumer
.
setMessageListener
(
listner
);
consumer
.
setMessageListener
(
list
e
ner
);
}
public
void
start
()
{
...
...
@@ -79,7 +79,7 @@ public class RMQNormalConsumer extends AbstractMQConsumer {
@Override
public
void
clearMsg
()
{
this
.
listner
.
clearMsg
();
this
.
list
e
ner
.
clearMsg
();
}
public
void
restart
()
{
...
...
test/src/main/java/org/apache/rocketmq/test/clientinterface/AbstractMQConsumer.java
浏览文件 @
f619e451
...
...
@@ -20,7 +20,7 @@ package org.apache.rocketmq.test.clientinterface;
import
org.apache.rocketmq.test.listener.AbstractListener
;
public
abstract
class
AbstractMQConsumer
implements
MQConsumer
{
protected
AbstractListener
listner
=
null
;
protected
AbstractListener
list
e
ner
=
null
;
protected
String
nsAddr
=
null
;
protected
String
topic
=
null
;
protected
String
subExpression
=
null
;
...
...
@@ -31,11 +31,11 @@ public abstract class AbstractMQConsumer implements MQConsumer {
}
public
AbstractMQConsumer
(
String
nsAddr
,
String
topic
,
String
subExpression
,
String
consumerGroup
,
AbstractListener
listner
)
{
String
consumerGroup
,
AbstractListener
list
e
ner
)
{
this
.
topic
=
topic
;
this
.
subExpression
=
subExpression
;
this
.
consumerGroup
=
consumerGroup
;
this
.
list
ner
=
list
ner
;
this
.
list
ener
=
liste
ner
;
this
.
nsAddr
=
nsAddr
;
}
...
...
@@ -45,16 +45,16 @@ public abstract class AbstractMQConsumer implements MQConsumer {
}
public
void
setDebug
()
{
if
(
listner
!=
null
)
{
listner
.
setDebug
(
true
);
if
(
list
e
ner
!=
null
)
{
list
e
ner
.
setDebug
(
true
);
}
isDebug
=
true
;
}
public
void
setDebug
(
boolean
isDebug
)
{
if
(
listner
!=
null
)
{
listner
.
setDebug
(
isDebug
);
if
(
list
e
ner
!=
null
)
{
list
e
ner
.
setDebug
(
isDebug
);
}
this
.
isDebug
=
isDebug
;
...
...
@@ -65,12 +65,12 @@ public abstract class AbstractMQConsumer implements MQConsumer {
this
.
subExpression
=
subExpression
;
}
public
AbstractListener
getListner
()
{
return
listner
;
public
AbstractListener
getList
e
ner
()
{
return
list
e
ner
;
}
public
void
setList
ner
(
AbstractListener
list
ner
)
{
this
.
list
ner
=
list
ner
;
public
void
setList
ener
(
AbstractListener
liste
ner
)
{
this
.
list
ener
=
liste
ner
;
}
public
String
getNsAddr
()
{
...
...
@@ -106,7 +106,7 @@ public abstract class AbstractMQConsumer implements MQConsumer {
}
public
void
clearMsg
()
{
listner
.
clearMsg
();
list
e
ner
.
clearMsg
();
}
}
test/src/main/java/org/apache/rocketmq/test/listener/AbstractListener.java
浏览文件 @
f619e451
...
...
@@ -30,16 +30,16 @@ import org.apache.rocketmq.test.util.TestUtil;
public
class
AbstractListener
extends
MQCollector
implements
MessageListener
{
public
static
Logger
logger
=
Logger
.
getLogger
(
AbstractListener
.
class
);
protected
boolean
isDebug
=
false
;
protected
String
listnerName
=
null
;
protected
String
list
e
nerName
=
null
;
protected
Collection
<
Object
>
allSendMsgs
=
null
;
public
AbstractListener
()
{
super
();
}
public
AbstractListener
(
String
listnerName
)
{
public
AbstractListener
(
String
list
e
nerName
)
{
super
();
this
.
list
nerName
=
list
nerName
;
this
.
list
enerName
=
liste
nerName
;
}
public
AbstractListener
(
String
originMsgCollector
,
String
msgBodyCollector
)
{
...
...
@@ -82,10 +82,10 @@ public class AbstractListener extends MQCollector implements MessageListener {
}
else
{
if
(
System
.
currentTimeMillis
()
-
curTime
>=
timeoutMills
)
{
logger
.
error
(
String
.
format
(
"timeout but [%s] not recv all send messages!"
,
listnerName
));
list
e
nerName
));
break
;
}
else
{
logger
.
info
(
String
.
format
(
"[%s] still [%s] msg not recv!"
,
listnerName
,
logger
.
info
(
String
.
format
(
"[%s] still [%s] msg not recv!"
,
list
e
nerName
,
sendMsgs
.
size
()));
TestUtil
.
waitForMonment
(
500
);
}
...
...
@@ -105,10 +105,10 @@ public class AbstractListener extends MQCollector implements MessageListener {
}
if
(
System
.
currentTimeMillis
()
-
curTime
>=
timeoutMills
)
{
logger
.
error
(
String
.
format
(
"timeout but [%s] not recv all send messages!"
,
listnerName
));
list
e
nerName
));
break
;
}
else
{
logger
.
info
(
String
.
format
(
"[%s] still [%s] msg not recv!"
,
listnerName
,
logger
.
info
(
String
.
format
(
"[%s] still [%s] msg not recv!"
,
list
e
nerName
,
size
-
msgBodys
.
getDataSize
()));
TestUtil
.
waitForMonment
(
500
);
}
...
...
test/src/main/java/org/apache/rocketmq/test/listener/rmq/concurrent/RMQDelayListner.java
浏览文件 @
f619e451
...
...
@@ -49,7 +49,7 @@ public class RMQDelayListner extends AbstractListener implements MessageListener
long
recvTime
=
System
.
currentTimeMillis
();
for
(
MessageExt
msg
:
msgs
)
{
if
(
isDebug
)
{
logger
.
info
(
listnerName
+
":"
+
msg
);
logger
.
info
(
list
e
nerName
+
":"
+
msg
);
}
msgBodys
.
addData
(
new
String
(
msg
.
getBody
()));
...
...
test/src/main/java/org/apache/rocketmq/test/listener/rmq/concurrent/RMQNormalListner.java
浏览文件 @
f619e451
...
...
@@ -51,8 +51,8 @@ public class RMQNormalListner extends AbstractListener implements MessageListene
for
(
MessageExt
msg
:
msgs
)
{
msgIndex
.
getAndIncrement
();
if
(
isDebug
)
{
if
(
list
nerName
!=
null
&&
list
nerName
!=
""
)
{
logger
.
info
(
listnerName
+
":"
+
msgIndex
.
get
()
+
":"
if
(
list
enerName
!=
null
&&
liste
nerName
!=
""
)
{
logger
.
info
(
list
e
nerName
+
":"
+
msgIndex
.
get
()
+
":"
+
String
.
format
(
"msgid:%s broker:%s queueId:%s offset:%s"
,
msg
.
getMsgId
(),
msg
.
getStoreHost
(),
msg
.
getQueueId
(),
msg
.
getQueueOffset
()));
...
...
test/src/main/java/org/apache/rocketmq/test/listener/rmq/order/RMQOrderListener.java
浏览文件 @
f619e451
...
...
@@ -68,8 +68,8 @@ public class RMQOrderListener extends AbstractListener implements MessageListene
ConsumeOrderlyContext
context
)
{
for
(
MessageExt
msg
:
msgs
)
{
if
(
isDebug
)
{
if
(
list
nerName
!=
null
&&
list
nerName
!=
""
)
{
logger
.
info
(
listnerName
+
": "
+
msg
);
if
(
list
enerName
!=
null
&&
liste
nerName
!=
""
)
{
logger
.
info
(
list
e
nerName
+
": "
+
msg
);
}
else
{
logger
.
info
(
msg
);
}
...
...
test/src/test/java/org/apache/rocketmq/test/client/consumer/balance/NormalMsgDynamicBalanceIT.java
浏览文件 @
f619e451
...
...
@@ -57,22 +57,22 @@ public class NormalMsgDynamicBalanceIT extends BaseConf {
producer
.
send
(
msgSize
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getListner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
e
ner
());
consumer2
.
shutdown
();
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
boolean
balance
=
VerifyUtils
.
verifyBalance
(
msgSize
,
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllUndupMsgBody
()).
size
()
-
msgSize
,
consumer1
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
()
-
msgSize
,
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllUndupMsgBody
()).
size
());
consumer2
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
());
assertThat
(
balance
).
isEqualTo
(
true
);
}
...
...
@@ -87,8 +87,8 @@ public class NormalMsgDynamicBalanceIT extends BaseConf {
producer
.
send
(
msgSize
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getList
ner
(),
consumer3
.
getList
ner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
ener
(),
consumer3
.
getListe
ner
());
consumer3
.
shutdown
();
producer
.
clearMsg
();
consumer1
.
clearMsg
();
...
...
@@ -98,14 +98,14 @@ public class NormalMsgDynamicBalanceIT extends BaseConf {
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
boolean
balance
=
VerifyUtils
.
verifyBalance
(
msgSize
,
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllUndupMsgBody
()).
size
(),
consumer1
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllUndupMsgBody
()).
size
());
consumer2
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
());
assertThat
(
balance
).
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/balance/NormalMsgStaticBalanceIT.java
浏览文件 @
f619e451
...
...
@@ -61,14 +61,14 @@ public class NormalMsgStaticBalanceIT extends BaseConf {
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
boolean
balance
=
VerifyUtils
.
verifyBalance
(
msgSize
,
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllUndupMsgBody
()).
size
(),
consumer1
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllUndupMsgBody
()).
size
());
consumer2
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
());
assertThat
(
balance
).
isEqualTo
(
true
);
}
...
...
@@ -88,22 +88,22 @@ public class NormalMsgStaticBalanceIT extends BaseConf {
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getListner
(),
consumer3
.
getList
ner
(),
consumer4
.
getListner
());
consumer1
.
getList
ener
(),
consumer2
.
getListener
(),
consumer3
.
getListe
ner
(),
consumer4
.
getList
e
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
boolean
balance
=
VerifyUtils
.
verifyBalance
(
msgSize
,
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllUndupMsgBody
())
consumer1
.
getList
e
ner
().
getAllUndupMsgBody
())
.
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllUndupMsgBody
()).
size
(),
consumer2
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer3
.
getListner
().
getAllUndupMsgBody
()).
size
(),
consumer3
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer4
.
getListner
().
getAllUndupMsgBody
()).
size
());
consumer4
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
());
assertThat
(
balance
).
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgNotRecvIT.java
浏览文件 @
f619e451
...
...
@@ -60,14 +60,14 @@ public class BroadCastNormalMsgNotRecvIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
RMQBroadCastConsumer
consumer2
=
getBroadCastConsumer
(
nsAddr
,
consumer1
.
getConsumerGroup
(),
topic
,
"*"
,
new
RMQNormalListner
(
group
+
"_2"
));
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
waitTime
);
assertThat
(
consumer2
.
getListner
().
getAllMsgBody
().
size
()).
isEqualTo
(
0
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
waitTime
);
assertThat
(
consumer2
.
getList
e
ner
().
getAllMsgBody
().
size
()).
isEqualTo
(
0
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgRecvCrashIT.java
浏览文件 @
f619e451
...
...
@@ -64,13 +64,13 @@ public class BroadCastNormalMsgRecvCrashIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllMsgBody
()))
consumer2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
consumer2
.
shutdown
();
...
...
@@ -81,9 +81,9 @@ public class BroadCastNormalMsgRecvCrashIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgRecvFailIT.java
浏览文件 @
f619e451
...
...
@@ -63,10 +63,10 @@ public class BroadCastNormalMsgRecvFailIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgRecvStartLaterIT.java
浏览文件 @
f619e451
...
...
@@ -62,9 +62,9 @@ public class BroadCastNormalMsgRecvStartLaterIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
producer
.
clearMsg
();
...
...
@@ -76,13 +76,13 @@ public class BroadCastNormalMsgRecvStartLaterIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllMsgBody
()))
consumer2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/BroadCastNormalMsgTwoDiffGroupRecvIT.java
浏览文件 @
f619e451
...
...
@@ -65,14 +65,14 @@ public class BroadCastNormalMsgTwoDiffGroupRecvIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllMsgBody
()))
consumer2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/normal/NormalMsgTwoSameGroupConsumerIT.java
浏览文件 @
f619e451
...
...
@@ -64,14 +64,14 @@ public class NormalMsgTwoSameGroupConsumerIT extends BaseBroadCastIT {
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllMsgBody
()))
consumer2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/order/OrderMsgBroadCastIT.java
浏览文件 @
f619e451
...
...
@@ -64,12 +64,12 @@ public class OrderMsgBroadCastIT extends BaseBroadCastIT {
MessageQueueMsg
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/tag/BroadCastTwoConsumerFilterIT.java
浏览文件 @
f619e451
...
...
@@ -65,14 +65,14 @@ public class BroadCastTwoConsumerFilterIT extends BaseBroadCastIT {
producer
.
clearMsg
();
producer
.
send
(
tag1
,
msgSize
);
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllMsgBody
()))
consumer2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/tag/BroadCastTwoConsumerSubDiffTagIT.java
浏览文件 @
f619e451
...
...
@@ -62,14 +62,14 @@ public class BroadCastTwoConsumerSubDiffTagIT extends BaseBroadCastIT {
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllMsgBody
()))
consumer2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/tag/BroadCastTwoConsumerSubTagIT.java
浏览文件 @
f619e451
...
...
@@ -62,14 +62,14 @@ public class BroadCastTwoConsumerSubTagIT extends BaseBroadCastIT {
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer2
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllMsgBody
()))
consumer2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/cluster/DynamicAddAndCrashIT.java
浏览文件 @
f619e451
...
...
@@ -65,11 +65,11 @@ public class DynamicAddAndCrashIT extends BaseConf {
asyncDefaultMQProducer
.
waitSendAll
(
waitTime
*
6
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getListner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
e
ner
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
...
...
@@ -93,11 +93,11 @@ public class DynamicAddAndCrashIT extends BaseConf {
asyncDefaultMQProducer
.
waitSendAll
(
waitTime
*
6
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getList
ner
(),
consumer3
.
getList
ner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
ener
(),
consumer3
.
getListe
ner
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getListner
(),
consumer3
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListener
(),
consumer3
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/cluster/DynamicAddConsumerIT.java
浏览文件 @
f619e451
...
...
@@ -63,11 +63,11 @@ public class DynamicAddConsumerIT extends BaseConf {
asyncDefaultMQProducer
.
waitSendAll
(
waitTime
*
6
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getListner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
e
ner
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
...
...
@@ -87,11 +87,11 @@ public class DynamicAddConsumerIT extends BaseConf {
asyncDefaultMQProducer
.
waitSendAll
(
waitTime
*
6
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getList
ner
(),
consumer3
.
getList
ner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
ener
(),
consumer3
.
getListe
ner
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getListner
(),
consumer3
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListener
(),
consumer3
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/cluster/DynamicCrashConsumerIT.java
浏览文件 @
f619e451
...
...
@@ -64,11 +64,11 @@ public class DynamicCrashConsumerIT extends BaseConf {
asyncDefaultMQProducer
.
waitSendAll
(
waitTime
*
6
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getListner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
e
ner
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
...
...
@@ -90,11 +90,11 @@ public class DynamicCrashConsumerIT extends BaseConf {
asyncDefaultMQProducer
.
waitSendAll
(
waitTime
*
6
);
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getList
ner
(),
consumer3
.
getList
ner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
ener
(),
consumer3
.
getListe
ner
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getListner
(),
consumer3
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListener
(),
consumer3
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
浏览文件 @
f619e451
...
...
@@ -20,9 +20,6 @@ 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
;
...
...
@@ -64,11 +61,11 @@ public class SqlFilterIT extends BaseConf {
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
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
msgSize
*
2
,
consumeTime
);
assertThat
(
producer
.
getAllMsgBody
())
.
containsAllIn
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()));
consumer
.
getList
e
ner
().
getAllMsgBody
()));
assertThat
(
consumer
.
getListner
().
getAllMsgBody
().
size
()).
isEqualTo
(
msgSize
*
2
);
assertThat
(
consumer
.
getList
e
ner
().
getAllMsgBody
().
size
()).
isEqualTo
(
msgSize
*
2
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/MulTagSubIT.java
浏览文件 @
f619e451
...
...
@@ -61,10 +61,10 @@ public class MulTagSubIT extends BaseConf {
new
RMQNormalListner
());
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -83,10 +83,10 @@ public class MulTagSubIT extends BaseConf {
producer
.
send
(
tag2Msgs
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
),
consumer
.
getList
e
ner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
));
}
...
...
@@ -104,10 +104,10 @@ public class MulTagSubIT extends BaseConf {
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
*
tags
.
length
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getAllTagMessageBody
());
}
...
...
@@ -125,11 +125,11 @@ public class MulTagSubIT extends BaseConf {
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
*
tags
.
length
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
consumer
.
getList
e
ner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
],
tags
[
1
]),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
())).
containsExactlyElementsIn
(
consumer
.
getList
e
ner
().
getAllMsgBody
())).
containsExactlyElementsIn
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
],
tags
[
1
]));
}
...
...
@@ -150,7 +150,7 @@ public class MulTagSubIT extends BaseConf {
TestUtils
.
waitForSeconds
(
5
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
())
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getList
e
ner
().
getAllMsgBody
())
.
size
()).
isEqualTo
(
0
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWith1ConsumerIT.java
浏览文件 @
f619e451
...
...
@@ -57,9 +57,9 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
RMQNormalConsumer
consumer
=
getConsumer
(
nsAddr
,
topic
,
tag
,
new
RMQNormalListner
());
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -71,10 +71,10 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
new
RMQNormalListner
());
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -87,10 +87,10 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
new
RMQNormalListner
());
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -103,10 +103,10 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
new
RMQNormalListner
());
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -119,10 +119,10 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
new
RMQNormalListner
());
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -142,10 +142,10 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
producer
.
send
(
tag2Msgs
);
producer
.
send
(
10
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
3
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -164,10 +164,10 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
producer
.
send
(
tag1Msgs
);
producer
.
send
(
tag2Msgs
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -187,11 +187,11 @@ public class TagMessageWith1ConsumerIT extends BaseConf {
producer
.
send
(
tag2Msgs
);
producer
.
send
(
10
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
3
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
),
consumer
.
getList
e
ner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
));
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWithMulConsumerIT.java
浏览文件 @
f619e451
...
...
@@ -69,16 +69,16 @@ public class TagMessageWithMulConsumerIT extends BaseConf {
producer
.
send
(
tag2Msgs
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
consumerTag1
.
getListner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag1Msgs
),
consumerTag1
.
getList
e
ner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag1Msgs
),
consumeTime
);
consumerTag2
.
getListner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
),
consumerTag2
.
getList
e
ner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerTag1
.
getListner
().
getAllMsgBody
()))
consumerTag1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
MQMessageFactory
.
getMessageBody
(
tag1Msgs
));
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerTag2
.
getListner
().
getAllMsgBody
()))
consumerTag2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
MQMessageFactory
.
getMessageBody
(
tag2Msgs
));
}
...
...
@@ -98,16 +98,16 @@ public class TagMessageWithMulConsumerIT extends BaseConf {
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
tags
.
length
,
producer
.
getAllUndupMsgBody
().
size
());
consumerTag1
.
getListner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]),
consumerTag1
.
getList
e
ner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]),
consumeTime
);
consumerTag2
.
getListner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
1
]),
consumerTag2
.
getList
e
ner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
1
]),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerTag1
.
getListner
().
getAllMsgBody
()))
consumerTag1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]));
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerTag2
.
getListner
().
getAllMsgBody
()))
consumerTag2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getMessageBodyByTag
(
tags
[
1
]));
}
...
...
@@ -129,16 +129,16 @@ public class TagMessageWithMulConsumerIT extends BaseConf {
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
tags
.
length
,
producer
.
getAllUndupMsgBody
().
size
());
consumerTag1
.
getListner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
),
consumerTag1
.
getList
e
ner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
),
consumeTime
);
consumerTag2
.
getListner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]),
consumerTag2
.
getList
e
ner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerTag1
.
getListner
().
getAllMsgBody
()))
consumerTag1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getAllTagMessageBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerTag2
.
getListner
().
getAllMsgBody
()))
consumerTag2
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]));
}
...
...
@@ -169,27 +169,27 @@ public class TagMessageWithMulConsumerIT extends BaseConf {
producer
.
send
(
tagMsgs
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
3
,
producer
.
getAllUndupMsgBody
().
size
());
consumerSubTwoMatchAll
.
getListner
()
consumerSubTwoMatchAll
.
getList
e
ner
()
.
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
),
consumeTime
);
consumerSubTwoMachieOne
.
getListner
()
consumerSubTwoMachieOne
.
getList
e
ner
()
.
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]),
consumeTime
);
consumerSubTag1
.
getListner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]),
consumerSubTag1
.
getList
e
ner
().
waitForMessageConsume
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]),
consumeTime
);
consumerSubAll
.
getListner
().
waitForMessageConsume
(
consumerSubAll
.
getList
e
ner
().
waitForMessageConsume
(
MQMessageFactory
.
getMessage
(
msgsWithNoTag
,
tagMessage
.
getAllTagMessageBody
()),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerSubTwoMatchAll
.
getListner
().
getAllMsgBody
()))
consumerSubTwoMatchAll
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getAllTagMessageBody
());
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerSubTwoMachieOne
.
getListner
().
getAllMsgBody
()))
consumerSubTwoMachieOne
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]));
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerSubTag1
.
getListner
().
getAllMsgBody
()))
consumerSubTag1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
tagMessage
.
getMessageBodyByTag
(
tags
[
0
]));
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumerSubAll
.
getListner
().
getAllMsgBody
()))
consumerSubAll
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
MQMessageFactory
.
getMessage
(
msgsWithNoTag
,
tagMessage
.
getAllTagMessageBody
()));
}
...
...
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWithSameGroupConsumerIT.java
浏览文件 @
f619e451
...
...
@@ -61,10 +61,10 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
new
RMQNormalListner
(
originMsgDCName
,
msgBodyDCName
));
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -82,9 +82,9 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
new
RMQNormalListner
(
originMsgDCName
,
msgBodyDCName
));
TestUtils
.
waitForMoment
(
5
);
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -105,9 +105,9 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
mqClients
.
remove
(
1
);
TestUtils
.
waitForMoment
(
5
);
consumer1
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer1
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllMsgBody
()))
consumer1
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/topic/MulConsumerMulTopicIT.java
浏览文件 @
f619e451
...
...
@@ -59,7 +59,7 @@ public class MulConsumerMulTopicIT extends BaseConf {
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
...
...
@@ -80,7 +80,7 @@ public class MulConsumerMulTopicIT extends BaseConf {
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
...
...
@@ -102,7 +102,7 @@ public class MulConsumerMulTopicIT extends BaseConf {
producer
.
send
(
MQMessageFactory
.
getMsg
(
topic2
,
msgSize
,
tag1
));
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/consumer/topic/OneConsumerMulTopicIT.java
浏览文件 @
f619e451
...
...
@@ -55,9 +55,9 @@ public class OneConsumerMulTopicIT extends BaseConf {
producer
.
send
(
MQMessageFactory
.
getMsg
(
topic2
,
msgSize
));
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -74,9 +74,9 @@ public class OneConsumerMulTopicIT extends BaseConf {
producer
.
send
(
MQMessageFactory
.
getMsg
(
topic2
,
msgSize
,
tag
));
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
@@ -96,9 +96,9 @@ public class OneConsumerMulTopicIT extends BaseConf {
producer
.
send
(
MQMessageFactory
.
getMsg
(
topic2
,
msgSize
,
tag1
));
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
*
2
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueIT.java
浏览文件 @
f619e451
...
...
@@ -59,12 +59,12 @@ public class AsyncSendWithMessageQueueIT extends BaseConf {
producer
.
waitForResponse
(
10
*
1000
);
assertThat
(
producer
.
getSuccessMsgCount
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getListner
().
getAllOriginMsg
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getList
e
ner
().
getAllOriginMsg
());
producer
.
clearMsg
();
consumer
.
clearMsg
();
...
...
@@ -74,11 +74,11 @@ public class AsyncSendWithMessageQueueIT extends BaseConf {
producer
.
waitForResponse
(
10
*
1000
);
assertThat
(
producer
.
getSuccessMsgCount
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getListner
().
getAllOriginMsg
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getList
e
ner
().
getAllOriginMsg
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueSelectorIT.java
浏览文件 @
f619e451
...
...
@@ -71,12 +71,12 @@ public class AsyncSendWithMessageQueueSelectorIT extends BaseConf {
producer
.
waitForResponse
(
5
*
1000
);
assertThat
(
producer
.
getSuccessMsgCount
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getListner
().
getAllOriginMsg
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getList
e
ner
().
getAllOriginMsg
());
producer
.
clearMsg
();
consumer
.
clearMsg
();
...
...
@@ -96,11 +96,11 @@ public class AsyncSendWithMessageQueueSelectorIT extends BaseConf {
producer
.
waitForResponse
(
5
*
1000
);
assertThat
(
producer
.
getSuccessMsgCount
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getListner
().
getAllOriginMsg
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getList
e
ner
().
getAllOriginMsg
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithOnlySendCallBackIT.java
浏览文件 @
f619e451
...
...
@@ -55,9 +55,9 @@ public class AsyncSendWithOnlySendCallBackIT extends BaseConf {
producer
.
waitForResponse
(
10
*
1000
);
assertThat
(
producer
.
getSuccessMsgCount
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
...
...
test/src/test/java/org/apache/rocketmq/test/client/producer/exception/msg/MessageUserPropIT.java
浏览文件 @
f619e451
...
...
@@ -63,10 +63,10 @@ public class MessageUserPropIT extends BaseConf {
producer
.
send
(
msg
,
null
);
assertThat
(
producer
.
getAllMsgBody
().
size
()).
isEqualTo
(
1
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
Message
sendMsg
=
(
Message
)
producer
.
getFirstMsg
();
Message
recvMsg
=
(
Message
)
consumer
.
getListner
().
getFirstMsg
();
Message
recvMsg
=
(
Message
)
consumer
.
getList
e
ner
().
getFirstMsg
();
assertThat
(
recvMsg
.
getUserProperty
(
msgKey
)).
isEqualTo
(
sendMsg
.
getUserProperty
(
msgKey
));
}
...
...
@@ -85,10 +85,10 @@ public class MessageUserPropIT extends BaseConf {
producer
.
send
(
msg
,
null
);
assertThat
(
producer
.
getAllMsgBody
().
size
()).
isEqualTo
(
1
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
Message
sendMsg
=
(
Message
)
producer
.
getFirstMsg
();
Message
recvMsg
=
(
Message
)
consumer
.
getListner
().
getFirstMsg
();
Message
recvMsg
=
(
Message
)
consumer
.
getList
e
ner
().
getFirstMsg
();
assertThat
(
recvMsg
.
getUserProperty
(
msgKey
)).
isEqualTo
(
sendMsg
.
getUserProperty
(
msgKey
));
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/oneway/OneWaySendIT.java
浏览文件 @
f619e451
...
...
@@ -56,9 +56,9 @@ public class OneWaySendIT extends BaseConf {
producer
.
waitForResponse
(
5
*
1000
);
assertThat
(
producer
.
getAllMsgBody
().
size
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/oneway/OneWaySendWithMQIT.java
浏览文件 @
f619e451
...
...
@@ -59,9 +59,9 @@ public class OneWaySendWithMQIT extends BaseConf {
producer
.
sendOneWay
(
msgSize
,
mq
);
producer
.
waitForResponse
(
5
*
1000
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
producer
.
clearMsg
();
...
...
@@ -71,9 +71,9 @@ public class OneWaySendWithMQIT extends BaseConf {
producer
.
asyncSend
(
msgSize
,
mq
);
producer
.
waitForResponse
(
5
*
1000
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/oneway/OneWaySendWithSelectorIT.java
浏览文件 @
f619e451
...
...
@@ -71,12 +71,12 @@ public class OneWaySendWithSelectorIT extends BaseConf {
});
assertThat
(
producer
.
getAllMsgBody
().
size
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getListner
().
getAllOriginMsg
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getList
e
ner
().
getAllOriginMsg
());
producer
.
clearMsg
();
consumer
.
clearMsg
();
...
...
@@ -94,11 +94,11 @@ public class OneWaySendWithSelectorIT extends BaseConf {
});
assertThat
(
producer
.
getAllMsgBody
().
size
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getListner
().
getAllOriginMsg
());
VerifyUtils
.
verifyMessageQueueId
(
queueId
,
consumer
.
getList
e
ner
().
getAllOriginMsg
());
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgDynamicRebalanceIT.java
浏览文件 @
f619e451
...
...
@@ -63,20 +63,20 @@ public class OrderMsgDynamicRebalanceIT extends BaseConf {
MessageQueueMsg
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
MQWait
.
waitConsumeAll
(
30
*
1000
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getListner
());
MQWait
.
waitConsumeAll
(
30
*
1000
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
e
ner
());
consumer2
.
shutdown
();
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -94,22 +94,22 @@ public class OrderMsgDynamicRebalanceIT extends BaseConf {
MessageQueueMsg
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getListner
(),
consumer2
.
getList
ner
(),
consumer3
.
getList
ner
());
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
e
ner
(),
consumer2
.
getList
ener
(),
consumer3
.
getListe
ner
());
consumer3
.
shutdown
();
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
30
*
1000
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getListner
(),
consumer3
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListener
(),
consumer3
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer3
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer3
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgIT.java
浏览文件 @
f619e451
...
...
@@ -59,13 +59,13 @@ public class OrderMsgIT extends BaseConf {
MessageQueueMsg
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
mqMsgs
.
getMsgBodys
());
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -77,13 +77,13 @@ public class OrderMsgIT extends BaseConf {
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
mqMsgs
.
getMsgBodys
());
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -96,13 +96,13 @@ public class OrderMsgIT extends BaseConf {
msgSize
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
mqMsgs
.
getMsgBodys
());
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgRebalanceIT.java
浏览文件 @
f619e451
...
...
@@ -64,19 +64,19 @@ public class OrderMsgRebalanceIT extends BaseConf {
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
boolean
balance
=
VerifyUtils
.
verifyBalance
(
producer
.
getAllMsgBody
().
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllUndupMsgBody
()).
size
(),
consumer1
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllUndupMsgBody
()).
size
());
consumer2
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
());
assertThat
(
balance
).
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -97,36 +97,36 @@ public class OrderMsgRebalanceIT extends BaseConf {
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getListner
(),
consumer3
.
getList
ner
(),
consumer4
.
getListner
());
consumer1
.
getList
ener
(),
consumer2
.
getListener
(),
consumer3
.
getListe
ner
(),
consumer4
.
getList
e
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
boolean
balance
=
VerifyUtils
.
verifyBalance
(
producer
.
getAllMsgBody
().
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer1
.
getListner
().
getAllUndupMsgBody
())
consumer1
.
getList
e
ner
().
getAllUndupMsgBody
())
.
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer2
.
getListner
().
getAllUndupMsgBody
()).
size
(),
consumer2
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer3
.
getListner
().
getAllUndupMsgBody
()).
size
(),
consumer3
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
(),
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer4
.
getListner
().
getAllUndupMsgBody
()).
size
());
consumer4
.
getList
e
ner
().
getAllUndupMsgBody
()).
size
());
logger
.
info
(
String
.
format
(
"consumer1:%s;consumer2:%s;consumer3:%s,consumer4:%s"
,
consumer1
.
getListner
().
getAllMsgBody
().
size
(),
consumer2
.
getListner
().
getAllMsgBody
().
size
(),
consumer3
.
getListner
().
getAllMsgBody
().
size
(),
consumer4
.
getListner
().
getAllMsgBody
().
size
()));
consumer1
.
getList
e
ner
().
getAllMsgBody
().
size
(),
consumer2
.
getList
e
ner
().
getAllMsgBody
().
size
(),
consumer3
.
getList
e
ner
().
getAllMsgBody
().
size
(),
consumer4
.
getList
e
ner
().
getAllMsgBody
().
size
()));
assertThat
(
balance
).
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer3
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer3
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer4
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer4
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
test/src/test/java/org/apache/rocketmq/test/client/producer/order/OrderMsgWithTagIT.java
浏览文件 @
f619e451
...
...
@@ -60,13 +60,13 @@ public class OrderMsgWithTagIT extends BaseConf {
MessageQueueMsg
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
,
tag
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
mqMsgs
.
getMsgBodys
());
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -80,13 +80,13 @@ public class OrderMsgWithTagIT extends BaseConf {
MessageQueueMsg
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
,
tag
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
mqMsgs
.
getMsgBodys
());
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -106,13 +106,13 @@ public class OrderMsgWithTagIT extends BaseConf {
mqMsgs
=
new
MessageQueueMsg
(
mqs
,
msgSize
,
tag1
);
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
mqMsgs
.
getMsgBodys
());
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -134,12 +134,12 @@ public class OrderMsgWithTagIT extends BaseConf {
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer1
.
getList
ner
(),
consumer2
.
getList
ner
());
consumer1
.
getList
ener
(),
consumer2
.
getListe
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer1
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer2
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
...
...
@@ -160,10 +160,10 @@ public class OrderMsgWithTagIT extends BaseConf {
producer
.
send
(
mqMsgs
.
getMsgsWithMQ
());
boolean
recvAll
=
MQWait
.
waitConsumeAll
(
consumeTime
,
producer
.
getAllMsgBody
(),
consumer
.
getListner
());
consumer
.
getList
e
ner
());
assertThat
(
recvAll
).
isEqualTo
(
true
);
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getListner
()).
getMsgs
()))
assertThat
(
VerifyUtils
.
verifyOrder
(((
RMQOrderListener
)
consumer
.
getList
e
ner
()).
getMsgs
()))
.
isEqualTo
(
true
);
}
}
test/src/test/java/org/apache/rocketmq/test/client/producer/querymsg/QueryMsgByIdIT.java
浏览文件 @
f619e451
...
...
@@ -57,11 +57,11 @@ public class QueryMsgByIdIT extends BaseConf {
int
msgSize
=
20
;
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
Assert
.
assertEquals
(
"Not all are consumed"
,
0
,
VerifyUtils
.
verify
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()));
consumer
.
getList
e
ner
().
getAllMsgBody
()));
MessageExt
recvMsg
=
(
MessageExt
)
consumer
.
getListner
().
getFirstMsg
();
MessageExt
recvMsg
=
(
MessageExt
)
consumer
.
getList
e
ner
().
getFirstMsg
();
MessageExt
queryMsg
=
null
;
try
{
TestUtils
.
waitForMoment
(
3000
);
...
...
test/src/test/java/org/apache/rocketmq/test/delay/NormalMsgDelayIT.java
浏览文件 @
f619e451
...
...
@@ -57,12 +57,12 @@ public class NormalMsgDelayIT extends DelayConf {
producer
.
send
(
delayMsgs
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
Assert
.
assertEquals
(
"Not all are consumed"
,
0
,
VerifyUtils
.
verify
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()));
consumer
.
getList
e
ner
().
getAllMsgBody
()));
Assert
.
assertEquals
(
"Timer is not correct"
,
true
,
VerifyUtils
.
verifyDelay
(
DELAY_LEVEL
[
delayLevel
-
1
]
*
1000
,
((
RMQDelayListner
)
consumer
.
getListner
()).
getMsgDelayTimes
()));
((
RMQDelayListner
)
consumer
.
getList
e
ner
()).
getMsgDelayTimes
()));
}
@Test
...
...
@@ -72,13 +72,13 @@ public class NormalMsgDelayIT extends DelayConf {
producer
.
send
(
delayMsgs
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
DELAY_LEVEL
[
delayLevel
-
1
]
*
1000
*
2
);
Assert
.
assertEquals
(
"Not all are consumed"
,
0
,
VerifyUtils
.
verify
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()));
consumer
.
getList
e
ner
().
getAllMsgBody
()));
Assert
.
assertEquals
(
"Timer is not correct"
,
true
,
VerifyUtils
.
verifyDelay
(
DELAY_LEVEL
[
delayLevel
-
1
]
*
1000
,
((
RMQDelayListner
)
consumer
.
getListner
()).
getMsgDelayTimes
()));
((
RMQDelayListner
)
consumer
.
getList
e
ner
()).
getMsgDelayTimes
()));
}
@Test
...
...
@@ -88,13 +88,13 @@ public class NormalMsgDelayIT extends DelayConf {
producer
.
send
(
delayMsgs
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
DELAY_LEVEL
[
delayLevel
-
1
]
*
1000
*
2
);
Assert
.
assertEquals
(
"Not all are consumed"
,
0
,
VerifyUtils
.
verify
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()));
consumer
.
getList
e
ner
().
getAllMsgBody
()));
Assert
.
assertEquals
(
"Timer is not correct"
,
true
,
VerifyUtils
.
verifyDelay
(
DELAY_LEVEL
[
delayLevel
-
1
]
*
1000
,
((
RMQDelayListner
)
consumer
.
getListner
()).
getMsgDelayTimes
()));
((
RMQDelayListner
)
consumer
.
getList
e
ner
()).
getMsgDelayTimes
()));
}
@Test
...
...
@@ -104,12 +104,12 @@ public class NormalMsgDelayIT extends DelayConf {
producer
.
send
(
delayMsgs
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
DELAY_LEVEL
[
delayLevel
-
1
]
*
1000
*
2
);
Assert
.
assertEquals
(
"Not all are consumed"
,
0
,
VerifyUtils
.
verify
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()));
consumer
.
getList
e
ner
().
getAllMsgBody
()));
Assert
.
assertEquals
(
"Timer is not correct"
,
true
,
VerifyUtils
.
verifyDelay
(
DELAY_LEVEL
[
delayLevel
-
1
]
*
1000
,
((
RMQDelayListner
)
consumer
.
getListner
()).
getMsgDelayTimes
()));
((
RMQDelayListner
)
consumer
.
getList
e
ner
()).
getMsgDelayTimes
()));
}
}
test/src/test/java/org/apache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java
浏览文件 @
f619e451
...
...
@@ -54,9 +54,9 @@ public class NormalMessageSendAndRecvIT extends BaseConf {
int
msgSize
=
10
;
producer
.
send
(
msgSize
);
Assert
.
assertEquals
(
"Not all sent succeeded"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
consumer
.
getList
e
ner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
assertThat
(
VerifyUtils
.
getFilterdMessage
(
producer
.
getAllMsgBody
(),
consumer
.
getListner
().
getAllMsgBody
()))
consumer
.
getList
e
ner
().
getAllMsgBody
()))
.
containsExactlyElementsIn
(
producer
.
getAllMsgBody
());
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录