1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
12 "git.curoverse.com/arvados.git/sdk/go/keepclient"
13 "git.curoverse.com/arvados.git/sdk/go/manifest"
27 type ManifestStreamWriter struct {
29 *manifest.ManifestStream
36 type IKeepClient interface {
37 PutHB(hash string, buf []byte) (string, int, error)
40 func (m *ManifestStreamWriter) Write(p []byte) (int, error) {
41 n, err := m.ReadFrom(bytes.NewReader(p))
45 func (m *ManifestStreamWriter) ReadFrom(r io.Reader) (n int64, err error) {
51 m.Block = &Block{make([]byte, keepclient.BLOCKSIZE), 0}
53 count, err = r.Read(m.Block.data[m.Block.offset:])
55 m.Block.offset += int64(count)
56 if m.Block.offset == keepclient.BLOCKSIZE {
70 func (m *ManifestStreamWriter) goUpload() {
72 uploader := m.uploader
74 for block := range uploader {
75 hash := fmt.Sprintf("%x", md5.Sum(block.data[0:block.offset]))
76 signedHash, _, err := m.ManifestWriter.IKeepClient.PutHB(hash, block.data[0:block.offset])
78 errors = append(errors, err)
80 m.ManifestStream.Blocks = append(m.ManifestStream.Blocks, signedHash)
86 type ManifestWriter struct {
89 Streams map[string]*ManifestStreamWriter
92 func (m *ManifestWriter) WalkFunc(path string, info os.FileInfo, err error) error {
98 if len(path) > (len(m.stripPrefix) + len(info.Name()) + 1) {
99 dir = path[len(m.stripPrefix)+1 : (len(path) - len(info.Name()) - 1)]
105 fn := path[(len(path) - len(info.Name())):]
107 if m.Streams[dir] == nil {
108 m.Streams[dir] = &ManifestStreamWriter{
110 &manifest.ManifestStream{StreamName: dir},
115 go m.Streams[dir].goUpload()
118 stream := m.Streams[dir]
120 fileStart := stream.offset
122 file, err := os.Open(path)
127 log.Printf("Uploading %v/%v (%v bytes)", dir, fn, info.Size())
130 count, err = io.Copy(stream, file)
135 stream.offset += count
137 stream.ManifestStream.FileStreamSegments = append(stream.ManifestStream.FileStreamSegments,
138 manifest.FileStreamSegment{uint64(fileStart), uint64(count), fn})
143 func (m *ManifestWriter) Finish() error {
145 for _, stream := range m.Streams {
146 if stream.uploader == nil {
149 if stream.Block != nil {
150 stream.uploader <- stream.Block
152 close(stream.uploader)
153 stream.uploader = nil
155 errors := <-stream.finish
159 for _, r := range errors {
160 errstring = fmt.Sprintf("%v%v\n", errstring, r.Error())
164 return errors.New(errstring)
170 func (m *ManifestWriter) ManifestText() string {
174 dirs := make([]string, len(m.Streams))
176 for k := range m.Streams {
182 for _, k := range dirs {
188 k = strings.Replace(k, " ", "\\040", -1)
189 k = strings.Replace(k, "\n", "", -1)
190 buf.WriteString("./" + k)
192 for _, b := range v.Blocks {
196 for _, f := range v.FileStreamSegments {
198 name := strings.Replace(f.Name, " ", "\\040", -1)
199 name = strings.Replace(name, "\n", "", -1)
200 buf.WriteString(fmt.Sprintf("%d:%d:%s", f.SegPos, f.SegLen, name))
202 buf.WriteString("\n")
207 func WriteTree(kc IKeepClient, root string) (manifest string, err error) {
208 mw := ManifestWriter{kc, root, map[string]*ManifestStreamWriter{}}
209 err = filepath.Walk(root, mw.WalkFunc)
220 return mw.ManifestText(), nil