blob: 56ee4e4a4f567f0b5834c47a46cb519c7ed4886d [file]
/**
* Copyright (C) 2013 kangliqiang ,kangliq@163.com
*
* Licensed 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.
*/
#ifndef __OFFSETSERIALIZEWRAPPER_H__
#define __OFFSETSERIALIZEWRAPPER_H__
#include <map>
#include <string>
#include "RemotingSerializable.h"
#include "MessageQueue.h"
#include "AtomicValue.h"
#include "UtilAll.h"
#include "json/json.h"
namespace rmq
{
class OffsetSerializeWrapper : public RemotingSerializable
{
public:
void encode(std::string& outData)
{
Json::Value offsetTable;
RMQ_FOR_EACH(m_offsetTable, it)
{
MessageQueue mq = it->first;
kpr::AtomicLong& offset = it->second;
std::string mqStr = mq.toJsonString();
offsetTable[mqStr] = offset.get();
}
Json::Value obj;
obj["offsetTable"] = offsetTable;
Json::FastWriter writer;
outData = writer.write(obj);
}
static OffsetSerializeWrapper* decode(const char* pData, int len)
{
/*
{
"offsetTable":{
'{"brokerName":"broker-a","queueId":3,"topic":"TopicTest"}':0,
'{"brokerName":"broker-a","queueId":2,"topic":"TopicTest"}':0
}
}
*/
RMQ_DEBUG("decode, data:%s", pData);
Json::Reader reader;
Json::Value obj;
if (!reader.parse(pData, pData + len, obj))
{
return NULL;
}
RMQ_DEBUG("decode ok");
if (obj.isObject())
{
Json::Value objOffsetTable = obj["offsetTable"];
if (objOffsetTable.isObject())
{
std::map<MessageQueue, kpr::AtomicLong> offsetTable;
OffsetSerializeWrapper* offsetWrapper = new OffsetSerializeWrapper();
Json::Value::Members members = objOffsetTable.getMemberNames();
for (typeof(members.begin()) it = members.begin(); it != members.end(); it++)
{
std::string key = *it;
Json::Value objMq;
RMQ_DEBUG("decode, key:%s", key.c_str());
if (!reader.parse(key, objMq))
{
continue;
}
RMQ_DEBUG("decode, key ok");
MessageQueue mq(objMq["topic"].asString(), objMq["brokerName"].asString(),
objMq["queueId"].asInt());
long long offset = objOffsetTable[key].asInt64();
offsetTable[mq] = kpr::AtomicLong(offset);
}
offsetWrapper->setOffsetTable(offsetTable);
return offsetWrapper;
}
}
return NULL;
}
std::string toString() const
{
std::stringstream ss;
ss << "{offsetTable=" << UtilAll::toString(m_offsetTable)
<< "}";
return ss.str();
}
std::map<MessageQueue, kpr::AtomicLong>& getOffsetTable()
{
return m_offsetTable;
}
void setOffsetTable(const std::map<MessageQueue, kpr::AtomicLong>& table)
{
m_offsetTable = table;
}
private:
std::map<MessageQueue, kpr::AtomicLong> m_offsetTable;
};
typedef kpr::RefHandleT<OffsetSerializeWrapper> OffsetSerializeWrapperPtr;
}
#endif