X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b65d8b9008c4d0e6b5816d21bf6f1ae81167ee56..b9d2799dfebae724dda3b3e28641116ca5daf5c7:/sdk/go/manifest/manifest.go

diff --git a/sdk/go/manifest/manifest.go b/sdk/go/manifest/manifest.go
index f104d9a103..22b1c974e6 100644
--- a/sdk/go/manifest/manifest.go
+++ b/sdk/go/manifest/manifest.go
@@ -8,7 +8,6 @@ import (
 	"errors"
 	"fmt"
 	"git.curoverse.com/arvados.git/sdk/go/blockdigest"
-	"log"
 	"regexp"
 	"strconv"
 	"strings"
@@ -21,6 +20,7 @@ var LocatorPattern = regexp.MustCompile(
 
 type Manifest struct {
 	Text string
+	Err  error
 }
 
 type BlockLocator struct {
@@ -44,11 +44,19 @@ type FileSegment struct {
 	Len    int
 }
 
+// FileStreamSegment is a portion of a file described as a segment of a stream.
+type FileStreamSegment struct {
+	SegPos uint64
+	SegLen uint64
+	Name   string
+}
+
 // Represents a single line from a manifest.
 type ManifestStream struct {
-	StreamName string
-	Blocks     []string
-	FileTokens []string
+	StreamName         string
+	Blocks             []string
+	FileStreamSegments []FileStreamSegment
+	Err                error
 }
 
 var escapeSeq = regexp.MustCompile(`\\([0-9]{3}|\\)`)
@@ -96,21 +104,21 @@ func ParseBlockLocator(s string) (b BlockLocator, err error) {
 	return
 }
 
-func parseFileToken(tok string) (segPos, segLen uint64, name string, err error) {
+func parseFileStreamSegment(tok string) (ft FileStreamSegment, err error) {
 	parts := strings.SplitN(tok, ":", 3)
 	if len(parts) != 3 {
 		err = ErrInvalidToken
 		return
 	}
-	segPos, err = strconv.ParseUint(parts[0], 10, 64)
+	ft.SegPos, err = strconv.ParseUint(parts[0], 10, 64)
 	if err != nil {
 		return
 	}
-	segLen, err = strconv.ParseUint(parts[1], 10, 64)
+	ft.SegLen, err = strconv.ParseUint(parts[1], 10, 64)
 	if err != nil {
 		return
 	}
-	name = UnescapeName(parts[2])
+	ft.Name = UnescapeName(parts[2])
 	return
 }
 
@@ -127,12 +135,11 @@ func (s *ManifestStream) sendFileSegmentIterByName(filepath string, ch chan<- *F
 	blockLens := make([]int, 0, len(s.Blocks))
 	// This is what streamName+"/"+fileName will look like:
 	target := "./" + filepath
-	for _, fTok := range s.FileTokens {
-		wantPos, wantLen, name, err := parseFileToken(fTok)
-		if err != nil {
-			// Skip (!) invalid file tokens.
-			continue
-		}
+	for _, fTok := range s.FileStreamSegments {
+		wantPos := fTok.SegPos
+		wantLen := fTok.SegLen
+		name := fTok.Name
+
 		if s.StreamName+"/"+name != target {
 			continue
 		}
@@ -183,16 +190,42 @@ func (s *ManifestStream) sendFileSegmentIterByName(filepath string, ch chan<- *F
 
 func parseManifestStream(s string) (m ManifestStream) {
 	tokens := strings.Split(s, " ")
+
 	m.StreamName = UnescapeName(tokens[0])
+	if m.StreamName != "." && !strings.HasPrefix(m.StreamName, "./") {
+		m.Err = fmt.Errorf("Invalid stream name: %s", m.StreamName)
+		return
+	}
+
 	tokens = tokens[1:]
 	var i int
-	for i = range tokens {
+	for i = 0; i < len(tokens); i++ {
 		if !blockdigest.IsBlockLocator(tokens[i]) {
 			break
 		}
 	}
 	m.Blocks = tokens[:i]
-	m.FileTokens = tokens[i:]
+	fileTokens := tokens[i:]
+
+	if len(m.Blocks) == 0 {
+		m.Err = fmt.Errorf("No block locators found")
+		return
+	}
+
+	if len(fileTokens) == 0 {
+		m.Err = fmt.Errorf("No file tokens found")
+		return
+	}
+
+	for _, ft := range fileTokens {
+		pft, err := parseFileStreamSegment(ft)
+		if err != nil {
+			m.Err = fmt.Errorf("Invalid file token: %s", ft)
+			break
+		}
+		m.FileStreamSegments = append(m.FileStreamSegments, pft)
+	}
+
 	return
 }
 
@@ -232,18 +265,24 @@ func (m *Manifest) FileSegmentIterByName(filepath string) <-chan *FileSegment {
 	return ch
 }
 
-// Blocks may appear mulitple times within the same manifest if they
+// Blocks may appear multiple times within the same manifest if they
 // are used by multiple files. In that case this Iterator will output
 // the same block multiple times.
+//
+// In order to detect parse errors, caller must check m.Err after the returned channel closes.
 func (m *Manifest) BlockIterWithDuplicates() <-chan blockdigest.BlockLocator {
 	blockChannel := make(chan blockdigest.BlockLocator)
 	go func(streamChannel <-chan ManifestStream) {
-		for m := range streamChannel {
-			for _, block := range m.Blocks {
+		for ms := range streamChannel {
+			if ms.Err != nil {
+				m.Err = ms.Err
+				continue
+			}
+			for _, block := range ms.Blocks {
 				if b, err := blockdigest.ParseBlockLocator(block); err == nil {
 					blockChannel <- b
 				} else {
-					log.Printf("ERROR: Failed to parse block: %v", err)
+					m.Err = err
 				}
 			}
 		}