/*
 * 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 "string.h"

#include <stdio.h>
#include <vector>
#include <string>
#include "gtest/gtest.h"
#include "gmock/gmock.h"

#include "MemoryOutputStream.h"
#include "MemoryInputStream.h"

#include "MessageSysFlag.h"
#include "MQMessage.h"
#include "MQMessageExt.h"
#include "MQMessageId.h"
#include "MQDecoder.h"
#include "CommandHeader.h"
#include "UtilAll.h"
#include "RemotingCommand.h"

using namespace std;

using ::testing::InitGoogleTest;
using ::testing::InitGoogleMock;
using testing::Return;

using rocketmq::MessageSysFlag;
using rocketmq::MQMessage;
using rocketmq::MQMessageExt;
using rocketmq::MQMessageId;
using rocketmq::MQDecoder;
using rocketmq::SendMessageRequestHeader;
using rocketmq::RemotingCommand;
using rocketmq::UtilAll;
using rocketmq::MemoryInputStream;
using rocketmq::MemoryOutputStream;
using rocketmq::MemoryBlock;

TEST(decoder, messageId) {
	int host;
	int port;
	string msgIdStr = MQDecoder::createMessageId(
			rocketmq::IPPort2socketAddress(inet_addr("127.0.0.1"), 10091),
			(int64) 1024);
	MQMessageId msgId = MQDecoder::decodeMessageId(msgIdStr);

	EXPECT_EQ(msgId.getOffset(), 1024);

	rocketmq::socketAddress2IPPort(msgId.getAddress(), host, port);
	EXPECT_EQ(host, inet_addr("127.0.0.1"));
	EXPECT_EQ(port, 10091);
}

TEST( decoder , decoder ) {

	MQMessageExt mext;
	MemoryOutputStream *memoryOut = new MemoryOutputStream(1024);

	// 1 TOTALSIZE 4
	memoryOut->writeIntBigEndian(107);
	mext.setStoreSize(107);

	// 2 MAGICCODE sizeof(int)  8=4+4
	memoryOut->writeIntBigEndian(14);

	// 3 BODYCRC  12=8+4
	memoryOut->writeIntBigEndian(24);
	mext.setBodyCRC(24);
	// 4 QUEUEID 16=12+4
	memoryOut->writeIntBigEndian(4);
	mext.setQueueId(4);
	// 5 FLAG    20=16+4
	memoryOut->writeIntBigEndian(4);
	mext.setFlag(4);
	// 6 QUEUEOFFSET  28 = 20+8
	memoryOut->writeInt64BigEndian((int64) 1024);
	mext.setQueueOffset(1024);
	// 7 PHYSICALOFFSET 36=28+8
	memoryOut->writeInt64BigEndian((int64) 2048);
	mext.setCommitLogOffset(2048);
	// 8 SYSFLAG  40=36+4
	memoryOut->writeIntBigEndian(0);
	mext.setSysFlag(0);
	// 9 BORNTIMESTAMP 48 = 40+8
	memoryOut->writeInt64BigEndian((int64) 4096);
	mext.setBornTimestamp(4096);
	// 10 BORNHOST 56= 48+8
	memoryOut->writeIntBigEndian(inet_addr("127.0.0.1"));
	memoryOut->writeIntBigEndian(10091);
	mext.setBornHost(rocketmq::IPPort2socketAddress(inet_addr("127.0.0.1"), 10091));
	// 11 STORETIMESTAMP 64 =56+8
	memoryOut->writeInt64BigEndian((int64) 4096);
	mext.setStoreTimestamp(4096);
	// 12 STOREHOST 72 = 64+8
	memoryOut->writeIntBigEndian(inet_addr("127.0.0.2"));
	memoryOut->writeIntBigEndian(10092);
	mext.setStoreHost(rocketmq::IPPort2socketAddress(inet_addr("127.0.0.2"), 10092));
	// 13 RECONSUMETIMES 76 = 72+4
	mext.setReconsumeTimes(111111);
	memoryOut->writeIntBigEndian(mext.getReconsumeTimes());
	// 14 Prepared Transaction Offset 84 = 76+8
	memoryOut->writeInt64BigEndian((int64) 12);
	mext.setPreparedTransactionOffset(12);
	// 15 BODY 88 = 84+4   10
	string *body = new string("1234567890");
	mext.setBody(body->c_str());
	memoryOut->writeIntBigEndian(10);
	memoryOut->write(body->c_str(), body->size());

	// 16 TOPIC
	memoryOut->writeByte(10);
	memoryOut->write(body->c_str(), body->size());
	mext.setTopic(body->c_str());

	//17 PROPERTIES
	memoryOut->writeShortBigEndian(0);

	mext.setMsgId(MQDecoder::createMessageId(mext.getStoreHost(),(int64)mext.getCommitLogOffset()));

	vector<MQMessageExt> mqvec;
	MemoryBlock block = memoryOut->getMemoryBlock();
	MQDecoder::decodes(&block, mqvec);
	EXPECT_EQ(mqvec.size(), 1);
	std::cout<< mext.toString()<<"\n";
	std::cout<< mqvec[0].toString()<<"\n";
	EXPECT_EQ(mqvec[0].toString() , mext.toString());

	mqvec.clear();
	MQDecoder::decodes(&block, mqvec , false);
	EXPECT_FALSE(mqvec[0].getBody().size());

//===============================================================
	// 8 SYSFLAG  40=36+4
	mext.setSysFlag(0 | MessageSysFlag::CompressedFlag);
	memoryOut->setPosition(36);
	memoryOut->writeIntBigEndian( mext.getSysFlag());

	//15 Body 84
	string outBody;
	string boody("123123123");
	UtilAll::deflate(boody, outBody, 5);
	mext.setBody(outBody);

	memoryOut->setPosition(84);
	memoryOut->writeIntBigEndian(outBody.size());
	memoryOut->write(outBody.c_str(), outBody.size());

	// 16 TOPIC
	memoryOut->writeByte(10);
	memoryOut->write(body->c_str(), body->size());
	mext.setTopic(body->c_str());

	//17 PROPERTIES
	map<string, string> properties;
	properties["RocketMQ"] = "cpp-client";
	properties[MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX] = "123456";
	mext.setProperties(properties);
	mext.setMsgId("123456");

	string proString = MQDecoder::messageProperties2String(properties);

	memoryOut->writeShortBigEndian(proString.size());
	memoryOut->write(proString.c_str(), proString.size());

	mext.setStoreSize( memoryOut->getDataSize());
	memoryOut->setPosition(0);
	memoryOut->writeIntBigEndian( mext.getStoreSize() );


	block = memoryOut->getMemoryBlock();
	MQDecoder::decodes(&block, mqvec);
	EXPECT_EQ(mqvec[0].toString() , mext.toString());

}

TEST(decoder , messagePropertiesAndToString){
	map<string, string> properties;
	properties["RocketMQ"] = "cpp-client";
	string proString = MQDecoder::messageProperties2String(properties);

	map<string, string> newProperties;
	MQDecoder::string2messageProperties(proString ,newProperties );
	EXPECT_EQ(properties , newProperties);
}

int main(int argc, char* argv[]) {
	InitGoogleMock(&argc, argv);

	testing::GTEST_FLAG(filter) = "decoder.*";
	int itestts = RUN_ALL_TESTS();
	return itestts;
}