// Package internetarchive provides an interface to Internet Archive's Item // via their native API than using S3-compatible endpoints. package internetarchive import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "net/http" "net/url" "path" "regexp" "strconv" "strings" "time" "github.com/ncw/swift/v2" "github.com/rclone/rclone/fs" "github.com/rclone/rclone/fs/config" "github.com/rclone/rclone/fs/config/configmap" "github.com/rclone/rclone/fs/config/configstruct" "github.com/rclone/rclone/fs/fserrors" "github.com/rclone/rclone/fs/fshttp" "github.com/rclone/rclone/fs/hash" "github.com/rclone/rclone/lib/bucket" "github.com/rclone/rclone/lib/encoder" "github.com/rclone/rclone/lib/pacer" "github.com/rclone/rclone/lib/random" "github.com/rclone/rclone/lib/rest" ) // Register with Fs func init() { fs.Register(&fs.RegInfo{ Name: "internetarchive", Description: "Internet Archive", NewFs: NewFs, Options: []fs.Option{{ Name: "access_key_id", Help: "IAS3 Access Key.\n\nLeave blank for anonymous access.\nYou can find one here: https://archive.org/account/s3.php", }, { Name: "secret_access_key", Help: "IAS3 Secret Key (password).\n\nLeave blank for anonymous access.", }, { // their official client (https://github.com/jjjake/internetarchive) hardcodes following the two Name: "endpoint", Help: "IAS3 Endpoint.\n\nLeave blank for default value.", Default: "https://s3.us.archive.org", Advanced: true, }, { Name: "front_endpoint", Help: "Host of InternetArchive Frontend.\n\nLeave blank for default value.", Default: "https://archive.org", Advanced: true, }, { Name: "disable_checksum", Help: `Don't ask the server to test against MD5 checksum calculated by rclone. Normally rclone will calculate the MD5 checksum of the input before uploading it so it can ask the server to check the object against checksum. This is great for data integrity checking but can cause long delays for large files to start uploading.`, Default: true, Advanced: true, }, { Name: "wait_archive", Help: `Timeout for waiting the server's processing tasks (specifically archive and book_op) to finish. Only enable if you need to be guaranteed to be reflected after write operations. 0 to disable waiting. No errors to be thrown in case of timeout.`, Default: fs.Duration(0), Advanced: true, }, { Name: config.ConfigEncoding, Help: config.ConfigEncodingHelp, Advanced: true, Default: encoder.EncodeZero | encoder.EncodeSlash | encoder.EncodeLtGt | encoder.EncodeCrLf | encoder.EncodeDel | encoder.EncodeCtl | encoder.EncodeInvalidUtf8 | encoder.EncodeDot, }, }}) } // maximum size of an item. this is constant across all items const iaItemMaxSize int64 = 1099511627776 // Options defines the configuration for this backend type Options struct { AccessKeyID string `config:"access_key_id"` SecretAccessKey string `config:"secret_access_key"` Endpoint string `config:"endpoint"` FrontEndpoint string `config:"front_endpoint"` DisableChecksum bool `config:"disable_checksum"` WaitArchive fs.Duration `config:"wait_archive"` Enc encoder.MultiEncoder `config:"encoding"` } // Fs represents an IAS3 remote type Fs struct { name string // name of this remote root string // the path we are working on if any opt Options // parsed config options features *fs.Features // optional features srv *rest.Client // the connection to IAS3 front *rest.Client // the connection to frontend pacer *fs.Pacer // pacer for API calls ctx context.Context } // Object describes a file at IA type Object struct { fs *Fs // reference to Fs remote string // the remote path modTime time.Time // last modified time size int64 // size of the file in bytes md5 string // md5 hash of the file presented by the server sha1 string // sha1 hash of the file presented by the server crc32 string // crc32 of the file presented by the server } // IAFile reprensents a subset of object in MetadataResponse.Files type IAFile struct { Name string `json:"name"` // Source string `json:"source"` Mtime string `json:"mtime"` RcloneMtime json.RawMessage `json:"rclone-mtime"` UpdateTrack json.RawMessage `json:"rclone-update-track"` Size string `json:"size"` Md5 string `json:"md5"` Crc32 string `json:"crc32"` Sha1 string `json:"sha1"` } // MetadataResponse reprensents subset of the JSON object returned by (frontend)/metadata/ type MetadataResponse struct { Files []IAFile `json:"files"` ItemSize int64 `json:"item_size"` } // ModMetadataResponse represents response for amending metadata type ModMetadataResponse struct { // https://archive.org/services/docs/api/md-write.html#example Success bool `json:"success"` Error string `json:"error"` } // Name of the remote (as passed into NewFs) func (f *Fs) Name() string { return f.name } // Root of the remote (as passed into NewFs) func (f *Fs) Root() string { return f.root } // String converts this Fs to a string func (f *Fs) String() string { bucket, file := f.split("") if bucket == "" { return "Internet Archive root" } if file == "" { return fmt.Sprintf("Internet Archive item %s", bucket) } return fmt.Sprintf("Internet Archive item %s path %s", bucket, file) } // Features returns the optional features of this Fs func (f *Fs) Features() *fs.Features { return f.features } // Hashes returns type of hashes supported by IA func (f *Fs) Hashes() hash.Set { return hash.NewHashSet(hash.MD5, hash.SHA1, hash.CRC32) } // Precision returns the precision of mtime that the server responds func (f *Fs) Precision() time.Duration { if f.opt.WaitArchive == 0 { return fs.ModTimeNotSupported } return time.Nanosecond } // retryErrorCodes is a slice of error codes that we will retry // See: https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html var retryErrorCodes = []int{ 429, // Too Many Requests 500, // Internal Server Error - "We encountered an internal error. Please try again." 503, // Service Unavailable/Slow Down - "Reduce your request rate" } // NewFs constructs an Fs from the path func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, error) { // Parse config into Options struct opt := new(Options) err := configstruct.Set(m, opt) if err != nil { return nil, err } // Parse the endpoints ep, err := url.Parse(opt.Endpoint) if err != nil { return nil, err } fe, err := url.Parse(opt.FrontEndpoint) if err != nil { return nil, err } root = strings.Trim(root, "/") f := &Fs{ name: name, opt: *opt, ctx: ctx, } f.setRoot(root) f.features = (&fs.Features{ BucketBased: true, }).Fill(ctx, f) f.srv = rest.NewClient(fshttp.NewClient(ctx)) f.srv.SetRoot(ep.String()) f.front = rest.NewClient(fshttp.NewClient(ctx)) f.front.SetRoot(fe.String()) if opt.AccessKeyID != "" && opt.SecretAccessKey != "" { auth := fmt.Sprintf("LOW %s:%s", opt.AccessKeyID, opt.SecretAccessKey) f.srv.SetHeader("Authorization", auth) f.front.SetHeader("Authorization", auth) } f.pacer = fs.NewPacer(ctx, pacer.NewS3(pacer.MinSleep(10*time.Millisecond))) // test if the root exists as a file _, err = f.NewObject(ctx, "/") if err == nil { f.setRoot(betterPathDir(root)) return f, fs.ErrorIsFile } return f, nil } // setRoot changes the root of the Fs func (f *Fs) setRoot(root string) { f.root = strings.Trim(root, "/") } // Remote returns the remote path func (o *Object) Remote() string { return o.remote } // ModTime is the last modified time (read-only) func (o *Object) ModTime(ctx context.Context) time.Time { return o.modTime } // Size is the file length func (o *Object) Size() int64 { return o.size } // Fs returns the parent Fs func (o *Object) Fs() fs.Info { return o.fs } // Hash returns the hash value presented by IA func (o *Object) Hash(ctx context.Context, ty hash.Type) (string, error) { if ty == hash.MD5 { return o.md5, nil } if ty == hash.SHA1 { return o.sha1, nil } if ty == hash.CRC32 { return o.crc32, nil } return "", hash.ErrUnsupported } // Storable returns if this object is storable func (o *Object) Storable() bool { return true } // SetModTime sets modTime on a particular file func (o *Object) SetModTime(ctx context.Context, t time.Time) (err error) { bucket, reqDir := o.split() if bucket == "" { return fs.ErrorCantSetModTime } if reqDir == "" { return fs.ErrorCantSetModTime } // https://archive.org/services/docs/api/md-write.html var patch = []interface{}{ // we should drop it first to clear all rclone-provided mtimes struct { Op string `json:"op"` Path string `json:"path"` }{"remove", "/rclone-mtime"}, struct { Op string `json:"op"` Path string `json:"path"` Value string `json:"value"` }{"add", "/rclone-mtime", t.Format(time.RFC3339Nano)}, } res, err := json.Marshal(patch) if err != nil { return err } params := url.Values{} params.Add("-target", fmt.Sprintf("files/%s", reqDir)) params.Add("-patch", string(res)) body := []byte(params.Encode()) bodyLen := int64(len(body)) var resp *http.Response var result ModMetadataResponse // make a POST request to (frontend)/metadata/:item/ opts := rest.Opts{ Method: "POST", Path: path.Join("/metadata/", bucket), Body: bytes.NewReader(body), ContentLength: &bodyLen, ContentType: "application/x-www-form-urlencoded", } err = o.fs.pacer.Call(func() (bool, error) { resp, err = o.fs.front.CallJSON(ctx, &opts, nil, &result) return o.fs.shouldRetry(resp, err) }) if err != nil { return err } if result.Success { o.modTime = t return nil } return errors.New(result.Error) } // List files and directories in a directory func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err error) { bucket, reqDir := f.split(dir) if bucket == "" { if reqDir != "" { return nil, fs.ErrorListBucketRequired } return entries, nil } grandparent := f.opt.Enc.ToStandardPath(strings.Trim(path.Join(bucket, reqDir), "/") + "/") allEntries, err := f.listAllUnconstrained(ctx, bucket) if err != nil { return entries, err } for _, ent := range allEntries { obj, ok := ent.(*Object) if ok && strings.HasPrefix(obj.remote, grandparent) { path := trimPathPrefix(obj.remote, grandparent, f.opt.Enc) if !strings.Contains(path, "/") { obj.remote = trimPathPrefix(obj.remote, f.root, f.opt.Enc) entries = append(entries, obj) } } dire, ok := ent.(*fs.Dir) if ok && strings.HasPrefix(dire.Remote(), grandparent) { path := trimPathPrefix(dire.Remote(), grandparent, f.opt.Enc) if !strings.Contains(path, "/") { dire.SetRemote(trimPathPrefix(dire.Remote(), f.root, f.opt.Enc)) entries = append(entries, dire) } } } return entries, nil } // Mkdir can't be performed on IA like git repositories func (f *Fs) Mkdir(ctx context.Context, dir string) (err error) { return nil } // Rmdir as well, unless we're asked for recursive deletion func (f *Fs) Rmdir(ctx context.Context, dir string) error { return nil } // NewObject finds the Object at remote. If it can't be found // it returns the error fs.ErrorObjectNotFound. func (f *Fs) NewObject(ctx context.Context, remote string) (ret fs.Object, err error) { bucket, filepath := f.split(remote) filepath = strings.Trim(filepath, "/") if bucket == "" { if filepath != "" { return nil, fs.ErrorListBucketRequired } return nil, fs.ErrorIsDir } grandparent := f.opt.Enc.ToStandardPath(strings.Trim(path.Join(bucket, filepath), "/")) allEntries, err := f.listAllUnconstrained(ctx, bucket) if err != nil { return nil, err } for _, ent := range allEntries { obj, ok := ent.(*Object) if ok && obj.remote == grandparent { obj.remote = trimPathPrefix(obj.remote, f.root, f.opt.Enc) return obj, nil } } return nil, fs.ErrorObjectNotFound } // Put uploads a file func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) { o := &Object{ fs: f, remote: src.Remote(), modTime: src.ModTime(ctx), size: src.Size(), } err := o.Update(ctx, in, src, options...) if err == nil { return o, nil } return nil, err } // PublicLink generates a public link to the remote path (usually readable by anyone) func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration, unlink bool) (link string, err error) { if strings.HasSuffix(remote, "/") { return "", fs.ErrorCantShareDirectories } if _, err := f.NewObject(ctx, remote); err != nil { return "", err } bucket, bucketPath := f.split(remote) return path.Join(f.opt.FrontEndpoint, "/download/", bucket, bucketPath), nil } // Copy src to this remote using server-side copy operations. // // This is stored with the remote path given // // It returns the destination Object and a possible error // // Will only be called if src.Fs().Name() == f.Name() // // If it isn't possible then return fs.ErrorCantCopy func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (_ fs.Object, err error) { dstBucket, dstPath := f.split(remote) srcObj, ok := src.(*Object) if !ok { fs.Debugf(src, "Can't copy - not same remote type") return nil, fs.ErrorCantCopy } srcBucket, srcPath := srcObj.split() if dstBucket == srcBucket && dstPath == srcPath { // https://github.com/jjjake/internetarchive/blob/2456376533251df9d05e0a14d796ec1ced4959f5/internetarchive/cli/ia_copy.py#L68 fs.Debugf(src, "Can't copy - the source and destination files cannot be the same!") return nil, fs.ErrorCantCopy } updateTracker := random.String(32) headers := map[string]string{ "x-archive-auto-make-bucket": "1", "x-archive-queue-derive": "0", "x-archive-keep-old-version": "0", "x-amz-copy-source": quotePath(path.Join("/", srcBucket, srcPath)), "x-amz-metadata-directive": "COPY", "x-archive-filemeta-sha1": srcObj.sha1, "x-archive-filemeta-md5": srcObj.md5, "x-archive-filemeta-crc32": srcObj.crc32, "x-archive-filemeta-size": fmt.Sprint(srcObj.size), // add this too for sure "x-archive-filemeta-rclone-mtime": srcObj.modTime.Format(time.RFC3339Nano), "x-archive-filemeta-rclone-update-track": updateTracker, } // make a PUT request at (IAS3)/:item/:path without body var resp *http.Response opts := rest.Opts{ Method: "PUT", Path: "/" + url.PathEscape(path.Join(dstBucket, dstPath)), ExtraHeaders: headers, } err = f.pacer.Call(func() (bool, error) { resp, err = f.srv.Call(ctx, &opts) return f.shouldRetry(resp, err) }) if err != nil { return nil, err } // we can't update/find metadata here as IA will also // queue server-side copy as well as upload/delete. return f.waitFileUpload(ctx, trimPathPrefix(path.Join(dstBucket, dstPath), f.root, f.opt.Enc), updateTracker, srcObj.size) } // ListR lists the objects and directories of the Fs starting // from dir recursively into out. // // dir should be "" to start from the root, and should not // have trailing slashes. // // This should return ErrDirNotFound if the directory isn't // found. // // It should call callback for each tranche of entries read. // These need not be returned in any particular order. If // callback returns an error then the listing will stop // immediately. // // Don't implement this unless you have a more efficient way // of listing recursively than doing a directory traversal. func (f *Fs) ListR(ctx context.Context, dir string, callback fs.ListRCallback) (err error) { var allEntries, entries fs.DirEntries bucket, reqDir := f.split(dir) if bucket == "" { if reqDir != "" { return fs.ErrorListBucketRequired } return callback(entries) } grandparent := f.opt.Enc.ToStandardPath(strings.Trim(path.Join(bucket, reqDir), "/") + "/") allEntries, err = f.listAllUnconstrained(ctx, bucket) if err != nil { return err } for _, ent := range allEntries { obj, ok := ent.(*Object) if ok && strings.HasPrefix(obj.remote, grandparent) { obj.remote = trimPathPrefix(obj.remote, f.root, f.opt.Enc) entries = append(entries, obj) } dire, ok := ent.(*fs.Dir) if ok && strings.HasPrefix(dire.Remote(), grandparent) { dire.SetRemote(trimPathPrefix(dire.Remote(), f.root, f.opt.Enc)) entries = append(entries, dire) } } return callback(entries) } // CleanUp removes all files inside history/ func (f *Fs) CleanUp(ctx context.Context) (err error) { bucket, _ := f.split("/") if bucket == "" { return fs.ErrorListBucketRequired } entries, err := f.listAllUnconstrained(ctx, bucket) if err != nil { return err } for _, ent := range entries { obj, ok := ent.(*Object) if ok && strings.HasPrefix(obj.remote, bucket+"/history/") { err = obj.Remove(ctx) if err != nil { return err } } // we can fully ignore directories, as they're just virtual entries to // comply with rclone's requirement } return nil } // About returns things about remaining and used spaces func (f *Fs) About(ctx context.Context) (_ *fs.Usage, err error) { bucket, _ := f.split("/") if bucket == "" { return nil, fs.ErrorListBucketRequired } result, err := f.requestMetadata(ctx, bucket) if err != nil { return nil, err } // perform low-level operation here since it's ridiculous to make 2 same requests var historySize int64 for _, ent := range result.Files { if strings.HasPrefix(ent.Name, "history/") { size := parseSize(ent.Size) if size < 0 { // parse error can be ignored since it's not fatal continue } historySize += size } } usage := &fs.Usage{ Total: fs.NewUsageValue(iaItemMaxSize), Free: fs.NewUsageValue(iaItemMaxSize - result.ItemSize), Used: fs.NewUsageValue(result.ItemSize), Trashed: fs.NewUsageValue(historySize), // bytes in trash } return usage, nil } // Open an object for read func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.ReadCloser, err error) { var optionsFixed []fs.OpenOption for _, opt := range options { if optRange, ok := opt.(*fs.RangeOption); ok { // Ignore range option if file is empty if o.Size() == 0 && optRange.Start == 0 && optRange.End > 0 { continue } } optionsFixed = append(optionsFixed, opt) } var resp *http.Response // make a GET request to (frontend)/download/:item/:path opts := rest.Opts{ Method: "GET", Path: path.Join("/download/", o.fs.root, o.fs.opt.Enc.FromStandardPath(o.remote)), Options: optionsFixed, } err = o.fs.pacer.Call(func() (bool, error) { resp, err = o.fs.front.Call(ctx, &opts) return o.fs.shouldRetry(resp, err) }) if err != nil { return nil, err } return resp.Body, nil } // Update the Object from in with modTime and size func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (err error) { bucket, bucketPath := o.split() modTime := src.ModTime(ctx) size := src.Size() updateTracker := random.String(32) // Set the mtime in the metadata // internetarchive backend builds at header level as IAS3 has extension outside X-Amz- headers := map[string]string{ // https://github.com/jjjake/internetarchive/blob/2456376533251df9d05e0a14d796ec1ced4959f5/internetarchive/iarequest.py#L158 "x-amz-filemeta-rclone-mtime": modTime.Format(time.RFC3339Nano), "x-amz-filemeta-rclone-update-track": updateTracker, // we add some more headers for intuitive actions "x-amz-auto-make-bucket": "1", // create an item if does not exist, do nothing if already "x-archive-auto-make-bucket": "1", // same as above in IAS3 original way "x-archive-keep-old-version": "0", // do not keep old versions (a.k.a. trashes in other clouds) "x-archive-meta-mediatype": "data", // mark media type of the uploading file as "data" "x-archive-queue-derive": "0", // skip derivation process (e.g. encoding to smaller files, OCR on PDFs) "x-archive-cascade-delete": "1", // enable "cascate delete" (delete all derived files in addition to the file itself) } if size >= 0 { headers["Content-Length"] = fmt.Sprintf("%d", size) headers["x-archive-size-hint"] = fmt.Sprintf("%d", size) } // read the md5sum if available var md5sumHex string if !o.fs.opt.DisableChecksum { md5sumHex, err = src.Hash(ctx, hash.MD5) if err == nil && matchMd5.MatchString(md5sumHex) { // Set the md5sum in header on the object if // the user wants it // https://github.com/jjjake/internetarchive/blob/245637653/internetarchive/item.py#L969 headers["Content-MD5"] = md5sumHex } } // make a PUT request at (IAS3)/encoded(:item/:path) var resp *http.Response opts := rest.Opts{ Method: "PUT", Path: "/" + url.PathEscape(path.Join(bucket, bucketPath)), Body: in, ContentLength: &size, ExtraHeaders: headers, } err = o.fs.pacer.Call(func() (bool, error) { resp, err = o.fs.srv.Call(ctx, &opts) return o.fs.shouldRetry(resp, err) }) // we can't update/find metadata here as IA will "ingest" uploaded file(s) // upon uploads. (you can find its progress at https://archive.org/history/ItemNameHere ) // or we have to wait for finish? (needs polling (frontend)/metadata/:item or scraping (frontend)/history/:item) var newObj *Object if err == nil { newObj, err = o.fs.waitFileUpload(ctx, o.remote, updateTracker, size) } else { newObj = &Object{} } o.crc32 = newObj.crc32 o.md5 = newObj.md5 o.sha1 = newObj.sha1 o.modTime = newObj.modTime o.size = newObj.size return err } // Remove an object func (o *Object) Remove(ctx context.Context) (err error) { bucket, bucketPath := o.split() // make a DELETE request at (IAS3)/:item/:path var resp *http.Response opts := rest.Opts{ Method: "DELETE", Path: "/" + url.PathEscape(path.Join(bucket, bucketPath)), } err = o.fs.pacer.Call(func() (bool, error) { resp, err = o.fs.srv.Call(ctx, &opts) return o.fs.shouldRetry(resp, err) }) // deleting files can take bit longer as // it'll be processed on same queue as uploads if err == nil { err = o.fs.waitDelete(ctx, bucket, bucketPath) } return err } // String converts this Fs to a string func (o *Object) String() string { if o == nil { return "" } return o.remote } func (f *Fs) shouldRetry(resp *http.Response, err error) (bool, error) { if resp != nil { for _, e := range retryErrorCodes { if resp.StatusCode == e { return true, err } } } // Ok, not an awserr, check for generic failure conditions return fserrors.ShouldRetry(err), err } var matchMd5 = regexp.MustCompile(`^[0-9a-f]{32}$`) // split returns bucket and bucketPath from the rootRelativePath // relative to f.root func (f *Fs) split(rootRelativePath string) (bucketName, bucketPath string) { bucketName, bucketPath = bucket.Split(path.Join(f.root, rootRelativePath)) return f.opt.Enc.FromStandardName(bucketName), f.opt.Enc.FromStandardPath(bucketPath) } // split returns bucket and bucketPath from the object func (o *Object) split() (bucket, bucketPath string) { return o.fs.split(o.remote) } func (f *Fs) requestMetadata(ctx context.Context, bucket string) (result MetadataResponse, err error) { var resp *http.Response // make a GET request to (frontend)/metadata/:item/ opts := rest.Opts{ Method: "GET", Path: path.Join("/metadata/", bucket), } err = f.pacer.Call(func() (bool, error) { resp, err = f.front.CallJSON(ctx, &opts, nil, &result) return f.shouldRetry(resp, err) }) return result, err } // list up all files/directories without any filters func (f *Fs) listAllUnconstrained(ctx context.Context, bucket string) (entries fs.DirEntries, err error) { result, err := f.requestMetadata(ctx, bucket) if err != nil { return nil, err } knownDirs := map[string]time.Time{ "": time.Unix(0, 0), } for _, file := range result.Files { dir := strings.Trim(betterPathDir(file.Name), "/") nameWithBucket := path.Join(bucket, file.Name) mtimeTime := file.parseMtime() // populate children directories child := dir for { if _, ok := knownDirs[child]; ok { break } // directory d := fs.NewDir(f.opt.Enc.ToStandardPath(path.Join(bucket, child)), mtimeTime) entries = append(entries, d) knownDirs[child] = mtimeTime child = strings.Trim(betterPathDir(child), "/") } if _, ok := knownDirs[betterPathDir(file.Name)]; !ok { continue } size := parseSize(file.Size) o := makeValidObject(f, f.opt.Enc.ToStandardPath(nameWithBucket), file, mtimeTime, size) entries = append(entries, o) } return entries, nil } func (f *Fs) waitFileUpload(ctx context.Context, reqPath, tracker string, newSize int64) (ret *Object, err error) { bucket, bucketPath := f.split(reqPath) ret = &Object{ fs: f, remote: trimPathPrefix(path.Join(bucket, bucketPath), f.root, f.opt.Enc), modTime: time.Unix(0, 0), size: -1, } if f.opt.WaitArchive == 0 { // user doesn't want to poll, let's not ret2, err := f.NewObject(ctx, reqPath) if err == nil { ret2, ok := ret2.(*Object) if ok { ret = ret2 ret.crc32 = "" ret.md5 = "" ret.sha1 = "" ret.size = -1 } } return ret, nil } retC := make(chan struct { *Object error }, 1) go func() { isFirstTime := true existed := false for { if !isFirstTime { // depending on the queue, it takes time time.Sleep(10 * time.Second) } metadata, err := f.requestMetadata(ctx, bucket) if err != nil { retC <- struct { *Object error }{ret, err} return } var iaFile *IAFile for _, f := range metadata.Files { if f.Name == bucketPath { iaFile = &f break } } if isFirstTime { isFirstTime = false existed = iaFile != nil } if iaFile == nil { continue } if !existed && !isFirstTime { // fast path: file wasn't exited before retC <- struct { *Object error }{makeValidObject2(f, *iaFile, bucket), nil} return } fileTrackers, _ := listOrString(iaFile.UpdateTrack) trackerMatch := false for _, v := range fileTrackers { if v == tracker { trackerMatch = true break } } if !trackerMatch { continue } if !compareSize(parseSize(iaFile.Size), newSize) { continue } // voila! retC <- struct { *Object error }{makeValidObject2(f, *iaFile, bucket), nil} return } }() select { case res := <-retC: return res.Object, res.error case <-time.After(time.Duration(f.opt.WaitArchive)): return ret, nil } } func (f *Fs) waitDelete(ctx context.Context, bucket, bucketPath string) (err error) { if f.opt.WaitArchive == 0 { // user doesn't want to poll, let's not return nil } retC := make(chan error, 1) go func() { for { metadata, err := f.requestMetadata(ctx, bucket) if err != nil { retC <- err return } found := false for _, f := range metadata.Files { if f.Name == bucketPath { found = true break } } if !found { retC <- nil return } // depending on the queue, it takes time time.Sleep(10 * time.Second) } }() select { case res := <-retC: return res case <-time.After(time.Duration(f.opt.WaitArchive)): return nil } } func makeValidObject(f *Fs, remote string, file IAFile, mtime time.Time, size int64) *Object { return &Object{ fs: f, remote: remote, modTime: mtime, size: size, md5: file.Md5, crc32: file.Crc32, sha1: file.Sha1, } } func makeValidObject2(f *Fs, file IAFile, bucket string) *Object { mtimeTime := file.parseMtime() size := parseSize(file.Size) return makeValidObject(f, trimPathPrefix(path.Join(bucket, file.Name), f.root, f.opt.Enc), file, mtimeTime, size) } func listOrString(jm json.RawMessage) (rmArray []string, err error) { // rclone-metadata can be an array or string // try to deserialize it as array first err = json.Unmarshal(jm, &rmArray) if err != nil { // if not, it's a string dst := new(string) err = json.Unmarshal(jm, dst) if err == nil { rmArray = []string{*dst} } } return } func (file IAFile) parseMtime() (mtime time.Time) { // method 1: use metadata added by rclone rmArray, err := listOrString(file.RcloneMtime) // let's take the first value we can deserialize for _, value := range rmArray { mtime, err = time.Parse(time.RFC3339Nano, value) if err == nil { break } } if err != nil { // method 2: use metadata added by IAS3 mtime, err = swift.FloatStringToTime(file.Mtime) } if err != nil { // metadata files don't have some of the fields mtime = time.Unix(0, 0) } return mtime } func compareSize(a, b int64) bool { if a < 0 || b < 0 { // we won't compare if any of them is not known return true } return a == b } func parseSize(str string) int64 { size, err := strconv.ParseInt(str, 10, 64) if err != nil { size = -1 } return size } func betterPathDir(p string) string { d := path.Dir(p) if d == "." { return "" } return d } func betterPathClean(p string) string { d := path.Clean(p) if d == "." { return "" } return d } func trimPathPrefix(s, prefix string, enc encoder.MultiEncoder) string { // we need to clean the paths to make tests pass! s = betterPathClean(s) prefix = betterPathClean(prefix) if s == prefix || s == prefix+"/" { return "" } prefix = enc.ToStandardPath(strings.TrimRight(prefix, "/")) return enc.ToStandardPath(strings.TrimPrefix(s, prefix+"/")) } // mimicks urllib.parse.quote() on Python; exclude / from url.PathEscape func quotePath(s string) string { seg := strings.Split(s, "/") newValues := []string{} for _, v := range seg { newValues = append(newValues, url.PathEscape(v)) } return strings.Join(newValues, "/") } var ( _ fs.Fs = &Fs{} _ fs.Copier = &Fs{} _ fs.ListRer = &Fs{} _ fs.CleanUpper = &Fs{} _ fs.PublicLinker = &Fs{} _ fs.Abouter = &Fs{} _ fs.Object = &Object{} )