Merge pull request #280 from harshavardhana/pr_out_http_range_support_wip

Get object range request added

pkg/api/minioapi/api_test.go

@@ -364,16 +364,17 @@ func (s *MySuite) TestDateFormat(c *C) {
 
 func verifyHeaders(c *C, header http.Header, date time.Time, size int, contentType string, etag string) {
 	// Verify date
-	c.Assert(header["Last-Modified"][0], Equals, date.Format(time.RFC1123))
+	c.Log(header)
+	c.Assert(header.Get("Last-Modified"), Equals, date.Format(time.RFC1123))
 
 	// verify size
-	c.Assert(header["Content-Length"][0], Equals, strconv.Itoa(size))
+	c.Assert(header.Get("Content-Length"), Equals, strconv.Itoa(size))
 
 	// verify content type
-	c.Assert(header["Content-Type"][0], Equals, contentType)
+	c.Assert(header.Get("Content-Type"), Equals, contentType)
 
 	// verify etag
-	c.Assert(header["Etag"][0], Equals, etag)
+	c.Assert(header.Get("Etag"), Equals, etag)
 }
 
 func (s *MySuite) TestXMLNameNotInBucketListJson(c *C) {
@@ -448,11 +449,11 @@ func (s *MySuite) TestContentTypePersists(c *C) {
 	c.Assert(err, IsNil)
 	response, err = client.Do(request)
 	c.Assert(err, IsNil)
-	c.Assert(response.Header["Content-Type"][0], Equals, "application/octet-stream")
+	c.Assert(response.Header.Get("Content-Type"), Equals, "application/octet-stream")
 
 	// test get object
 	response, err = http.Get(testServer.URL + "/bucket/one")
-	c.Assert(response.Header["Content-Type"][0], Equals, "application/octet-stream")
+	c.Assert(response.Header.Get("Content-Type"), Equals, "application/octet-stream")
 
 	request, err = http.NewRequest("PUT", testServer.URL+"/bucket/two", bytes.NewBufferString("hello world"))
 	delete(request.Header, "Content-Type")
@@ -466,9 +467,9 @@ func (s *MySuite) TestContentTypePersists(c *C) {
 	c.Assert(err, IsNil)
 	response, err = client.Do(request)
 	c.Assert(err, IsNil)
-	c.Assert(response.Header["Content-Type"][0], Equals, "application/octet-stream")
+	c.Assert(response.Header.Get("Content-Type"), Equals, "application/octet-stream")
 
 	// test get object
 	response, err = http.Get(testServer.URL + "/bucket/two")
-	c.Assert(response.Header["Content-Type"][0], Equals, "application/octet-stream")
+	c.Assert(response.Header.Get("Content-Type"), Equals, "application/octet-stream")
 }

pkg/api/minioapi/headers.go

@@ -37,6 +37,7 @@ type encoder interface {
 // Write http common headers
 func writeCommonHeaders(w http.ResponseWriter, acceptsType string) {
 	w.Header().Set("Server", "Minio")
+	w.Header().Set("Accept-Ranges", "bytes")
 	w.Header().Set("Content-Type", acceptsType)
 }
 
@@ -67,6 +68,17 @@ func writeObjectHeaders(w http.ResponseWriter, metadata mstorage.ObjectMetadata)
 	w.Header().Set("Connection", "close")
 }
 
+// Write range object header
+func writeRangeObjectHeaders(w http.ResponseWriter, metadata mstorage.ObjectMetadata, ra string) {
+	lastModified := metadata.Created.Format(time.RFC1123)
+	// common headers
+	writeCommonHeaders(w, metadata.ContentType)
+	w.Header().Set("ETag", metadata.ETag)
+	w.Header().Set("Last-Modified", lastModified)
+	w.Header().Set("Content-Range", ra)
+	w.Header().Set("Content-Length", strconv.FormatInt(metadata.Size, 10))
+}
+
 // Write object header and response
 func writeObjectHeadersAndResponse(w http.ResponseWriter, response interface{}, acceptsType contentType) []byte {
 	var bytesBuffer bytes.Buffer

pkg/api/minioapi/object_handlers.go

@@ -30,7 +30,6 @@ import (
 // you must have READ access to the object.
 func (server *minioAPI) getObjectHandler(w http.ResponseWriter, req *http.Request) {
 	var object, bucket string
-
 	acceptsContentType := getContentType(req)
 	vars := mux.Vars(req)
 	bucket = vars["bucket"]
@@ -41,9 +40,40 @@ func (server *minioAPI) getObjectHandler(w http.ResponseWriter, req *http.Reques
 	case nil: // success
 		{
 			log.Println("Found: " + bucket + "#" + object)
+			httpRange, err := newRange(req, metadata.Size)
+			if err != nil {
+				log.Println(err)
+				error := errorCodeError(InvalidRange)
+				errorResponse := getErrorResponse(error, "/"+bucket+"/"+object)
+				w.WriteHeader(error.HTTPStatusCode)
+				w.Write(writeErrorResponse(w, errorResponse, acceptsContentType))
+				return
+			}
+			switch httpRange.start == 0 && httpRange.length == 0 {
+			case true:
 				writeObjectHeaders(w, metadata)
 				if _, err := server.storage.CopyObjectToWriter(w, bucket, object); err != nil {
 					log.Println(err)
+					error := errorCodeError(InternalError)
+					errorResponse := getErrorResponse(error, "/"+bucket+"/"+object)
+					w.WriteHeader(error.HTTPStatusCode)
+					w.Write(writeErrorResponse(w, errorResponse, acceptsContentType))
+					return
+				}
+			case false:
+				metadata.Size = httpRange.length
+				writeRangeObjectHeaders(w, metadata, httpRange.getContentRange())
+				w.WriteHeader(http.StatusPartialContent)
+				_, err := server.storage.CopyObjectToWriterRange(w, bucket, object, httpRange.start, httpRange.length)
+				if err != nil {
+					log.Println(err)
+					error := errorCodeError(InternalError)
+					errorResponse := getErrorResponse(error, "/"+bucket+"/"+object)
+					w.WriteHeader(error.HTTPStatusCode)
+					w.Write(writeErrorResponse(w, errorResponse, acceptsContentType))
+					return
+				}
+
 			}
 		}
 	case mstorage.ObjectNotFound:

pkg/api/minioapi/range.go

@@ -0,0 +1,118 @@
+/*
+ * Mini Object Storage, (C) 2015 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 minioapi
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"net/http"
+)
+
+const (
+	b = "bytes="
+)
+
+// HttpRange specifies the byte range to be sent to the client.
+type httpRange struct {
+	start, length, size int64
+}
+
+// GetContentRange populate range header
+func (r *httpRange) getContentRange() string {
+	return fmt.Sprintf("bytes %d-%d/%d", r.start, r.start+r.length-1, r.size)
+}
+
+// Grab new range from request header
+func newRange(req *http.Request, size int64) (*httpRange, error) {
+	r := &httpRange{
+		start:  0,
+		length: 0,
+		size:   0,
+	}
+	r.size = size
+	if s := req.Header.Get("Range"); s != "" {
+		err := r.parseRange(s)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return r, nil
+}
+
+// parseRange parses a Range header string as per RFC 2616.
+func (r *httpRange) parseRange(s string) error {
+	if s == "" {
+		return fmt.Errorf("header not present")
+	}
+	if !strings.HasPrefix(s, b) {
+		return fmt.Errorf("invalid range")
+	}
+
+	ras := strings.Split(s[len(b):], ",")
+	if len(ras) == 0 {
+		return fmt.Errorf("invalid request")
+	}
+	// Just pick the first one and ignore the rest, we only support one range per object
+	if len(ras) > 1 {
+		return fmt.Errorf("multiple ranges specified")
+	}
+
+	ra := strings.TrimSpace(ras[0])
+	if ra == "" {
+		return fmt.Errorf("invalid range")
+	}
+	i := strings.Index(ra, "-")
+	if i < 0 {
+		return fmt.Errorf("invalid range")
+	}
+	start, end := strings.TrimSpace(ra[:i]), strings.TrimSpace(ra[i+1:])
+	if start == "" {
+		// If no start is specified, end specifies the
+		// range start relative to the end of the file.
+		i, err := strconv.ParseInt(end, 10, 64)
+		if err != nil {
+			return fmt.Errorf("invalid range")
+		}
+		if i > r.size {
+			i = r.size
+		}
+		r.start = r.size - i
+		r.length = r.size - r.start
+	} else {
+		i, err := strconv.ParseInt(start, 10, 64)
+		if err != nil || i > r.size || i < 0 {
+			return fmt.Errorf("invalid range")
+		}
+		r.start = i
+		if end == "" {
+			// If no end is specified, range extends to end of the file.
+			r.length = r.size - r.start
+		} else {
+			i, err := strconv.ParseInt(end, 10, 64)
+			if err != nil || r.start > i {
+				return fmt.Errorf("invalid range")
+			}
+			if i >= r.size {
+				i = r.size - 1
+			}
+			r.length = i - r.start + 1
+		}
+	}
+	return nil
+}

pkg/storage/fs/fs_object.go

@@ -31,6 +31,54 @@ import (
 
 /// Object Operations
 
+// CopyObjectToWriterRange - GET object from range
+func (storage *Storage) CopyObjectToWriterRange(w io.Writer, bucket, object string, start, length int64) (int64, error) {
+	// validate bucket
+	if mstorage.IsValidBucket(bucket) == false {
+		return 0, mstorage.BucketNameInvalid{Bucket: bucket}
+	}
+
+	// validate object
+	if mstorage.IsValidObject(object) == false {
+		return 0, mstorage.ObjectNameInvalid{Bucket: bucket, Object: object}
+	}
+
+	objectPath := path.Join(storage.root, bucket, object)
+	filestat, err := os.Stat(objectPath)
+	switch err := err.(type) {
+	case nil:
+		{
+			if filestat.IsDir() {
+				return 0, mstorage.ObjectNotFound{Bucket: bucket, Object: object}
+			}
+		}
+	default:
+		{
+			if os.IsNotExist(err) {
+				return 0, mstorage.ObjectNotFound{Bucket: bucket, Object: object}
+			}
+			return 0, mstorage.EmbedError(bucket, object, err)
+		}
+	}
+	file, err := os.Open(objectPath)
+	defer file.Close()
+	if err != nil {
+		return 0, mstorage.EmbedError(bucket, object, err)
+	}
+
+	_, err = file.Seek(start, os.SEEK_SET)
+	if err != nil {
+		return 0, mstorage.EmbedError(bucket, object, err)
+	}
+
+	count, err := io.CopyN(w, file, length)
+	if err != nil {
+		return count, mstorage.EmbedError(bucket, object, err)
+	}
+
+	return count, nil
+}
+
 // CopyObjectToWriter - GET object
 func (storage *Storage) CopyObjectToWriter(w io.Writer, bucket string, object string) (int64, error) {
 	// validate bucket

pkg/storage/inmemory/inmemory.go

@@ -76,6 +76,11 @@ func (storage *Storage) CopyObjectToWriter(w io.Writer, bucket string, object st
 	return 0, mstorage.ObjectNotFound{Bucket: bucket, Object: object}
 }
 
+// CopyObjectToWriterRange - GET object from memory buffer range
+func (storage *Storage) CopyObjectToWriterRange(w io.Writer, bucket, object string, start, end int64) (int64, error) {
+	return 0, mstorage.APINotImplemented{API: "GetObjectRange"}
+}
+
 // StoreBucketPolicy - Not implemented
 func (storage *Storage) StoreBucketPolicy(bucket string, policy mstorage.BucketPolicy) error {
 	return mstorage.APINotImplemented{API: "PutBucketPolicy"}

pkg/storage/storage.go

@@ -33,6 +33,7 @@ type Storage interface {
 
 	// Object Operations
 	CopyObjectToWriter(w io.Writer, bucket string, object string) (int64, error)
+	CopyObjectToWriterRange(w io.Writer, bucket string, object string, start, length int64) (int64, error)
 	GetObjectMetadata(bucket string, object string, prefix string) (ObjectMetadata, error)
 	ListObjects(bucket string, resources BucketResourcesMetadata) ([]ObjectMetadata, BucketResourcesMetadata, error)
 	StoreObject(bucket string, key string, contentType string, data io.Reader) error