/* * Minio Cloud Storage, (C) 2016 Minio, Inc. * * 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 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package main import ( "io" "path/filepath" "strings" "github.com/minio/minio/pkg/mimedb" ) // fsObjects - Implements fs object layer. type fsObjects struct { storage StorageAPI } // newFSObjects - initialize new fs object layer. func newFSObjects(exportPath string) (ObjectLayer, error) { var storage StorageAPI var err error if !strings.ContainsRune(exportPath, ':') || filepath.VolumeName(exportPath) != "" { // Initialize filesystem storage API. storage, err = newPosix(exportPath) if err != nil { return nil, err } } else { // Initialize rpc client storage API. storage, err = newRPCClient(exportPath) if err != nil { return nil, err } } return fsObjects{storage}, nil } /// Bucket operations // MakeBucket - make a bucket. func (fs fsObjects) MakeBucket(bucket string) error { return makeBucket(fs.storage, bucket) } // GetBucketInfo - get bucket info. func (fs fsObjects) GetBucketInfo(bucket string) (BucketInfo, error) { return getBucketInfo(fs.storage, bucket) } // ListBuckets - list buckets. func (fs fsObjects) ListBuckets() ([]BucketInfo, error) { return listBuckets(fs.storage) } // DeleteBucket - delete a bucket. func (fs fsObjects) DeleteBucket(bucket string) error { return deleteBucket(fs.storage, bucket) } /// Object Operations // GetObject - get an object. func (fs fsObjects) GetObject(bucket, object string, startOffset int64) (io.ReadCloser, error) { // Verify if bucket is valid. if !IsValidBucketName(bucket) { return nil, (BucketNameInvalid{Bucket: bucket}) } // Verify if object is valid. if !IsValidObjectName(object) { return nil, (ObjectNameInvalid{Bucket: bucket, Object: object}) } fileReader, err := fs.storage.ReadFile(bucket, object, startOffset) if err != nil { return nil, toObjectErr(err, bucket, object) } return fileReader, nil } // GetObjectInfo - get object info. func (fs fsObjects) GetObjectInfo(bucket, object string) (ObjectInfo, error) { // Verify if bucket is valid. if !IsValidBucketName(bucket) { return ObjectInfo{}, (BucketNameInvalid{Bucket: bucket}) } // Verify if object is valid. if !IsValidObjectName(object) { return ObjectInfo{}, (ObjectNameInvalid{Bucket: bucket, Object: object}) } fi, err := fs.storage.StatFile(bucket, object) if err != nil { return ObjectInfo{}, toObjectErr(err, bucket, object) } contentType := "application/octet-stream" if objectExt := filepath.Ext(object); objectExt != "" { content, ok := mimedb.DB[strings.ToLower(strings.TrimPrefix(objectExt, "."))] if ok { contentType = content.ContentType } } return ObjectInfo{ Bucket: bucket, Name: object, ModTime: fi.ModTime, Size: fi.Size, IsDir: fi.Mode.IsDir(), ContentType: contentType, MD5Sum: "", // Read from metadata. }, nil } // PutObject - create an object. func (fs fsObjects) PutObject(bucket string, object string, size int64, data io.Reader, metadata map[string]string) (string, error) { return putObjectCommon(fs.storage, bucket, object, size, data, metadata) } func (fs fsObjects) DeleteObject(bucket, object string) error { // Verify if bucket is valid. if !IsValidBucketName(bucket) { return BucketNameInvalid{Bucket: bucket} } if !IsValidObjectName(object) { return ObjectNameInvalid{Bucket: bucket, Object: object} } if err := fs.storage.DeleteFile(bucket, object); err != nil { return toObjectErr(err, bucket, object) } return nil } func (fs fsObjects) ListObjects(bucket, prefix, marker, delimiter string, maxKeys int) (ListObjectsInfo, error) { // Verify if bucket is valid. if !IsValidBucketName(bucket) { return ListObjectsInfo{}, BucketNameInvalid{Bucket: bucket} } if !IsValidObjectPrefix(prefix) { return ListObjectsInfo{}, ObjectNameInvalid{Bucket: bucket, Object: prefix} } // Verify if delimiter is anything other than '/', which we do not support. if delimiter != "" && delimiter != slashSeparator { return ListObjectsInfo{}, UnsupportedDelimiter{ Delimiter: delimiter, } } // Verify if marker has prefix. if marker != "" { if !strings.HasPrefix(marker, prefix) { return ListObjectsInfo{}, InvalidMarkerPrefixCombination{ Marker: marker, Prefix: prefix, } } } // Default is recursive, if delimiter is set then list non recursive. recursive := true if delimiter == slashSeparator { recursive = false } fileInfos, eof, err := fs.storage.ListFiles(bucket, prefix, marker, recursive, maxKeys) if err != nil { return ListObjectsInfo{}, toObjectErr(err, bucket) } if maxKeys == 0 { return ListObjectsInfo{}, nil } result := ListObjectsInfo{IsTruncated: !eof} for _, fileInfo := range fileInfos { // With delimiter set we fill in NextMarker and Prefixes. if delimiter == slashSeparator { result.NextMarker = fileInfo.Name if fileInfo.Mode.IsDir() { result.Prefixes = append(result.Prefixes, fileInfo.Name) continue } } result.Objects = append(result.Objects, ObjectInfo{ Name: fileInfo.Name, ModTime: fileInfo.ModTime, Size: fileInfo.Size, IsDir: false, }) } return result, nil }