test/src/MQClientAPIImpTest.cpp - rocketmq-client-cpp - Git at Google

 /*
  * 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 "CommunicationMode.h"
 #include "MQClientAPIImpl.h"
 #include "MQClientException.h"

 using namespace std;
 using namespace rocketmq;
 using rocketmq::CommunicationMode;
 using rocketmq::RemotingCommand;
 using rocketmq::TcpRemotingClient;
 using testing::_;
 using ::testing::InitGoogleMock;
 using ::testing::InitGoogleTest;
 using testing::Mock;
 using testing::Return;

 class MockTcpRemotingClient : public TcpRemotingClient {
  public:
   MockTcpRemotingClient() : TcpRemotingClient(true, DEFAULT_SSL_PROPERTY_FILE) {}

   MOCK_METHOD3(invokeSync, RemotingCommand*(const string&, RemotingCommand&, int));
   MOCK_METHOD6(invokeAsync, bool(const string&, RemotingCommand&, std::shared_ptr<AsyncCallbackWrap>, int64, int, int));
 };
 class MockMQClientAPIImpl : public MQClientAPIImpl {
  public:
   MockMQClientAPIImpl(const string& mqClientId,
                       ClientRemotingProcessor* clientRemotingProcessor,
                       int pullThreadNum,
                       uint64_t tcpConnectTimeout,
                       uint64_t tcpTransportTryLockTimeout,
                       string unitName)
       : MQClientAPIImpl(mqClientId, true, DEFAULT_SSL_PROPERTY_FILE) {}
   void reInitRemoteClient(TcpRemotingClient* client) { m_pRemotingClient.reset(client); }
 };

 TEST(MQClientAPIImplTest, getMaxOffset) {
   SessionCredentials sc;
   MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
   Mock::AllowLeak(impl);
   MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
   Mock::AllowLeak(pClient);
   impl->reInitRemoteClient(pClient);
   GetMaxOffsetResponseHeader* pHead = new GetMaxOffsetResponseHeader();
   pHead->offset = 4096;
   RemotingCommand* pCommandFailed = new RemotingCommand(SYSTEM_ERROR, nullptr);
   RemotingCommand* pCommandSuccuss = new RemotingCommand(SUCCESS_VALUE, pHead);
   EXPECT_CALL(*pClient, invokeSync(_, _, _))
       .Times(3)
       .WillOnce(Return(nullptr))
       .WillOnce(Return(pCommandFailed))
       .WillOnce(Return(pCommandSuccuss));
   EXPECT_ANY_THROW(impl->getMaxOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
   EXPECT_ANY_THROW(impl->getMaxOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
   int64 offset = impl->getMaxOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc);
   EXPECT_EQ(4096, offset);
 }

 TEST(MQClientAPIImplTest, getMinOffset) {
   SessionCredentials sc;
   MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
   Mock::AllowLeak(impl);
   MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
   Mock::AllowLeak(pClient);
   impl->reInitRemoteClient(pClient);
   GetMinOffsetResponseHeader* pHead = new GetMinOffsetResponseHeader();
   pHead->offset = 2048;
   RemotingCommand* pCommandFailed = new RemotingCommand(SYSTEM_ERROR, nullptr);
   RemotingCommand* pCommandSuccuss = new RemotingCommand(SUCCESS_VALUE, pHead);
   EXPECT_CALL(*pClient, invokeSync(_, _, _))
       .Times(3)
       .WillOnce(Return(nullptr))
       .WillOnce(Return(pCommandFailed))
       .WillOnce(Return(pCommandSuccuss));
   EXPECT_ANY_THROW(impl->getMinOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
   EXPECT_ANY_THROW(impl->getMinOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
   int64 offset = impl->getMinOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc);
   EXPECT_EQ(2048, offset);
 }
 class MyMockAutoDeleteSendCallback : public AutoDeleteSendCallBack {
  public:
   virtual ~MyMockAutoDeleteSendCallback() {}
   virtual void onSuccess(SendResult& sendResult) {
     std::cout << "send Success" << std::endl;
     return;
   }
   virtual void onException(MQException& e) {
     std::cout << "send Exception" << e << std::endl;
     return;
   }
 };

 TEST(MQClientAPIImplTest, sendMessage) {
   string cid = "testClientId";
   SessionCredentials sc;
   MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
   Mock::AllowLeak(impl);
   MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
   Mock::AllowLeak(pClient);
   impl->reInitRemoteClient(pClient);
   SendMessageResponseHeader* pHead = new SendMessageResponseHeader();
   pHead->msgId = "MessageID";
   pHead->queueId = 1;
   pHead->queueOffset = 409600;
   RemotingCommand* pCommandSync = new RemotingCommand(SUCCESS_VALUE, pHead);
   EXPECT_CALL(*pClient, invokeSync(_, _, _)).Times(1).WillOnce(Return(pCommandSync));
   MQMessage message("testTopic", "Hello, RocketMQ");
   string unique_msgId = "UniqMessageID";
   message.setProperty(MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, unique_msgId);
   SendMessageRequestHeader* requestHeader = new SendMessageRequestHeader();
   requestHeader->producerGroup = cid;
   requestHeader->topic = (message.getTopic());
   requestHeader->defaultTopic = DEFAULT_TOPIC;
   requestHeader->defaultTopicQueueNums = 4;
   requestHeader->bornTimestamp = UtilAll::currentTimeMillis();
   SendResult result =
       impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader, 100, 1, ComMode_SYNC, nullptr, sc);
   EXPECT_EQ(result.getSendStatus(), SEND_OK);
   EXPECT_EQ(result.getMsgId(), unique_msgId);
   EXPECT_EQ(result.getQueueOffset(), 409600);
   EXPECT_EQ(result.getOffsetMsgId(), "MessageID");
   EXPECT_EQ(result.getMessageQueue().getBrokerName(), "testBroker");
   EXPECT_EQ(result.getMessageQueue().getTopic(), "testTopic");

   // Try to test Async send

   EXPECT_CALL(*pClient, invokeAsync(_, _, _, _, _, _))
       .Times(7)
       .WillOnce(Return(false))
       .WillOnce(Return(true))
       .WillOnce(Return(false))
       .WillOnce(Return(true))
       .WillOnce(Return(false))
       .WillOnce(Return(false))
       .WillOnce(Return(false));

   SendMessageRequestHeader* requestHeader2 = new SendMessageRequestHeader();
   requestHeader2->producerGroup = cid;
   requestHeader2->topic = (message.getTopic());
   requestHeader2->defaultTopic = DEFAULT_TOPIC;
   requestHeader2->defaultTopicQueueNums = 4;
   requestHeader2->bornTimestamp = UtilAll::currentTimeMillis();
   EXPECT_ANY_THROW(
       impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader2, 100, 1, ComMode_ASYNC, nullptr, sc));

   SendMessageRequestHeader* requestHeader3 = new SendMessageRequestHeader();
   requestHeader3->producerGroup = cid;
   requestHeader3->topic = (message.getTopic());
   requestHeader3->defaultTopic = DEFAULT_TOPIC;
   requestHeader3->defaultTopicQueueNums = 4;
   requestHeader3->bornTimestamp = UtilAll::currentTimeMillis();
   SendCallback* pSendCallback = new MyMockAutoDeleteSendCallback();
   EXPECT_NO_THROW(impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader3, 100, 1, ComMode_ASYNC,
                                     pSendCallback, sc));

   SendMessageRequestHeader* requestHeader4 = new SendMessageRequestHeader();
   requestHeader4->producerGroup = cid;
   requestHeader4->topic = (message.getTopic());
   requestHeader4->defaultTopic = DEFAULT_TOPIC;
   requestHeader4->defaultTopicQueueNums = 4;
   requestHeader4->bornTimestamp = UtilAll::currentTimeMillis();
   SendCallback* pSendCallback2 = new MyMockAutoDeleteSendCallback();
   EXPECT_NO_THROW(impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader4, 1000, 2, ComMode_ASYNC,
                                     pSendCallback2, sc));

   SendMessageRequestHeader* requestHeader5 = new SendMessageRequestHeader();
   requestHeader5->producerGroup = cid;
   requestHeader5->topic = (message.getTopic());
   requestHeader5->defaultTopic = DEFAULT_TOPIC;
   requestHeader5->defaultTopicQueueNums = 4;
   requestHeader5->bornTimestamp = UtilAll::currentTimeMillis();
   SendCallback* pSendCallback3 = new MyMockAutoDeleteSendCallback();
   EXPECT_NO_THROW(impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader5, 1000, 3, ComMode_ASYNC,
                                     pSendCallback3, sc));
 }

 TEST(MQClientAPIImplTest, consumerSendMessageBack) {
   SessionCredentials sc;
   MQMessageExt msg;
   MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
   Mock::AllowLeak(impl);
   MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
   Mock::AllowLeak(pClient);
   impl->reInitRemoteClient(pClient);
   RemotingCommand* pCommandFailed = new RemotingCommand(SYSTEM_ERROR, nullptr);
   RemotingCommand* pCommandSuccuss = new RemotingCommand(SUCCESS_VALUE, nullptr);
   EXPECT_CALL(*pClient, invokeSync(_, _, _))
       .Times(3)
       .WillOnce(Return(nullptr))
       .WillOnce(Return(pCommandFailed))
       .WillOnce(Return(pCommandSuccuss));
   EXPECT_ANY_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
   EXPECT_ANY_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
   EXPECT_NO_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
 }

 int main(int argc, char* argv[]) {
   InitGoogleMock(&argc, argv);
   testing::GTEST_FLAG(filter) = "MQClientAPIImplTest.*";
   return RUN_ALL_TESTS();
 }
	/*
	* 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 "CommunicationMode.h"
	#include "MQClientAPIImpl.h"
	#include "MQClientException.h"

	using namespace std;
	using namespace rocketmq;
	using rocketmq::CommunicationMode;
	using rocketmq::RemotingCommand;
	using rocketmq::TcpRemotingClient;
	using testing::_;
	using ::testing::InitGoogleMock;
	using ::testing::InitGoogleTest;
	using testing::Mock;
	using testing::Return;

	class MockTcpRemotingClient : public TcpRemotingClient {
	public:
	MockTcpRemotingClient() : TcpRemotingClient(true, DEFAULT_SSL_PROPERTY_FILE) {}

	MOCK_METHOD3(invokeSync, RemotingCommand*(const string&, RemotingCommand&, int));
	MOCK_METHOD6(invokeAsync, bool(const string&, RemotingCommand&, std::shared_ptr<AsyncCallbackWrap>, int64, int, int));
	};
	class MockMQClientAPIImpl : public MQClientAPIImpl {
	public:
	MockMQClientAPIImpl(const string& mqClientId,
	ClientRemotingProcessor* clientRemotingProcessor,
	int pullThreadNum,
	uint64_t tcpConnectTimeout,
	uint64_t tcpTransportTryLockTimeout,
	string unitName)
	: MQClientAPIImpl(mqClientId, true, DEFAULT_SSL_PROPERTY_FILE) {}
	void reInitRemoteClient(TcpRemotingClient* client) { m_pRemotingClient.reset(client); }
	};

	TEST(MQClientAPIImplTest, getMaxOffset) {
	SessionCredentials sc;
	MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
	Mock::AllowLeak(impl);
	MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
	Mock::AllowLeak(pClient);
	impl->reInitRemoteClient(pClient);
	GetMaxOffsetResponseHeader* pHead = new GetMaxOffsetResponseHeader();
	pHead->offset = 4096;
	RemotingCommand* pCommandFailed = new RemotingCommand(SYSTEM_ERROR, nullptr);
	RemotingCommand* pCommandSuccuss = new RemotingCommand(SUCCESS_VALUE, pHead);
	EXPECT_CALL(*pClient, invokeSync(_, _, _))
	.Times(3)
	.WillOnce(Return(nullptr))
	.WillOnce(Return(pCommandFailed))
	.WillOnce(Return(pCommandSuccuss));
	EXPECT_ANY_THROW(impl->getMaxOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
	EXPECT_ANY_THROW(impl->getMaxOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
	int64 offset = impl->getMaxOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc);
	EXPECT_EQ(4096, offset);
	}

	TEST(MQClientAPIImplTest, getMinOffset) {
	SessionCredentials sc;
	MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
	Mock::AllowLeak(impl);
	MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
	Mock::AllowLeak(pClient);
	impl->reInitRemoteClient(pClient);
	GetMinOffsetResponseHeader* pHead = new GetMinOffsetResponseHeader();
	pHead->offset = 2048;
	RemotingCommand* pCommandFailed = new RemotingCommand(SYSTEM_ERROR, nullptr);
	RemotingCommand* pCommandSuccuss = new RemotingCommand(SUCCESS_VALUE, pHead);
	EXPECT_CALL(*pClient, invokeSync(_, _, _))
	.Times(3)
	.WillOnce(Return(nullptr))
	.WillOnce(Return(pCommandFailed))
	.WillOnce(Return(pCommandSuccuss));
	EXPECT_ANY_THROW(impl->getMinOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
	EXPECT_ANY_THROW(impl->getMinOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc));
	int64 offset = impl->getMinOffset("127.0.0.0:10911", "testTopic", 0, 1000, sc);
	EXPECT_EQ(2048, offset);
	}
	class MyMockAutoDeleteSendCallback : public AutoDeleteSendCallBack {
	public:
	virtual ~MyMockAutoDeleteSendCallback() {}
	virtual void onSuccess(SendResult& sendResult) {
	std::cout << "send Success" << std::endl;
	return;
	}
	virtual void onException(MQException& e) {
	std::cout << "send Exception" << e << std::endl;
	return;
	}
	};

	TEST(MQClientAPIImplTest, sendMessage) {
	string cid = "testClientId";
	SessionCredentials sc;
	MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
	Mock::AllowLeak(impl);
	MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
	Mock::AllowLeak(pClient);
	impl->reInitRemoteClient(pClient);
	SendMessageResponseHeader* pHead = new SendMessageResponseHeader();
	pHead->msgId = "MessageID";
	pHead->queueId = 1;
	pHead->queueOffset = 409600;
	RemotingCommand* pCommandSync = new RemotingCommand(SUCCESS_VALUE, pHead);
	EXPECT_CALL(*pClient, invokeSync(_, _, _)).Times(1).WillOnce(Return(pCommandSync));
	MQMessage message("testTopic", "Hello, RocketMQ");
	string unique_msgId = "UniqMessageID";
	message.setProperty(MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, unique_msgId);
	SendMessageRequestHeader* requestHeader = new SendMessageRequestHeader();
	requestHeader->producerGroup = cid;
	requestHeader->topic = (message.getTopic());
	requestHeader->defaultTopic = DEFAULT_TOPIC;
	requestHeader->defaultTopicQueueNums = 4;
	requestHeader->bornTimestamp = UtilAll::currentTimeMillis();
	SendResult result =
	impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader, 100, 1, ComMode_SYNC, nullptr, sc);
	EXPECT_EQ(result.getSendStatus(), SEND_OK);
	EXPECT_EQ(result.getMsgId(), unique_msgId);
	EXPECT_EQ(result.getQueueOffset(), 409600);
	EXPECT_EQ(result.getOffsetMsgId(), "MessageID");
	EXPECT_EQ(result.getMessageQueue().getBrokerName(), "testBroker");
	EXPECT_EQ(result.getMessageQueue().getTopic(), "testTopic");

	// Try to test Async send

	EXPECT_CALL(*pClient, invokeAsync(_, _, _, _, _, _))
	.Times(7)
	.WillOnce(Return(false))
	.WillOnce(Return(true))
	.WillOnce(Return(false))
	.WillOnce(Return(true))
	.WillOnce(Return(false))
	.WillOnce(Return(false))
	.WillOnce(Return(false));

	SendMessageRequestHeader* requestHeader2 = new SendMessageRequestHeader();
	requestHeader2->producerGroup = cid;
	requestHeader2->topic = (message.getTopic());
	requestHeader2->defaultTopic = DEFAULT_TOPIC;
	requestHeader2->defaultTopicQueueNums = 4;
	requestHeader2->bornTimestamp = UtilAll::currentTimeMillis();
	EXPECT_ANY_THROW(
	impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader2, 100, 1, ComMode_ASYNC, nullptr, sc));

	SendMessageRequestHeader* requestHeader3 = new SendMessageRequestHeader();
	requestHeader3->producerGroup = cid;
	requestHeader3->topic = (message.getTopic());
	requestHeader3->defaultTopic = DEFAULT_TOPIC;
	requestHeader3->defaultTopicQueueNums = 4;
	requestHeader3->bornTimestamp = UtilAll::currentTimeMillis();
	SendCallback* pSendCallback = new MyMockAutoDeleteSendCallback();
	EXPECT_NO_THROW(impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader3, 100, 1, ComMode_ASYNC,
	pSendCallback, sc));

	SendMessageRequestHeader* requestHeader4 = new SendMessageRequestHeader();
	requestHeader4->producerGroup = cid;
	requestHeader4->topic = (message.getTopic());
	requestHeader4->defaultTopic = DEFAULT_TOPIC;
	requestHeader4->defaultTopicQueueNums = 4;
	requestHeader4->bornTimestamp = UtilAll::currentTimeMillis();
	SendCallback* pSendCallback2 = new MyMockAutoDeleteSendCallback();
	EXPECT_NO_THROW(impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader4, 1000, 2, ComMode_ASYNC,
	pSendCallback2, sc));

	SendMessageRequestHeader* requestHeader5 = new SendMessageRequestHeader();
	requestHeader5->producerGroup = cid;
	requestHeader5->topic = (message.getTopic());
	requestHeader5->defaultTopic = DEFAULT_TOPIC;
	requestHeader5->defaultTopicQueueNums = 4;
	requestHeader5->bornTimestamp = UtilAll::currentTimeMillis();
	SendCallback* pSendCallback3 = new MyMockAutoDeleteSendCallback();
	EXPECT_NO_THROW(impl->sendMessage("127.0.0.0:10911", "testBroker", message, requestHeader5, 1000, 3, ComMode_ASYNC,
	pSendCallback3, sc));
	}

	TEST(MQClientAPIImplTest, consumerSendMessageBack) {
	SessionCredentials sc;
	MQMessageExt msg;
	MockMQClientAPIImpl* impl = new MockMQClientAPIImpl("testMockAPIImpl", nullptr, 1, 2, 3, "testUnit");
	Mock::AllowLeak(impl);
	MockTcpRemotingClient* pClient = new MockTcpRemotingClient();
	Mock::AllowLeak(pClient);
	impl->reInitRemoteClient(pClient);
	RemotingCommand* pCommandFailed = new RemotingCommand(SYSTEM_ERROR, nullptr);
	RemotingCommand* pCommandSuccuss = new RemotingCommand(SUCCESS_VALUE, nullptr);
	EXPECT_CALL(*pClient, invokeSync(_, _, _))
	.Times(3)
	.WillOnce(Return(nullptr))
	.WillOnce(Return(pCommandFailed))
	.WillOnce(Return(pCommandSuccuss));
	EXPECT_ANY_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
	EXPECT_ANY_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
	EXPECT_NO_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
	}

	int main(int argc, char* argv[]) {
	InitGoogleMock(&argc, argv);
	testing::GTEST_FLAG(filter) = "MQClientAPIImplTest.*";
	return RUN_ALL_TESTS();
	}