| /** |
| * 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 |