| /* |
| * 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. |
| */ |
| #include <gmock/gmock.h> |
| #include <gtest/gtest.h> |
| |
| #include <string> |
| #include <vector> |
| |
| #include "DefaultLitePullConsumer.h" |
| #include "MQClientInstance.h" |
| #include "MessageExtImpl.h" |
| #include "MQMessageQueue.h" |
| #include "PullResult.h" |
| #include "SessionCredentials.h" |
| #include "c/CCommon.h" |
| #include "c/CPullConsumer.h" |
| |
| using testing::_; |
| using testing::Expectation; |
| using testing::InitGoogleMock; |
| using testing::InitGoogleTest; |
| using testing::Mock; |
| using testing::Return; |
| using testing::SetArgReferee; |
| |
| using rocketmq::DefaultMQPullConsumer; |
| using rocketmq::MessageModel; |
| using rocketmq::MessageExtPtr; |
| using rocketmq::MessageClientExtImpl; |
| using rocketmq::MQMessageQueue; |
| using rocketmq::PullResult; |
| using rocketmq::PullStatus; |
| using rocketmq::SessionCredentials; |
| |
| class MockDefaultMQPullConsumer : public DefaultMQPullConsumer { |
| public: |
| MockDefaultMQPullConsumer(const std::string& groupname) : DefaultMQPullConsumer(groupname) {} |
| |
| MOCK_METHOD(void, start, (), (override)); |
| MOCK_METHOD(void, shutdown, (), (override)); |
| MOCK_METHOD(void, fetchSubscribeMessageQueues, (const std::string&, std::vector<MQMessageQueue>&), (override)); |
| MOCK_METHOD(PullResult, pull, (const MQMessageQueue&, const std::string&, int64_t, int), (override)); |
| }; |
| |
| TEST(CPullConsumerTest, Pull) { |
| MockDefaultMQPullConsumer* mqPullConsumer = new MockDefaultMQPullConsumer("groudId"); |
| CPullConsumer* pullConsumer = reinterpret_cast<CPullConsumer*>(static_cast<DefaultMQPullConsumer*>(mqPullConsumer)); |
| |
| CMessageQueue* cMessageQueue = (CMessageQueue*)malloc(sizeof(CMessageQueue)); |
| strncpy(cMessageQueue->topic, "testTopic", 8); |
| strncpy(cMessageQueue->brokerName, "testBroker", 9); |
| cMessageQueue->queueId = 1; |
| |
| PullResult timeOutPullResult(PullStatus::BROKER_TIMEOUT, 1, 2, 3); |
| PullResult noNewMsgPullResult(PullStatus::NO_NEW_MSG, 1, 2, 3); |
| PullResult noMatchedMsgPullResult(PullStatus::NO_MATCHED_MSG, 1, 2, 3); |
| PullResult offsetIllegalPullResult(PullStatus::OFFSET_ILLEGAL, 1, 2, 3); |
| PullResult defaultPullResult((PullStatus)-1, 1, 2, 3); |
| |
| std::vector<MessageExtPtr> src; |
| for (int i = 0; i < 5; i++) { |
| auto ext = std::make_shared<MessageClientExtImpl>(); |
| src.push_back(ext); |
| } |
| PullResult foundPullResult(PullStatus::FOUND, 1, 2, 3, src); |
| |
| EXPECT_CALL(*mqPullConsumer, pull(_, _, _, _)) |
| .WillOnce(Return(timeOutPullResult)) |
| .WillOnce(Return(noNewMsgPullResult)) |
| .WillOnce(Return(noMatchedMsgPullResult)) |
| .WillOnce(Return(offsetIllegalPullResult)) |
| .WillOnce(Return(defaultPullResult)) |
| /*.WillOnce(Return(timeOutPullResult))*/.WillOnce(Return(foundPullResult)); |
| |
| CPullResult timeOutcPullResult = Pull(pullConsumer, cMessageQueue, "123123", 0, 0); |
| EXPECT_EQ(timeOutcPullResult.pullStatus, E_BROKER_TIMEOUT); |
| |
| CPullResult noNewMsgcPullResult = Pull(pullConsumer, cMessageQueue, "123123", 0, 0); |
| EXPECT_EQ(noNewMsgcPullResult.pullStatus, E_NO_NEW_MSG); |
| |
| CPullResult noMatchedMsgcPullResult = Pull(pullConsumer, cMessageQueue, "123123", 0, 0); |
| EXPECT_EQ(noMatchedMsgcPullResult.pullStatus, E_NO_MATCHED_MSG); |
| |
| CPullResult offsetIllegalcPullResult = Pull(pullConsumer, cMessageQueue, "123123", 0, 0); |
| EXPECT_EQ(offsetIllegalcPullResult.pullStatus, E_OFFSET_ILLEGAL); |
| |
| CPullResult defaultcPullResult = Pull(pullConsumer, cMessageQueue, "123123", 0, 0); |
| EXPECT_EQ(defaultcPullResult.pullStatus, E_NO_NEW_MSG); |
| |
| CPullResult exceptionPullResult = Pull(pullConsumer, cMessageQueue, NULL, 0, 0); |
| EXPECT_EQ(exceptionPullResult.pullStatus, E_BROKER_TIMEOUT); |
| |
| CPullResult foundcPullResult = Pull(pullConsumer, cMessageQueue, "123123", 0, 0); |
| EXPECT_EQ(foundcPullResult.pullStatus, E_FOUND); |
| |
| delete mqPullConsumer; |
| } |
| |
| TEST(CPullConsumerTest, InfoMock) { |
| MockDefaultMQPullConsumer* mqPullConsumer = new MockDefaultMQPullConsumer("groudId"); |
| CPullConsumer* pullConsumer = (CPullConsumer*)mqPullConsumer; |
| |
| Expectation exp = EXPECT_CALL(*mqPullConsumer, start()).Times(1); |
| EXPECT_EQ(StartPullConsumer(pullConsumer), OK); |
| |
| EXPECT_CALL(*mqPullConsumer, shutdown()).Times(1); |
| EXPECT_EQ(ShutdownPullConsumer(pullConsumer), OK); |
| |
| // EXPECT_CALL(*mqPullConsumer,setLogFileSizeAndNum(_,_)).Times(1); |
| EXPECT_EQ(SetPullConsumerLogFileNumAndSize(pullConsumer, 1, 2), OK); |
| |
| // EXPECT_CALL(*mqPullConsumer,set_log_level(_)).Times(1); |
| EXPECT_EQ(SetPullConsumerLogLevel(pullConsumer, E_LOG_LEVEL_INFO), OK); |
| |
| std::vector<MQMessageQueue> fullMQ; |
| for (int i = 0; i < 5; i++) { |
| MQMessageQueue queue("testTopic", "testsBroker", i); |
| fullMQ.push_back(queue); |
| } |
| |
| EXPECT_CALL(*mqPullConsumer, fetchSubscribeMessageQueues(_, _)).Times(1).WillOnce(SetArgReferee<1>(fullMQ)); |
| CMessageQueue* mqs = NULL; |
| int size = 0; |
| FetchSubscriptionMessageQueues(pullConsumer, "testTopic", &mqs, &size); |
| EXPECT_EQ(size, 5); |
| |
| delete mqPullConsumer; |
| } |
| |
| TEST(CPullConsumerTest, Init) { |
| CPullConsumer* pullConsumer = CreatePullConsumer("testGroupId"); |
| DefaultMQPullConsumer* defaultMQPullConsumer = (DefaultMQPullConsumer*)pullConsumer; |
| EXPECT_FALSE(pullConsumer == NULL); |
| |
| EXPECT_EQ(SetPullConsumerGroupID(pullConsumer, "groupId"), OK); |
| EXPECT_EQ(GetPullConsumerGroupID(pullConsumer), defaultMQPullConsumer->group_name().c_str()); |
| |
| EXPECT_EQ(SetPullConsumerNameServerAddress(pullConsumer, "127.0.0.1:10091"), OK); |
| EXPECT_EQ(defaultMQPullConsumer->namesrv_addr(), "127.0.0.1:10091"); |
| |
| EXPECT_EQ(SetPullConsumerSessionCredentials(pullConsumer, "accessKey", "secretKey", "channel"), OK); |
| // SessionCredentials sessionCredentials = defaultMQPullConsumer->getSessionCredentials(); |
| // EXPECT_EQ(sessionCredentials.getAccessKey(), "accessKey"); |
| |
| EXPECT_EQ(SetPullConsumerLogPath(pullConsumer, NULL), OK); |
| |
| // EXPECT_EQ(SetPullConsumerLogFileNumAndSize(pullConsumer,NULL,NULL),NULL_POINTER); |
| EXPECT_EQ(SetPullConsumerLogLevel(pullConsumer, E_LOG_LEVEL_DEBUG), OK); |
| } |
| |
| TEST(CPullConsumerTest, CheckNull) { |
| CPullConsumer* pullConsumer = CreatePullConsumer("testGroupId"); |
| DefaultMQPullConsumer* defaultMQPullConsumer = (DefaultMQPullConsumer*)pullConsumer; |
| EXPECT_FALSE(pullConsumer == NULL); |
| |
| EXPECT_EQ(SetPullConsumerGroupID(pullConsumer, "groupId"), OK); |
| EXPECT_EQ(GetPullConsumerGroupID(pullConsumer), defaultMQPullConsumer->group_name().c_str()); |
| |
| EXPECT_EQ(SetPullConsumerNameServerAddress(pullConsumer, "127.0.0.1:10091"), OK); |
| EXPECT_EQ(defaultMQPullConsumer->namesrv_addr(), "127.0.0.1:10091"); |
| |
| EXPECT_EQ(SetPullConsumerSessionCredentials(pullConsumer, "accessKey", "secretKey", "channel"), OK); |
| // SessionCredentials sessionCredentials = defaultMQPullConsumer->getSessionCredentials(); |
| // EXPECT_EQ(sessionCredentials.getAccessKey(), "accessKey"); |
| |
| EXPECT_EQ(SetPullConsumerLogPath(pullConsumer, NULL), OK); |
| // EXPECT_EQ(SetPullConsumerLogFileNumAndSize(pullConsumer,NULL,NULL),NULL_POINTER); |
| EXPECT_EQ(SetPullConsumerLogLevel(pullConsumer, E_LOG_LEVEL_DEBUG), OK); |
| EXPECT_EQ(DestroyPullConsumer(pullConsumer), OK); |
| EXPECT_EQ(StartPullConsumer(NULL), NULL_POINTER); |
| EXPECT_EQ(ShutdownPullConsumer(NULL), NULL_POINTER); |
| } |
| |
| int main(int argc, char* argv[]) { |
| InitGoogleMock(&argc, argv); |
| testing::GTEST_FLAG(throw_on_failure) = true; |
| testing::GTEST_FLAG(filter) = "CPullConsumerTest.Skipped"; |
| return RUN_ALL_TESTS(); |
| } |