Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fixed MQDecoder::createMessageId and MQDecoder::decodeMessageId. #92

Merged
merged 1 commit into from
Feb 15, 2019
Merged

fixed MQDecoder::createMessageId and MQDecoder::decodeMessageId. #92

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 29 additions & 3 deletions src/common/UtilAll.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,8 +50,33 @@ bool UtilAll::isBlank(const string &str) {
return false;
}

const int hex2int[256] = {
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, -1, -1, -1, -1, -1, -1,
-1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1
};

uint64 UtilAll::hexstr2ull(const char *str) {
return boost::lexical_cast<uint64>(str);
uint64 num = 0;
unsigned char *ch = (unsigned char *) str;
while (*ch != '\0') {
num = (num << 4) + hex2int[*ch];
ch++;
}
return num;
}

int64 UtilAll::str2ll(const char *str) {
Expand All @@ -73,10 +98,11 @@ string UtilAll::bytes2string(const char *bytes, int len) {

return buffer;
#else
char hex_str[] = "0123456789ABCDEF";
static const char hex_str[] = "0123456789ABCDEF";

char result[len * 2 + 1];
result[len * 2] = 0;

result[len * 2] = 0;
for (int i = 0; i < len; i++) {
result[i * 2 + 0] = hex_str[(bytes[i] >> 4) & 0x0F];
result[i * 2 + 1] = hex_str[(bytes[i]) & 0x0F];
Expand Down
93 changes: 39 additions & 54 deletions src/message/MQDecoder.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,9 @@
#include "MemoryOutputStream.h"
#include "MessageSysFlag.h"
#include "UtilAll.h"

namespace rocketmq {

//<!***************************************************************************
const int MQDecoder::MSG_ID_LENGTH = 8 + 8;

Expand All @@ -34,58 +36,50 @@ int MQDecoder::MessageMagicCodePostion = 4;
int MQDecoder::MessageFlagPostion = 16;
int MQDecoder::MessagePhysicOffsetPostion = 28;
int MQDecoder::MessageStoreTimestampPostion = 56;

//<!***************************************************************************
string MQDecoder::createMessageId(sockaddr addr, int64 offset) {
int host, port;
socketAddress2IPPort(addr, host, port);
struct sockaddr_in *sa = (struct sockaddr_in *) &addr;

MemoryOutputStream outputmen(MSG_ID_LENGTH);
outputmen.writeIntBigEndian(host);
outputmen.writeIntBigEndian(port);
outputmen.writeIntBigEndian(sa->sin_addr.s_addr);
outputmen.writeRepeatedByte(0, 2);
outputmen.write(&(sa->sin_port), 2);
outputmen.writeInt64BigEndian(offset);

const char* bytes = static_cast<const char*>(outputmen.getData());
const char *bytes = static_cast<const char *>(outputmen.getData());
int len = outputmen.getDataSize();

return UtilAll::bytes2string(bytes, len);
}

MQMessageId MQDecoder::decodeMessageId(const string& msgId) {

string ipstr = msgId.substr(0, 8);
string portstr = msgId.substr(8, 8);
string offsetstr = msgId.substr(16);
MQMessageId MQDecoder::decodeMessageId(const string &msgId) {

char* end;
int ipint = strtoul(ipstr.c_str(), &end, 16);
int portint = strtoul(portstr.c_str(), &end, 16);
string ipStr = msgId.substr(0, 8);
string portStr = msgId.substr(8, 8);
string offsetStr = msgId.substr(16);

int64 offset = UtilAll::hexstr2ull(offsetstr.c_str());
char *end;
int ipInt = strtoul(ipStr.c_str(), &end, 16);
int portInt = strtoul(portStr.c_str(), &end, 16);

offset = n2hll(offset);

portint = ntohl(portint);
short port = portint;
int64 offset = UtilAll::hexstr2ull(offsetStr.c_str());

struct sockaddr_in sa;
sa.sin_family = AF_INET;
sa.sin_port = htons(port);
sa.sin_addr.s_addr = ipint;

sockaddr addr;
memcpy(&addr, &sa, sizeof(sockaddr));

MQMessageId id(addr, offset);
sa.sin_port = htons(portInt);
sa.sin_addr.s_addr = htonl(ipInt);

MQMessageId id(*((sockaddr*) &sa), offset);
return id;
}

MQMessageExt* MQDecoder::decode(MemoryInputStream& byteBuffer) {
MQMessageExt *MQDecoder::decode(MemoryInputStream &byteBuffer) {
return decode(byteBuffer, true);
}

MQMessageExt* MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody) {
MQMessageExt* msgExt = new MQMessageExt();
MQMessageExt *MQDecoder::decode(MemoryInputStream &byteBuffer, bool readBody) {
MQMessageExt *msgExt = new MQMessageExt();

// 1 TOTALSIZE
int storeSize = byteBuffer.readIntBigEndian();
Expand Down Expand Up @@ -153,13 +147,12 @@ MQMessageExt* MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody) {
MemoryBlock block;
byteBuffer.readIntoMemoryBlock(block, bodyLen);

const char* const pBody = static_cast<const char*>(block.getData());
const char *const pBody = static_cast<const char *>(block.getData());
int len = block.getSize();
string msgbody(pBody, len);

// decompress body
if ((sysFlag & MessageSysFlag::CompressedFlag) ==
MessageSysFlag::CompressedFlag) {
if ((sysFlag & MessageSysFlag::CompressedFlag) == MessageSysFlag::CompressedFlag) {
string outbody;
if (UtilAll::inflate(msgbody, outbody)) {
msgExt->setBody(outbody);
Expand All @@ -173,10 +166,10 @@ MQMessageExt* MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody) {
}

// 16 TOPIC
int topicLen = (int)byteBuffer.readByte();
int topicLen = (int) byteBuffer.readByte();
MemoryBlock block;
byteBuffer.readIntoMemoryBlock(block, topicLen);
const char* const pTopic = static_cast<const char*>(block.getData());
const char *const pTopic = static_cast<const char *>(block.getData());
topicLen = block.getSize();
msgExt->setTopic(pTopic, topicLen);

Expand All @@ -185,7 +178,7 @@ MQMessageExt* MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody) {
if (propertiesLen > 0) {
MemoryBlock block;
byteBuffer.readIntoMemoryBlock(block, propertiesLen);
const char* const pProperty = static_cast<const char*>(block.getData());
const char *const pProperty = static_cast<const char *>(block.getData());
int len = block.getSize();
string propertiesString(pProperty, len);

Expand All @@ -196,29 +189,25 @@ MQMessageExt* MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody) {
}

// 18 msg ID
string offsetMsgId = createMessageId(msgExt->getStoreHost(),
(int64)msgExt->getCommitLogOffset());
string offsetMsgId = createMessageId(msgExt->getStoreHost(), (int64) msgExt->getCommitLogOffset());
msgExt->setOffsetMsgId(offsetMsgId);

string msgId = msgExt->getProperty(MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (msgId.empty())
{
msgId = offsetMsgId;
if (msgId.empty()) {
msgId = offsetMsgId;
}
msgExt->setMsgId(msgId);

// LOG_INFO("get msgExt from remote server, its contents
// are:%s",msgExt->toString().c_str());
// LOG_INFO("get msgExt from remote server, its contents are:%s", msgExt->toString().c_str());
return msgExt;
}

void MQDecoder::decodes(const MemoryBlock* mem, vector<MQMessageExt>& mqvec) {
void MQDecoder::decodes(const MemoryBlock *mem, vector<MQMessageExt> &mqvec) {
mqvec.clear();
decodes(mem, mqvec, true);
}

void MQDecoder::decodes(const MemoryBlock* mem, vector<MQMessageExt>& mqvec,
bool readBody) {
void MQDecoder::decodes(const MemoryBlock *mem, vector<MQMessageExt> &mqvec, bool readBody) {
MemoryInputStream rawInput(*mem, true);

while (rawInput.getNumBytesRemaining() > 0) {
Expand All @@ -227,25 +216,21 @@ void MQDecoder::decodes(const MemoryBlock* mem, vector<MQMessageExt>& mqvec,
}
}

string MQDecoder::messageProperties2String(
const map<string, string>& properties) {
string MQDecoder::messageProperties2String(const map<string, string> &properties) {
string os;
map<string, string>::const_iterator it = properties.begin();

for (; it != properties.end(); ++it) {
// os << it->first << NAME_VALUE_SEPARATOR << it->second <<
// PROPERTY_SEPARATOR;
os.append(it->first);
for (const auto &it : properties) {
// os << it->first << NAME_VALUE_SEPARATOR << it->second << PROPERTY_SEPARATOR;
os.append(it.first);
os += NAME_VALUE_SEPARATOR;
os.append(it->second);
os.append(it.second);
os += PROPERTY_SEPARATOR;
}

return os;
}

void MQDecoder::string2messageProperties(const string& propertiesString,
map<string, string>& properties) {
void MQDecoder::string2messageProperties(const string &propertiesString, map<string, string> &properties) {
vector<string> out;
UtilAll::Split(out, propertiesString, PROPERTY_SEPARATOR);

Expand Down
6 changes: 2 additions & 4 deletions src/message/MQDecoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,12 +32,10 @@ class MQDecoder {

static void decodes(const MemoryBlock* mem, vector<MQMessageExt>& mqvec);

static void decodes(const MemoryBlock* mem, vector<MQMessageExt>& mqvec,
bool readBody);
static void decodes(const MemoryBlock* mem, vector<MQMessageExt>& mqvec, bool readBody);

static string messageProperties2String(const map<string, string>& properties);
static void string2messageProperties(const string& propertiesString,
map<string, string>& properties);
static void string2messageProperties(const string& propertiesString, map<string, string>& properties);

private:
static MQMessageExt* decode(MemoryInputStream& byteBuffer);
Expand Down