Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 bucket.go // Copyright (C) MongoDB, Inc. 2017-present. // // Licensed under the Apache License, Version 2.0 (the "License"); you may // not use this file except in compliance with the License. You may obtain // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 package gridfs // import "go.mongodb.org/mongo-driver/mongo/gridfs" import ( "bytes" "context" "errors" "fmt" "io" "time" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/internal/csot" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" "go.mongodb.org/mongo-driver/mongo/readconcern" "go.mongodb.org/mongo-driver/mongo/readpref" "go.mongodb.org/mongo-driver/mongo/writeconcern" "go.mongodb.org/mongo-driver/x/bsonx/bsoncore" ) // TODO: add sessions options // DefaultChunkSize is the default size of each file chunk. const DefaultChunkSize int32 = 255 * 1024 // 255 KiB // ErrFileNotFound occurs if a user asks to download a file with a file ID that isn't found in the files collection. var ErrFileNotFound = errors.New("file with given parameters not found") // ErrMissingChunkSize occurs when downloading a file if the files collection document is missing the "chunkSize" field. var ErrMissingChunkSize = errors.New("files collection document does not contain a 'chunkSize' field") // Bucket represents a GridFS bucket. type Bucket struct { db mongo.Database chunksColl mongo.Collection // collection to store file chunks filesColl mongo.Collection // collection to store file metadata name string chunkSize int32 wc writeconcern.WriteConcern rc readconcern.ReadConcern rp readpref.ReadPref firstWriteDone bool readBuf []byte writeBuf []byte readDeadline time.Time writeDeadline time.Time } // Upload contains options to upload a file to a bucket. type Upload struct { chunkSize int32 metadata bson.D } // NewBucket creates a GridFS bucket. func NewBucket(db mongo.Database, opts ...options.BucketOptions) (Bucket, error) { b := &Bucket{ name: "fs", chunkSize: DefaultChunkSize, db: db, wc: db.WriteConcern(), rc: db.ReadConcern(), rp: db.ReadPreference(), } bo := options.MergeBucketOptions(opts...) if bo.Name != nil { b.name = bo.Name } if bo.ChunkSizeBytes != nil { b.chunkSize = bo.ChunkSizeBytes } if bo.WriteConcern != nil { b.wc = bo.WriteConcern } if bo.ReadConcern != nil { b.rc = bo.ReadConcern } if bo.ReadPreference != nil { b.rp = bo.ReadPreference } var collOpts = options.Collection().SetWriteConcern(b.wc).SetReadConcern(b.rc).SetReadPreference(b.rp) b.chunksColl = db.Collection(b.name+".chunks", collOpts) b.filesColl = db.Collection(b.name+".files", collOpts) b.readBuf = make([]byte, b.chunkSize) b.writeBuf = make([]byte, b.chunkSize) return b, nil } // SetWriteDeadline sets the write deadline for this bucket. func (b Bucket) SetWriteDeadline(t time.Time) error { b.writeDeadline = t return nil } // SetReadDeadline sets the read deadline for this bucket func (b Bucket) SetReadDeadline(t time.Time) error { b.readDeadline = t return nil } // OpenUploadStream creates a file ID new upload stream for a file given the filename. func (b Bucket) OpenUploadStream(filename string, opts ...options.UploadOptions) (UploadStream, error) { return b.OpenUploadStreamWithID(primitive.NewObjectID(), filename, opts...) } // OpenUploadStreamWithID creates a new upload stream for a file given the file ID and filename. func (b Bucket) OpenUploadStreamWithID(fileID interface{}, filename string, opts ...options.UploadOptions) (UploadStream, error) { ctx, cancel := deadlineContext(b.writeDeadline) if cancel != nil { defer cancel() } if err := b.checkFirstWrite(ctx); err != nil { return nil, err } upload, err := b.parseUploadOptions(opts...) if err != nil { return nil, err } return newUploadStream(upload, fileID, filename, b.chunksColl, b.filesColl), nil } // UploadFromStream creates a fileID and uploads a file given a source stream. // // If this upload requires a custom write deadline to be set on the bucket, it cannot be done concurrently with other // write operations operations on this bucket that also require a custom deadline. func (b Bucket) UploadFromStream(filename string, source io.Reader, opts ...options.UploadOptions) (primitive.ObjectID, error) { fileID := primitive.NewObjectID() err := b.UploadFromStreamWithID(fileID, filename, source, opts...) return fileID, err } // UploadFromStreamWithID uploads a file given a source stream. // // If this upload requires a custom write deadline to be set on the bucket, it cannot be done concurrently with other // write operations operations on this bucket that also require a custom deadline. func (b Bucket) UploadFromStreamWithID(fileID interface{}, filename string, source io.Reader, opts ...options.UploadOptions) error { us, err := b.OpenUploadStreamWithID(fileID, filename, opts...) if err != nil { return err } err = us.SetWriteDeadline(b.writeDeadline) if err != nil { _ = us.Close() return err } for { n, err := source.Read(b.readBuf) if err != nil && err != io.EOF { _ = us.Abort() // upload considered aborted if source stream returns an error return err } if n > 0 { _, err := us.Write(b.readBuf[:n]) if err != nil { return err } } if n == 0 \|\| err == io.EOF { break } } return us.Close() } // OpenDownloadStream creates a stream from which the contents of the file can be read. func (b Bucket) OpenDownloadStream(fileID interface{}) (DownloadStream, error) { return b.openDownloadStream(bson.D{ {"_id", fileID}, }) } // DownloadToStream downloads the file with the specified fileID and writes it to the provided io.Writer. // Returns the number of bytes written to the stream and an error, or nil if there was no error. // // If this download requires a custom read deadline to be set on the bucket, it cannot be done concurrently with other // read operations operations on this bucket that also require a custom deadline. func (b Bucket) DownloadToStream(fileID interface{}, stream io.Writer) (int64, error) { ds, err := b.OpenDownloadStream(fileID) if err != nil { return 0, err } return b.downloadToStream(ds, stream) } // OpenDownloadStreamByName opens a download stream for the file with the given filename. func (b Bucket) OpenDownloadStreamByName(filename string, opts ...options.NameOptions) (DownloadStream, error) { var numSkip int32 = -1 var sortOrder int32 = 1 nameOpts := options.MergeNameOptions(opts...) if nameOpts.Revision != nil { numSkip = nameOpts.Revision } if numSkip < 0 { sortOrder = -1 numSkip = (-1 * numSkip) - 1 } findOpts := options.Find().SetSkip(int64(numSkip)).SetSort(bson.D{{"uploadDate", sortOrder}}) return b.openDownloadStream(bson.D{{"filename", filename}}, findOpts) } // DownloadToStreamByName downloads the file with the given name to the given io.Writer. // // If this download requires a custom read deadline to be set on the bucket, it cannot be done concurrently with other // read operations operations on this bucket that also require a custom deadline. func (b Bucket) DownloadToStreamByName(filename string, stream io.Writer, opts ...options.NameOptions) (int64, error) { ds, err := b.OpenDownloadStreamByName(filename, opts...) if err != nil { return 0, err } return b.downloadToStream(ds, stream) } // Delete deletes all chunks and metadata associated with the file with the given file ID. // // If this operation requires a custom write deadline to be set on the bucket, it cannot be done concurrently with other // write operations operations on this bucket that also require a custom deadline. // // Use SetWriteDeadline to set a deadline for the delete operation. func (b Bucket) Delete(fileID interface{}) error { ctx, cancel := deadlineContext(b.writeDeadline) if cancel != nil { defer cancel() } return b.DeleteContext(ctx, fileID) } // DeleteContext deletes all chunks and metadata associated with the file with the given file ID and runs the underlying // delete operations with the provided context. // // Use the context parameter to time-out or cancel the delete operation. The deadline set by SetWriteDeadline is ignored. func (b Bucket) DeleteContext(ctx context.Context, fileID interface{}) error { // If no deadline is set on the passed-in context, Timeout is set on the Client, and context is // not already a Timeout context, honor Timeout in new Timeout context for operation execution to // be shared by both delete operations. if _, deadlineSet := ctx.Deadline(); !deadlineSet && b.db.Client().Timeout() != nil && !csot.IsTimeoutContext(ctx) { newCtx, cancelFunc := csot.MakeTimeoutContext(ctx, b.db.Client().Timeout()) // Redefine ctx to be the new timeout-derived context. ctx = newCtx // Cancel the timeout-derived context at the end of Execute to avoid a context leak. defer cancelFunc() } // Delete document in files collection and then chunks to minimize race conditions. res, err := b.filesColl.DeleteOne(ctx, bson.D{{"_id", fileID}}) if err == nil && res.DeletedCount == 0 { err = ErrFileNotFound } if err != nil { _ = b.deleteChunks(ctx, fileID) // Can attempt to delete chunks even if no docs in files collection matched. return err } return b.deleteChunks(ctx, fileID) } // Find returns the files collection documents that match the given filter. // // If this download requires a custom read deadline to be set on the bucket, it cannot be done concurrently with other // read operations operations on this bucket that also require a custom deadline. // // Use SetReadDeadline to set a deadline for the find operation. func (b Bucket) Find(filter interface{}, opts ...options.GridFSFindOptions) (mongo.Cursor, error) { ctx, cancel := deadlineContext(b.readDeadline) if cancel != nil { defer cancel() } return b.FindContext(ctx, filter, opts...) } // FindContext returns the files collection documents that match the given filter and runs the underlying // find query with the provided context. // // Use the context parameter to time-out or cancel the find operation. The deadline set by SetReadDeadline // is ignored. func (b Bucket) FindContext(ctx context.Context, filter interface{}, opts ...options.GridFSFindOptions) (mongo.Cursor, error) { gfsOpts := options.MergeGridFSFindOptions(opts...) find := options.Find() if gfsOpts.AllowDiskUse != nil { find.SetAllowDiskUse(gfsOpts.AllowDiskUse) } if gfsOpts.BatchSize != nil { find.SetBatchSize(gfsOpts.BatchSize) } if gfsOpts.Limit != nil { find.SetLimit(int64(gfsOpts.Limit)) } if gfsOpts.MaxTime != nil { find.SetMaxTime(gfsOpts.MaxTime) } if gfsOpts.NoCursorTimeout != nil { find.SetNoCursorTimeout(gfsOpts.NoCursorTimeout) } if gfsOpts.Skip != nil { find.SetSkip(int64(gfsOpts.Skip)) } if gfsOpts.Sort != nil { find.SetSort(gfsOpts.Sort) } return b.filesColl.Find(ctx, filter, find) } // Rename renames the stored file with the specified file ID. // // If this operation requires a custom write deadline to be set on the bucket, it cannot be done concurrently with other // write operations operations on this bucket that also require a custom deadline // // Use SetWriteDeadline to set a deadline for the rename operation. func (b Bucket) Rename(fileID interface{}, newFilename string) error { ctx, cancel := deadlineContext(b.writeDeadline) if cancel != nil { defer cancel() } return b.RenameContext(ctx, fileID, newFilename) } // RenameContext renames the stored file with the specified file ID and runs the underlying update with the provided // context. // // Use the context parameter to time-out or cancel the rename operation. The deadline set by SetWriteDeadline is ignored. func (b Bucket) RenameContext(ctx context.Context, fileID interface{}, newFilename string) error { res, err := b.filesColl.UpdateOne(ctx, bson.D{{"_id", fileID}}, bson.D{{"$set", bson.D{{"filename", newFilename}}}}, ) if err != nil { return err } if res.MatchedCount == 0 { return ErrFileNotFound } return nil } // Drop drops the files and chunks collections associated with this bucket. // // If this operation requires a custom write deadline to be set on the bucket, it cannot be done concurrently with other // write operations operations on this bucket that also require a custom deadline // // Use SetWriteDeadline to set a deadline for the drop operation. func (b Bucket) Drop() error { ctx, cancel := deadlineContext(b.writeDeadline) if cancel != nil { defer cancel() } return b.DropContext(ctx) } // DropContext drops the files and chunks collections associated with this bucket and runs the drop operations with // the provided context. // // Use the context parameter to time-out or cancel the drop operation. The deadline set by SetWriteDeadline is ignored. func (b Bucket) DropContext(ctx context.Context) error { // If no deadline is set on the passed-in context, Timeout is set on the Client, and context is // not already a Timeout context, honor Timeout in new Timeout context for operation execution to // be shared by both drop operations. if _, deadlineSet := ctx.Deadline(); !deadlineSet && b.db.Client().Timeout() != nil && !csot.IsTimeoutContext(ctx) { newCtx, cancelFunc := csot.MakeTimeoutContext(ctx, b.db.Client().Timeout()) // Redefine ctx to be the new timeout-derived context. ctx = newCtx // Cancel the timeout-derived context at the end of Execute to avoid a context leak. defer cancelFunc() } err := b.filesColl.Drop(ctx) if err != nil { return err } return b.chunksColl.Drop(ctx) } // GetFilesCollection returns a handle to the collection that stores the file documents for this bucket. func (b Bucket) GetFilesCollection() mongo.Collection { return b.filesColl } // GetChunksCollection returns a handle to the collection that stores the file chunks for this bucket. func (b Bucket) GetChunksCollection() mongo.Collection { return b.chunksColl } func (b Bucket) openDownloadStream(filter interface{}, opts ...options.FindOptions) (DownloadStream, error) { ctx, cancel := deadlineContext(b.readDeadline) if cancel != nil { defer cancel() } cursor, err := b.findFile(ctx, filter, opts...) if err != nil { return nil, err } // Unmarshal the data into a File instance, which can be passed to newDownloadStream. The _id value has to be // parsed out separately because "_id" will not match the File.ID field and we want to avoid exposing BSON tags // in the File type. After parsing it, use RawValue.Unmarshal to ensure File.ID is set to the appropriate value. var foundFile File if err = cursor.Decode(&foundFile); err != nil { return nil, fmt.Errorf("error decoding files collection document: %w", err) } if foundFile.Length == 0 { return newDownloadStream(nil, foundFile.ChunkSize, &foundFile), nil } // For a file with non-zero length, chunkSize must exist so we know what size to expect when downloading chunks. if _, err := cursor.Current.LookupErr("chunkSize"); err != nil { return nil, ErrMissingChunkSize } chunksCursor, err := b.findChunks(ctx, foundFile.ID) if err != nil { return nil, err } // The chunk size can be overridden for individual files, so the expected chunk size should be the "chunkSize" // field from the files collection document, not the bucket's chunk size. return newDownloadStream(chunksCursor, foundFile.ChunkSize, &foundFile), nil } func deadlineContext(deadline time.Time) (context.Context, context.CancelFunc) { if deadline.Equal(time.Time{}) { return context.Background(), nil } return context.WithDeadline(context.Background(), deadline) } func (b Bucket) downloadToStream(ds DownloadStream, stream io.Writer) (int64, error) { err := ds.SetReadDeadline(b.readDeadline) if err != nil { _ = ds.Close() return 0, err } copied, err := io.Copy(stream, ds) if err != nil { _ = ds.Close() return 0, err } return copied, ds.Close() } func (b Bucket) deleteChunks(ctx context.Context, fileID interface{}) error { _, err := b.chunksColl.DeleteMany(ctx, bson.D{{"files_id", fileID}}) return err } func (b Bucket) findFile(ctx context.Context, filter interface{}, opts ...options.FindOptions) (mongo.Cursor, error) { cursor, err := b.filesColl.Find(ctx, filter, opts...) if err != nil { return nil, err } if !cursor.Next(ctx) { _ = cursor.Close(ctx) return nil, ErrFileNotFound } return cursor, nil } func (b Bucket) findChunks(ctx context.Context, fileID interface{}) (mongo.Cursor, error) { chunksCursor, err := b.chunksColl.Find(ctx, bson.D{{"files_id", fileID}}, options.Find().SetSort(bson.D{{"n", 1}})) // sort by chunk index if err != nil { return nil, err } return chunksCursor, nil } // returns true if the 2 index documents are equal func numericalIndexDocsEqual(expected, actual bsoncore.Document) (bool, error) { if bytes.Equal(expected, actual) { return true, nil } actualElems, err := actual.Elements() if err != nil { return false, err } expectedElems, err := expected.Elements() if err != nil { return false, err } if len(actualElems) != len(expectedElems) { return false, nil } for idx, expectedElem := range expectedElems { actualElem := actualElems[idx] if actualElem.Key() != expectedElem.Key() { return false, nil } actualVal := actualElem.Value() expectedVal := expectedElem.Value() actualInt, actualOK := actualVal.AsInt64OK() expectedInt, expectedOK := expectedVal.AsInt64OK() //GridFS indexes always have numeric values if !actualOK \|\| !expectedOK { return false, nil } if actualInt != expectedInt { return false, nil } } return true, nil } // Create an index if it doesn't already exist func createNumericalIndexIfNotExists(ctx context.Context, iv mongo.IndexView, model mongo.IndexModel) error { c, err := iv.List(ctx) if err != nil { return err } defer func() { _ = c.Close(ctx) }() modelKeysBytes, err := bson.Marshal(model.Keys) if err != nil { return err } modelKeysDoc := bsoncore.Document(modelKeysBytes) for c.Next(ctx) { keyElem, err := c.Current.LookupErr("key") if err != nil { return err } keyElemDoc := keyElem.Document() found, err := numericalIndexDocsEqual(modelKeysDoc, bsoncore.Document(keyElemDoc)) if err != nil { return err } if found { return nil } } _, err = iv.CreateOne(ctx, model) return err } // create indexes on the files and chunks collection if needed func (b Bucket) createIndexes(ctx context.Context) error { // must use primary read pref mode to check if files coll empty cloned, err := b.filesColl.Clone(options.Collection().SetReadPreference(readpref.Primary())) if err != nil { return err } docRes := cloned.FindOne(ctx, bson.D{}, options.FindOne().SetProjection(bson.D{{"_id", 1}})) _, err = docRes.Raw() if !errors.Is(err, mongo.ErrNoDocuments) { // nil, or error that occurred during the FindOne operation return err } filesIv := b.filesColl.Indexes() chunksIv := b.chunksColl.Indexes() filesModel := mongo.IndexModel{ Keys: bson.D{ {"filename", int32(1)}, {"uploadDate", int32(1)}, }, } chunksModel := mongo.IndexModel{ Keys: bson.D{ {"files_id", int32(1)}, {"n", int32(1)}, }, Options: options.Index().SetUnique(true), } if err = createNumericalIndexIfNotExists(ctx, filesIv, filesModel); err != nil { return err } return createNumericalIndexIfNotExists(ctx, chunksIv, chunksModel) } func (b Bucket) checkFirstWrite(ctx context.Context) error { if !b.firstWriteDone { // before the first write operation, must determine if files collection is empty // if so, create indexes if they do not already exist if err := b.createIndexes(ctx); err != nil { return err } b.firstWriteDone = true } return nil } func (b Bucket) parseUploadOptions(opts ...options.UploadOptions) (Upload, error) { upload := &Upload{ chunkSize: b.chunkSize, // upload chunk size defaults to bucket's value } uo := options.MergeUploadOptions(opts...) if uo.ChunkSizeBytes != nil { upload.chunkSize = *uo.ChunkSizeBytes } if uo.Registry == nil { uo.Registry = bson.DefaultRegistry } if uo.Metadata != nil { // TODO(GODRIVER-2726): Replace with marshal() and unmarshal() once the // TODO gridfs package is merged into the mongo package. raw, err := bson.MarshalWithRegistry(uo.Registry, uo.Metadata) if err != nil { return nil, err } var doc bson.D unMarErr := bson.UnmarshalWithRegistry(uo.Registry, raw, &doc) if unMarErr != nil { return nil, unMarErr } upload.metadata = doc } return upload, nil }	20.32 KB	2025-05-25 12:27
📄 doc.go // Copyright (C) MongoDB, Inc. 2017-present. // // Licensed under the Apache License, Version 2.0 (the "License"); you may // not use this file except in compliance with the License. You may obtain // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 // Package gridfs provides a MongoDB GridFS API. See https://www.mongodb.com/docs/manual/core/gridfs/ for more // information about GridFS and its use cases. // // # Buckets // // The main type defined in this package is Bucket. A Bucket wraps a mongo.Database instance and operates on two // collections in the database. The first is the files collection, which contains one metadata document per file stored // in the bucket. This collection is named "<bucket name>.files". The second is the chunks collection, which contains // chunks of files. This collection is named "<bucket name>.chunks". // // # Uploading a File // // Files can be uploaded in two ways: // // 1. OpenUploadStream/OpenUploadStreamWithID - These methods return an UploadStream instance. UploadStream // implements the io.Writer interface and the Write() method can be used to upload a file to the database. // // 2. UploadFromStream/UploadFromStreamWithID - These methods take an io.Reader, which represents the file to // upload. They internally create a new UploadStream and close it once the operation is complete. // // # Downloading a File // // Similar to uploads, files can be downloaded in two ways: // // 1. OpenDownloadStream/OpenDownloadStreamByName - These methods return a DownloadStream instance. DownloadStream // implements the io.Reader interface. A file can be read either using the Read() method or any standard library // methods that reads from an io.Reader such as io.Copy. // // 2. DownloadToStream/DownloadToStreamByName - These methods take an io.Writer, which represents the download // destination. They internally create a new DownloadStream and close it once the operation is complete. package gridfs	1.95 KB	2025-05-25 12:27
📄 download_stream.go // Copyright (C) MongoDB, Inc. 2017-present. // // Licensed under the Apache License, Version 2.0 (the "License"); you may // not use this file except in compliance with the License. You may obtain // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 package gridfs import ( "context" "errors" "io" "math" "time" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" ) // ErrWrongIndex is used when the chunk retrieved from the server does not have the expected index. var ErrWrongIndex = errors.New("chunk index does not match expected index") // ErrWrongSize is used when the chunk retrieved from the server does not have the expected size. var ErrWrongSize = errors.New("chunk size does not match expected size") var errNoMoreChunks = errors.New("no more chunks remaining") // DownloadStream is a io.Reader that can be used to download a file from a GridFS bucket. type DownloadStream struct { numChunks int32 chunkSize int32 cursor mongo.Cursor done bool closed bool buffer []byte // store up to 1 chunk if the user provided buffer isn't big enough bufferStart int bufferEnd int expectedChunk int32 // index of next expected chunk readDeadline time.Time fileLen int64 // The pointer returned by GetFile. This should not be used in the actual DownloadStream code outside of the // newDownloadStream constructor because the values can be mutated by the user after calling GetFile. Instead, // any values needed in the code should be stored separately and copied over in the constructor. file File } // File represents a file stored in GridFS. This type can be used to access file information when downloading using the // DownloadStream.GetFile method. type File struct { // ID is the file's ID. This will match the file ID specified when uploading the file. If an upload helper that // does not require a file ID was used, this field will be a primitive.ObjectID. ID interface{} // Length is the length of this file in bytes. Length int64 // ChunkSize is the maximum number of bytes for each chunk in this file. ChunkSize int32 // UploadDate is the time this file was added to GridFS in UTC. This field is set by the driver and is not configurable. // The Metadata field can be used to store a custom date. UploadDate time.Time // Name is the name of this file. Name string // Metadata is additional data that was specified when creating this file. This field can be unmarshalled into a // custom type using the bson.Unmarshal family of functions. Metadata bson.Raw } var _ bson.Unmarshaler = (File)(nil) // unmarshalFile is a temporary type used to unmarshal documents from the files collection and can be transformed into // a File instance. This type exists to avoid adding BSON struct tags to the exported File type. type unmarshalFile struct { ID interface{} `bson:"_id"` Length int64 `bson:"length"` ChunkSize int32 `bson:"chunkSize"` UploadDate time.Time `bson:"uploadDate"` Name string `bson:"filename"` Metadata bson.Raw `bson:"metadata"` } // UnmarshalBSON implements the bson.Unmarshaler interface. // // Deprecated: Unmarshaling a File from BSON will not be supported in Go Driver 2.0. func (f File) UnmarshalBSON(data []byte) error { var temp unmarshalFile if err := bson.Unmarshal(data, &temp); err != nil { return err } f.ID = temp.ID f.Length = temp.Length f.ChunkSize = temp.ChunkSize f.UploadDate = temp.UploadDate f.Name = temp.Name f.Metadata = temp.Metadata return nil } func newDownloadStream(cursor mongo.Cursor, chunkSize int32, file File) DownloadStream { numChunks := int32(math.Ceil(float64(file.Length) / float64(chunkSize))) return &DownloadStream{ numChunks: numChunks, chunkSize: chunkSize, cursor: cursor, buffer: make([]byte, chunkSize), done: cursor == nil, fileLen: file.Length, file: file, } } // Close closes this download stream. func (ds DownloadStream) Close() error { if ds.closed { return ErrStreamClosed } ds.closed = true if ds.cursor != nil { return ds.cursor.Close(context.Background()) } return nil } // SetReadDeadline sets the read deadline for this download stream. func (ds DownloadStream) SetReadDeadline(t time.Time) error { if ds.closed { return ErrStreamClosed } ds.readDeadline = t return nil } // Read reads the file from the server and writes it to a destination byte slice. func (ds DownloadStream) Read(p []byte) (int, error) { if ds.closed { return 0, ErrStreamClosed } if ds.done { return 0, io.EOF } ctx, cancel := deadlineContext(ds.readDeadline) if cancel != nil { defer cancel() } bytesCopied := 0 var err error for bytesCopied < len(p) { if ds.bufferStart >= ds.bufferEnd { // Buffer is empty and can load in data from new chunk. err = ds.fillBuffer(ctx) if err != nil { if errors.Is(err, errNoMoreChunks) { if bytesCopied == 0 { ds.done = true return 0, io.EOF } return bytesCopied, nil } return bytesCopied, err } } copied := copy(p[bytesCopied:], ds.buffer[ds.bufferStart:ds.bufferEnd]) bytesCopied += copied ds.bufferStart += copied } return len(p), nil } // Skip skips a given number of bytes in the file. func (ds DownloadStream) Skip(skip int64) (int64, error) { if ds.closed { return 0, ErrStreamClosed } if ds.done { return 0, nil } ctx, cancel := deadlineContext(ds.readDeadline) if cancel != nil { defer cancel() } var skipped int64 var err error for skipped < skip { if ds.bufferStart >= ds.bufferEnd { // Buffer is empty and can load in data from new chunk. err = ds.fillBuffer(ctx) if err != nil { if errors.Is(err, errNoMoreChunks) { return skipped, nil } return skipped, err } } toSkip := skip - skipped // Cap the amount to skip to the remaining bytes in the buffer to be consumed. bufferRemaining := ds.bufferEnd - ds.bufferStart if toSkip > int64(bufferRemaining) { toSkip = int64(bufferRemaining) } skipped += toSkip ds.bufferStart += int(toSkip) } return skip, nil } // GetFile returns a File object representing the file being downloaded. func (ds DownloadStream) GetFile() File { return ds.file } func (ds DownloadStream) fillBuffer(ctx context.Context) error { if !ds.cursor.Next(ctx) { ds.done = true // Check for cursor error, otherwise there are no more chunks. if ds.cursor.Err() != nil { _ = ds.cursor.Close(ctx) return ds.cursor.Err() } // If there are no more chunks, but we didn't read the expected number of chunks, return an // ErrWrongIndex error to indicate that we're missing chunks at the end of the file. if ds.expectedChunk != ds.numChunks { return ErrWrongIndex } return errNoMoreChunks } chunkIndex, err := ds.cursor.Current.LookupErr("n") if err != nil { return err } var chunkIndexInt32 int32 if chunkIndexInt64, ok := chunkIndex.Int64OK(); ok { chunkIndexInt32 = int32(chunkIndexInt64) } else { chunkIndexInt32 = chunkIndex.Int32() } if chunkIndexInt32 != ds.expectedChunk { return ErrWrongIndex } ds.expectedChunk++ data, err := ds.cursor.Current.LookupErr("data") if err != nil { return err } _, dataBytes := data.Binary() copied := copy(ds.buffer, dataBytes) bytesLen := int32(len(dataBytes)) if ds.expectedChunk == ds.numChunks { // final chunk can be fewer than ds.chunkSize bytes bytesDownloaded := int64(ds.chunkSize) * (int64(ds.expectedChunk) - int64(1)) bytesRemaining := ds.fileLen - bytesDownloaded if int64(bytesLen) != bytesRemaining { return ErrWrongSize } } else if bytesLen != ds.chunkSize { // all intermediate chunks must have size ds.chunkSize return ErrWrongSize } ds.bufferStart = 0 ds.bufferEnd = copied return nil }	7.7 KB	2025-05-25 12:27
📄 gridfs_examples_test.go // Copyright (C) MongoDB, Inc. 2017-present. // // Licensed under the Apache License, Version 2.0 (the "License"); you may // not use this file except in compliance with the License. You may obtain // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 package gridfs_test import ( "bytes" "context" "fmt" "io" "log" "time" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo/gridfs" "go.mongodb.org/mongo-driver/mongo/options" ) func ExampleBucket_OpenUploadStream() { var fileContent []byte var bucket gridfs.Bucket // Specify the Metadata option to include a "metadata" field in the files // collection document. uploadOpts := options.GridFSUpload(). SetMetadata(bson.D{{"metadata tag", "tag"}}) uploadStream, err := bucket.OpenUploadStream("filename", uploadOpts) if err != nil { log.Fatal(err) } defer func() { if err = uploadStream.Close(); err != nil { log.Fatal(err) } }() // Use SetWriteDeadline to force a timeout if the upload does not succeed in // 2 seconds. err = uploadStream.SetWriteDeadline(time.Now().Add(2 time.Second)) if err != nil { log.Fatal(err) } if _, err = uploadStream.Write(fileContent); err != nil { log.Fatal(err) } } func ExampleBucket_UploadFromStream() { var fileContent []byte var bucket gridfs.Bucket // Specify the Metadata option to include a "metadata" field in the files // collection document. uploadOpts := options.GridFSUpload(). SetMetadata(bson.D{{"metadata tag", "tag"}}) fileID, err := bucket.UploadFromStream( "filename", bytes.NewBuffer(fileContent), uploadOpts) if err != nil { log.Fatal(err) } fmt.Printf("new file created with ID %s", fileID) } func ExampleBucket_OpenDownloadStream() { var bucket gridfs.Bucket var fileID primitive.ObjectID downloadStream, err := bucket.OpenDownloadStream(fileID) if err != nil { log.Fatal(err) } defer func() { if err := downloadStream.Close(); err != nil { log.Fatal(err) } }() // Use SetReadDeadline to force a timeout if the download does not succeed // in 2 seconds. err = downloadStream.SetReadDeadline(time.Now().Add(2 * time.Second)) if err != nil { log.Fatal(err) } fileBuffer := bytes.NewBuffer(nil) if _, err := io.Copy(fileBuffer, downloadStream); err != nil { log.Fatal(err) } } func ExampleBucket_DownloadToStream() { var bucket gridfs.Bucket var fileID primitive.ObjectID fileBuffer := bytes.NewBuffer(nil) if _, err := bucket.DownloadToStream(fileID, fileBuffer); err != nil { log.Fatal(err) } } func ExampleBucket_Delete() { var bucket gridfs.Bucket var fileID primitive.ObjectID if err := bucket.Delete(fileID); err != nil { log.Fatal(err) } } func ExampleBucket_Find() { var bucket gridfs.Bucket // Specify a filter to find all files with a length greater than 1000 bytes. filter := bson.D{ {"length", bson.D{{"$gt", 1000}}}, } cursor, err := bucket.Find(filter) if err != nil { log.Fatal(err) } defer func() { if err := cursor.Close(context.TODO()); err != nil { log.Fatal(err) } }() type gridfsFile struct { Name string `bson:"filename"` Length int64 `bson:"length"` } var foundFiles []gridfsFile if err = cursor.All(context.TODO(), &foundFiles); err != nil { log.Fatal(err) } for _, file := range foundFiles { fmt.Printf("filename: %s, length: %d\n", file.Name, file.Length) } } func ExampleBucket_Rename() { var bucket gridfs.Bucket var fileID primitive.ObjectID if err := bucket.Rename(fileID, "new file name"); err != nil { log.Fatal(err) } } func ExampleBucket_Drop() { var bucket *gridfs.Bucket if err := bucket.Drop(); err != nil { log.Fatal(err) } }	3.64 KB	2025-05-25 12:27
📄 gridfs_test.go // Copyright (C) MongoDB, Inc. 2017-present. // // Licensed under the Apache License, Version 2.0 (the "License"); you may // not use this file except in compliance with the License. You may obtain // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 package gridfs import ( "context" "testing" "go.mongodb.org/mongo-driver/event" "go.mongodb.org/mongo-driver/internal/assert" "go.mongodb.org/mongo-driver/internal/integtest" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" "go.mongodb.org/mongo-driver/mongo/readpref" "go.mongodb.org/mongo-driver/mongo/writeconcern" ) var ( connsCheckedOut int ) func TestGridFS(t testing.T) { if testing.Short() { t.Skip("skipping integration test in short mode") } cs := integtest.ConnString(t) poolMonitor := &event.PoolMonitor{ Event: func(evt event.PoolEvent) { switch evt.Type { case event.GetSucceeded: connsCheckedOut++ case event.ConnectionReturned: connsCheckedOut-- } }, } clientOpts := options.Client(). ApplyURI(cs.Original). SetReadPreference(readpref.Primary()). SetWriteConcern(writeconcern.New(writeconcern.WMajority())). SetPoolMonitor(poolMonitor). // Connect to a single host. For sharded clusters, this will pin to a single mongos, which avoids // non-deterministic versioning errors in the server. This has no effect for replica sets because the driver // will discover the other hosts during SDAM checks. SetHosts(cs.Hosts[:1]) client, err := mongo.Connect(context.Background(), clientOpts) assert.Nil(t, err, "Connect error: %v", err) db := client.Database("gridfs") defer func() { sessions := client.NumberSessionsInProgress() conns := connsCheckedOut _ = db.Drop(context.Background()) _ = client.Disconnect(context.Background()) assert.Equal(t, 0, sessions, "%v sessions checked out", sessions) assert.Equal(t, 0, conns, "%v connections checked out", conns) }() // Unit tests showing the chunk size is set correctly on the bucket and upload stream objects. t.Run("ChunkSize", func(t testing.T) { chunkSizeTests := []struct { testName string bucketOpts options.BucketOptions uploadOpts options.UploadOptions }{ {"Default values", nil, nil}, {"Options provided without chunk size", options.GridFSBucket(), options.GridFSUpload()}, {"Bucket chunk size set", options.GridFSBucket().SetChunkSizeBytes(27), nil}, {"Upload stream chunk size set", nil, options.GridFSUpload().SetChunkSizeBytes(27)}, {"Bucket and upload set to different values", options.GridFSBucket().SetChunkSizeBytes(27), options.GridFSUpload().SetChunkSizeBytes(31)}, } for _, tt := range chunkSizeTests { t.Run(tt.testName, func(t testing.T) { bucket, err := NewBucket(db, tt.bucketOpts) assert.Nil(t, err, "NewBucket error: %v", err) us, err := bucket.OpenUploadStream("filename", tt.uploadOpts) assert.Nil(t, err, "OpenUploadStream error: %v", err) expectedBucketChunkSize := DefaultChunkSize if tt.bucketOpts != nil && tt.bucketOpts.ChunkSizeBytes != nil { expectedBucketChunkSize = tt.bucketOpts.ChunkSizeBytes } assert.Equal(t, expectedBucketChunkSize, bucket.chunkSize, "expected chunk size %v, got %v", expectedBucketChunkSize, bucket.chunkSize) expectedUploadChunkSize := expectedBucketChunkSize if tt.uploadOpts != nil && tt.uploadOpts.ChunkSizeBytes != nil { expectedUploadChunkSize = tt.uploadOpts.ChunkSizeBytes } assert.Equal(t, expectedUploadChunkSize, us.chunkSize, "expected chunk size %v, got %v", expectedUploadChunkSize, us.chunkSize) }) } }) }	3.56 KB	2025-05-25 12:27
📄 upload_stream.go // Copyright (C) MongoDB, Inc. 2017-present. // // Licensed under the Apache License, Version 2.0 (the "License"); you may // not use this file except in compliance with the License. You may obtain // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 package gridfs import ( "errors" "context" "time" "math" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" ) // UploadBufferSize is the size in bytes of one stream batch. Chunks will be written to the db after the sum of chunk // lengths is equal to the batch size. const UploadBufferSize = 16 * 1024 * 1024 // 16 MiB // ErrStreamClosed is an error returned if an operation is attempted on a closed/aborted stream. var ErrStreamClosed = errors.New("stream is closed or aborted") // UploadStream is used to upload a file in chunks. This type implements the io.Writer interface and a file can be // uploaded using the Write method. After an upload is complete, the Close method must be called to write file // metadata. type UploadStream struct { Upload // chunk size and metadata FileID interface{} chunkIndex int chunksColl mongo.Collection // collection to store file chunks filename string filesColl mongo.Collection // collection to store file metadata closed bool buffer []byte bufferIndex int fileLen int64 writeDeadline time.Time } // NewUploadStream creates a new upload stream. func newUploadStream(upload Upload, fileID interface{}, filename string, chunks, files mongo.Collection) UploadStream { return &UploadStream{ Upload: upload, FileID: fileID, chunksColl: chunks, filename: filename, filesColl: files, buffer: make([]byte, UploadBufferSize), } } // Close writes file metadata to the files collection and cleans up any resources associated with the UploadStream. func (us UploadStream) Close() error { if us.closed { return ErrStreamClosed } ctx, cancel := deadlineContext(us.writeDeadline) if cancel != nil { defer cancel() } if us.bufferIndex != 0 { if err := us.uploadChunks(ctx, true); err != nil { return err } } if err := us.createFilesCollDoc(ctx); err != nil { return err } us.closed = true return nil } // SetWriteDeadline sets the write deadline for this stream. func (us UploadStream) SetWriteDeadline(t time.Time) error { if us.closed { return ErrStreamClosed } us.writeDeadline = t return nil } // Write transfers the contents of a byte slice into this upload stream. If the stream's underlying buffer fills up, // the buffer will be uploaded as chunks to the server. Implements the io.Writer interface. func (us UploadStream) Write(p []byte) (int, error) { if us.closed { return 0, ErrStreamClosed } var ctx context.Context ctx, cancel := deadlineContext(us.writeDeadline) if cancel != nil { defer cancel() } origLen := len(p) for { if len(p) == 0 { break } n := copy(us.buffer[us.bufferIndex:], p) // copy as much as possible p = p[n:] us.bufferIndex += n if us.bufferIndex == UploadBufferSize { err := us.uploadChunks(ctx, false) if err != nil { return 0, err } } } return origLen, nil } // Abort closes the stream and deletes all file chunks that have already been written. func (us UploadStream) Abort() error { if us.closed { return ErrStreamClosed } ctx, cancel := deadlineContext(us.writeDeadline) if cancel != nil { defer cancel() } _, err := us.chunksColl.DeleteMany(ctx, bson.D{{"files_id", us.FileID}}) if err != nil { return err } us.closed = true return nil } // uploadChunks uploads the current buffer as a series of chunks to the bucket // if uploadPartial is true, any data at the end of the buffer that is smaller than a chunk will be uploaded as a partial // chunk. if it is false, the data will be moved to the front of the buffer. // uploadChunks sets us.bufferIndex to the next available index in the buffer after uploading func (us UploadStream) uploadChunks(ctx context.Context, uploadPartial bool) error { chunks := float64(us.bufferIndex) / float64(us.chunkSize) numChunks := int(math.Ceil(chunks)) if !uploadPartial { numChunks = int(math.Floor(chunks)) } docs := make([]interface{}, numChunks) begChunkIndex := us.chunkIndex for i := 0; i < us.bufferIndex; i += int(us.chunkSize) { endIndex := i + int(us.chunkSize) if us.bufferIndex-i < int(us.chunkSize) { // partial chunk if !uploadPartial { break } endIndex = us.bufferIndex } chunkData := us.buffer[i:endIndex] docs[us.chunkIndex-begChunkIndex] = bson.D{ {"_id", primitive.NewObjectID()}, {"files_id", us.FileID}, {"n", int32(us.chunkIndex)}, {"data", primitive.Binary{Subtype: 0x00, Data: chunkData}}, } us.chunkIndex++ us.fileLen += int64(len(chunkData)) } _, err := us.chunksColl.InsertMany(ctx, docs) if err != nil { return err } // copy any remaining bytes to beginning of buffer and set buffer index bytesUploaded := numChunks int(us.chunkSize) if bytesUploaded != UploadBufferSize && !uploadPartial { copy(us.buffer[0:], us.buffer[bytesUploaded:us.bufferIndex]) } us.bufferIndex = UploadBufferSize - bytesUploaded return nil } func (us *UploadStream) createFilesCollDoc(ctx context.Context) error { doc := bson.D{ {"_id", us.FileID}, {"length", us.fileLen}, {"chunkSize", us.chunkSize}, {"uploadDate", primitive.DateTime(time.Now().UnixNano() / int64(time.Millisecond))}, {"filename", us.filename}, } if us.metadata != nil { doc = append(doc, bson.E{"metadata", us.metadata}) } _, err := us.filesColl.InsertOne(ctx, doc) if err != nil { return err } return nil }	5.59 KB	2025-05-25 12:27

Terminal