stream/mts/meta: added ExtractAll func and added testing utilities

stream/mts/meta/meta.go

@@ -52,6 +52,7 @@ var (
 	errKeyAbsent            = errors.New("Key does not exist in map")
 	errNoHeader             = errors.New("Metadata string does not contain header")
 	errInvalidHeader        = errors.New("Metadata string does not contain valid header")
+	errUnexpectedMetaFormat = errors.New("Unexpected meta format")
 )
 
 // Metadata provides functionality for the storage and encoding of metadata
@@ -168,10 +169,9 @@ func (m *Data) Encode() []byte {
 // key is not present in the metadata string, an error is returned. If the
 // metadata header is not present in the string, an error is returned.
 func Extract(key string, d []byte) (string, error) {
-	if d[0] != 0 {
+	err := checkHeader(d)
-		return "", errNoHeader
+	if err != nil {
-	} else if d[0] == 0 && binary.BigEndian.Uint16(d[2:headSize]) != uint16(len(d[headSize:])) {
+		return "", err
-		return "", errInvalidHeader
 	}
 	d = d[headSize:]
 	entries := strings.Split(string(d), "\t")
@@ -183,3 +183,34 @@ func Extract(key string, d []byte) (string, error) {
 	}
 	return "", errKeyAbsent
 }
+
+// ExtractAll extracts all metadata entries from given data. An Error is returned
+// if the metadata does not have a valid header, or if the meta format is unexpected.
+func ExtractAll(d []byte) ([][2]string, error) {
+	err := checkHeader(d)
+	if err != nil {
+		return nil, err
+	}
+	d = d[headSize:]
+	entries := strings.Split(string(d), "\t")
+	all := make([][2]string, len(entries))
+	for i, entry := range entries {
+		kv := strings.Split(entry, "=")
+		if len(kv) != 2 {
+			return nil, errUnexpectedMetaFormat
+		}
+		copy(all[i][:], kv)
+	}
+	return all, nil
+}
+
+// checkHeader checks that a valid metadata header exists in the given data. An
+// error is returned if the header is absent, or if the header is not valid.
+func checkHeader(d []byte) error {
+	if d[0] != 0 {
+		return errNoHeader
+	} else if d[0] == 0 && binary.BigEndian.Uint16(d[2:headSize]) != uint16(len(d[headSize:])) {
+		return errInvalidHeader
+	}
+	return nil
+}

stream/mts/meta/meta_test.go

@@ -165,3 +165,26 @@ func TestReadFrom(t *testing.T) {
 		}
 	}
 }
+
+// TestExtractAll checks that meta.ExtractAll can correctly extract all metadata
+// from descriptor data.
+func TestExtractAll(t *testing.T) {
+	tstMeta := append([]byte{0x00, 0x10, 0x00, 0x12}, "loc=a,b,c\tts=12345"...)
+	want := [][2]string{
+		{
+			"loc",
+			"a,b,c",
+		},
+		{
+			"ts",
+			"12345",
+		},
+	}
+	got, err := ExtractAll(tstMeta)
+	if err != nil {
+		t.Errorf("Unexpected error: %v\n", err)
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("Did not get expected out. \nGot : %v, \nWant: %v\n", got, want)
+	}
+}