[ISSUE #211]fix unmarhsal messageID bug (#210)

[ISSUE #211]fix unmarhsal messageID bug

internal/utils/helper.go

@@ -21,10 +21,11 @@ import (
 	"bytes"
 	"compress/zlib"
 	"io/ioutil"
+	"net"
 )
 
 func GetAddressByBytes(data []byte) string {
-	return "127.0.0.1"
+	return net.IPv4(data[0], data[1], data[2], data[3]).String()
 }
 
 func UnCompress(data []byte) []byte {

primitive/message.go

@@ -28,8 +28,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pkg/errors"
-
 	"github.com/apache/rocketmq-client-go/internal/utils"
 )
 
@@ -394,38 +392,23 @@ func createMessageId(addr []byte, port int32, offset int64) string {
 	return strings.ToUpper(hex.EncodeToString(buffer.Bytes()))
 }
 
-func UnmarshalMsgID(msgID []byte) (*MessageID, error) {
-	if len(msgID) < 32 {
-		return nil, errors.Errorf("%s len < 32", string(msgID))
-	}
-	ip := make([]byte, 8)
-	port := make([]byte, 8)
-	offset := make([]byte, 16)
-	var portVal int
-	var offsetVal int64
-
-	_, err := hex.Decode(ip, msgID[0:8])
-	if err != nil {
-		_, err = hex.Decode(port, msgID[8:16])
-	}
-	if err != nil {
-		_, err = hex.Decode(offset, msgID[16:32])
-	}
-	if err != nil {
-		portVal, err = strconv.Atoi(string(port))
-	}
-	if err != nil {
-		offsetVal, err = strconv.ParseInt(string(offset), 10, 0)
-	}
-
-	if err != nil {
-		return nil, err
+func UnmarshalMsgID(id []byte) (*MessageID, error) {
+	if len(id) < 32 {
+		return nil, fmt.Errorf("%s len < 32", string(id))
 	}
+	var (
+		ipBytes     = make([]byte, 4)
+		portBytes   = make([]byte, 4)
+		offsetBytes = make([]byte, 8)
+	)
+	hex.Decode(ipBytes, id[0:8])
+	hex.Decode(portBytes, id[8:16])
+	hex.Decode(offsetBytes, id[16:32])
 
 	return &MessageID{
-		Addr:   string(ip),
-		Port:   portVal,
-		Offset: offsetVal,
+		Addr:   utils.GetAddressByBytes(ipBytes),
+		Port:   int(binary.BigEndian.Uint32(portBytes)),
+		Offset: int64(binary.BigEndian.Uint64(offsetBytes)),
 	}, nil
 }
 

primitive/message_test.go

@@ -0,0 +1,38 @@
+/*
+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.
+*/
+
+package primitive
+
+import "testing"
+
+func TestMessageID(t *testing.T) {
+	id := []byte("0AAF0895000078BF000000000009BB4A")
+	msgID, err := UnmarshalMsgID(id)
+	if err != nil {
+		t.Fatalf("unmarshal msg id error, ms is: %s", err.Error())
+	}
+	if msgID.Addr != "10.175.8.149" {
+		t.Fatalf("parse messageID %s error", id)
+	}
+	if msgID.Port != 30911 {
+		t.Fatalf("parse messageID %s error", id)
+	}
+	if msgID.Offset != 637770 {
+		t.Fatalf("parse messageID %s error", id)
+	}
+	t.Log(msgID)
+}