blob: 11490e58f0d4ddcb2f1813faf48305a26fe7f50c [file] [log] [blame]
/*
* 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 <map>
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "MQClientFactory.h"
using namespace std;
using namespace rocketmq;
using rocketmq::MQClientFactory;
using rocketmq::TopicRouteData;
using testing::_;
using ::testing::InitGoogleMock;
using ::testing::InitGoogleTest;
using testing::Return;
class MockMQClientAPIImpl : public MQClientAPIImpl {
public:
MockMQClientAPIImpl(const string& mqClientId,
ClientRemotingProcessor* clientRemotingProcessor,
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
string unitName)
: MQClientAPIImpl(mqClientId,
clientRemotingProcessor,
pullThreadNum,
tcpConnectTimeout,
tcpTransportTryLockTimeout,
unitName) {}
MOCK_METHOD5(getMinOffset, int64(const string&, const string&, int, int, const SessionCredentials&));
MOCK_METHOD3(getTopicRouteInfoFromNameServer, TopicRouteData*(const string&, int, const SessionCredentials&));
};
class MockMQClientFactory : public MQClientFactory {
public:
MockMQClientFactory(const string& mqClientId,
int pullThreadNum,
uint64_t tcpConnectTimeout,
uint64_t tcpTransportTryLockTimeout,
string unitName)
: MQClientFactory(mqClientId, pullThreadNum, tcpConnectTimeout, tcpTransportTryLockTimeout, unitName) {}
void reInitClientImpl(MQClientAPIImpl* pImpl) { m_pClientAPIImpl.reset(pImpl); }
void reInitRemotingProcessor(ClientRemotingProcessor* pImpl) { m_pClientRemotingProcessor.reset(pImpl); }
ClientRemotingProcessor* getRemotingProcessor() { return m_pClientRemotingProcessor.release(); }
};
TEST(MQClientFactoryTest, minOffset) {
string clientId = "testClientId";
int pullThreadNum = 1;
uint64_t tcpConnectTimeout = 3000;
uint64_t tcpTransportTryLockTimeout = 3000;
string unitName = "central";
MockMQClientFactory* factory =
new MockMQClientFactory(clientId, pullThreadNum, tcpConnectTimeout, tcpTransportTryLockTimeout, unitName);
MockMQClientAPIImpl* pImpl = new MockMQClientAPIImpl(clientId, factory->getRemotingProcessor(), pullThreadNum,
tcpConnectTimeout, tcpTransportTryLockTimeout, unitName);
factory->reInitClientImpl(pImpl);
MQMessageQueue mq;
mq.setTopic("testTopic");
mq.setBrokerName("testBroker");
mq.setQueueId(1);
SessionCredentials session_credentials;
TopicRouteData* pData = new TopicRouteData();
pData->setOrderTopicConf("OrderTopicConf");
QueueData qd;
qd.brokerName = "testBroker";
qd.readQueueNums = 8;
qd.writeQueueNums = 8;
qd.perm = 1;
pData->getQueueDatas().push_back(qd);
BrokerData bd;
bd.brokerName = "testBroker";
bd.brokerAddrs[0] = "127.0.0.1:10091";
bd.brokerAddrs[1] = "127.0.0.2:10092";
pData->getBrokerDatas().push_back(bd);
EXPECT_CALL(*pImpl, getMinOffset(_, _, _, _, _)).Times(1).WillOnce(Return(1024));
EXPECT_CALL(*pImpl, getTopicRouteInfoFromNameServer(_, _, _)).Times(1).WillOnce(Return(pData));
int64 offset = factory->minOffset(mq, session_credentials);
EXPECT_EQ(1024, offset);
delete factory;
}
int main(int argc, char* argv[]) {
InitGoogleMock(&argc, argv);
return RUN_ALL_TESTS();
}