1 /* Deals with parsing Manifest Text. */
3 // Inspired by the Manifest class in arvados/sdk/ruby/lib/arvados/keep.rb
10 "git.curoverse.com/arvados.git/sdk/go/blockdigest"
16 var ErrInvalidToken = errors.New("Invalid token")
18 var LocatorPattern = regexp.MustCompile(
19 "^[0-9a-fA-F]{32}\\+[0-9]+(\\+[A-Z][A-Za-z0-9@_-]+)*$")
21 type Manifest struct {
26 type BlockLocator struct {
27 Digest blockdigest.BlockDigest
32 type DataSegment struct {
38 // FileSegment is a portion of a file that is contained within a
40 type FileSegment struct {
42 // Offset (within this block) of this data segment
47 // Represents a single line from a manifest.
48 type ManifestStream struct {
55 var escapeSeq = regexp.MustCompile(`\\([0-9]{3}|\\)`)
57 func unescapeSeq(seq string) string {
61 i, err := strconv.ParseUint(seq[1:], 8, 8)
63 // Invalid escape sequence: can't unescape.
66 return string([]byte{byte(i)})
69 func UnescapeName(s string) string {
70 return escapeSeq.ReplaceAllStringFunc(s, unescapeSeq)
73 func ParseBlockLocator(s string) (b BlockLocator, err error) {
74 if !LocatorPattern.MatchString(s) {
75 err = fmt.Errorf("String \"%s\" does not match BlockLocator pattern "+
78 LocatorPattern.String())
80 tokens := strings.Split(s, "+")
82 var blockDigest blockdigest.BlockDigest
83 // We expect both of the following to succeed since LocatorPattern
84 // restricts the strings appropriately.
85 blockDigest, err = blockdigest.FromString(tokens[0])
89 blockSize, err = strconv.ParseInt(tokens[1], 10, 0)
93 b.Digest = blockDigest
94 b.Size = int(blockSize)
100 func parseFileToken(tok string) (segPos, segLen uint64, name string, err error) {
101 parts := strings.SplitN(tok, ":", 3)
103 err = ErrInvalidToken
106 segPos, err = strconv.ParseUint(parts[0], 10, 64)
110 segLen, err = strconv.ParseUint(parts[1], 10, 64)
114 name = UnescapeName(parts[2])
118 func (s *ManifestStream) FileSegmentIterByName(filepath string) <-chan *FileSegment {
119 ch := make(chan *FileSegment)
121 s.sendFileSegmentIterByName(filepath, ch)
127 func (s *ManifestStream) sendFileSegmentIterByName(filepath string, ch chan<- *FileSegment) {
128 blockLens := make([]int, 0, len(s.Blocks))
129 // This is what streamName+"/"+fileName will look like:
130 target := "./" + filepath
131 for _, fTok := range s.FileTokens {
132 wantPos, wantLen, name, err := parseFileToken(fTok)
134 // Skip (!) invalid file tokens.
137 if s.StreamName+"/"+name != target {
141 ch <- &FileSegment{Locator: "d41d8cd98f00b204e9800998ecf8427e+0", Offset: 0, Len: 0}
144 // Linear search for blocks containing data for this
146 var blockPos uint64 = 0 // position of block in stream
147 for i, loc := range s.Blocks {
148 if blockPos >= wantPos+wantLen {
151 if len(blockLens) <= i {
152 blockLens = blockLens[:i+1]
153 b, err := ParseBlockLocator(loc)
155 // Unparseable locator -> unusable
160 blockLens[i] = b.Size
162 blockLen := uint64(blockLens[i])
163 if blockPos+blockLen <= wantPos {
172 if blockPos < wantPos {
173 fseg.Offset = int(wantPos - blockPos)
174 fseg.Len -= fseg.Offset
176 if blockPos+blockLen > wantPos+wantLen {
177 fseg.Len = int(wantPos+wantLen-blockPos) - fseg.Offset
185 func parseManifestStream(s string) (m ManifestStream) {
186 tokens := strings.Split(s, " ")
188 m.StreamName = UnescapeName(tokens[0])
189 if m.StreamName != "." && !strings.HasPrefix(m.StreamName, "./") {
190 m.Err = fmt.Errorf("Invalid stream name: %s", m.StreamName)
196 for i = 0; i < len(tokens); i++ {
197 if !blockdigest.IsBlockLocator(tokens[i]) {
201 m.Blocks = tokens[:i]
202 m.FileTokens = tokens[i:]
204 if len(m.Blocks) == 0 {
205 m.Err = fmt.Errorf("No block locators found")
209 if len(m.FileTokens) == 0 {
210 m.Err = fmt.Errorf("No file tokens found")
214 for _, ft := range m.FileTokens {
215 _, _, _, err := parseFileToken(ft)
217 m.Err = fmt.Errorf("Invalid file token: %s", ft)
225 func (m *Manifest) StreamIter() <-chan ManifestStream {
226 ch := make(chan ManifestStream)
227 go func(input string) {
228 // This slice holds the current line and the remainder of the
229 // manifest. We parse one line at a time, to save effort if we
230 // only need the first few lines.
231 lines := []string{"", input}
233 lines = strings.SplitN(lines[1], "\n", 2)
234 if len(lines[0]) > 0 {
235 // Only parse non-blank lines
236 ch <- parseManifestStream(lines[0])
247 func (m *Manifest) FileSegmentIterByName(filepath string) <-chan *FileSegment {
248 ch := make(chan *FileSegment)
250 for stream := range m.StreamIter() {
251 if !strings.HasPrefix("./"+filepath, stream.StreamName+"/") {
254 stream.sendFileSegmentIterByName(filepath, ch)
261 // Blocks may appear mulitple times within the same manifest if they
262 // are used by multiple files. In that case this Iterator will output
263 // the same block multiple times.
265 // In order to detect parse errors, caller must check m.Err after the returned channel closes.
266 func (m *Manifest) BlockIterWithDuplicates() <-chan blockdigest.BlockLocator {
267 blockChannel := make(chan blockdigest.BlockLocator)
268 go func(streamChannel <-chan ManifestStream) {
269 for ms := range streamChannel {
274 for _, block := range ms.Blocks {
275 if b, err := blockdigest.ParseBlockLocator(block); err == nil {