distribution/Godeps/_workspace/src/github.com/ncw/swift/swift.go

package swift

import (
	"bufio"
	"bytes"
	"crypto/hmac"
	"crypto/md5"
	"crypto/sha1"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"hash"
	"io"
	"mime"
	"net/http"
	"net/url"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	DefaultUserAgent    = "goswift/1.0"         // Default user agent
	DefaultRetries      = 3                     // Default number of retries on token expiry
	TimeFormat          = "2006-01-02T15:04:05" // Python date format for json replies parsed as UTC
	UploadTar           = "tar"                 // Data format specifier for Connection.BulkUpload().
	UploadTarGzip       = "tar.gz"              // Data format specifier for Connection.BulkUpload().
	UploadTarBzip2      = "tar.bz2"             // Data format specifier for Connection.BulkUpload().
	allContainersLimit  = 10000                 // Number of containers to fetch at once
	allObjectsLimit     = 10000                 // Number objects to fetch at once
	allObjectsChanLimit = 1000                  // ...when fetching to a channel
)

// Connection holds the details of the connection to the swift server.
//
// You need to provide UserName, ApiKey and AuthUrl when you create a
// connection then call Authenticate on it.
//
// The auth version in use will be detected from the AuthURL - you can
// override this with the AuthVersion parameter.
//
// If using v2 auth you can also set Region in the Connection
// structure.  If you don't set Region you will get the default region
// which may not be what you want.
//
// For reference some common AuthUrls looks like this:
//
//  Rackspace US        https://auth.api.rackspacecloud.com/v1.0
//  Rackspace UK        https://lon.auth.api.rackspacecloud.com/v1.0
//  Rackspace v2        https://identity.api.rackspacecloud.com/v2.0
//  Memset Memstore UK  https://auth.storage.memset.com/v1.0
//  Memstore v2         https://auth.storage.memset.com/v2.0
//
// When using Google Appengine you must provide the Connection with an
// appengine-specific Transport:
//
//	import (
//		"appengine/urlfetch"
//		"fmt"
//		"github.com/ncw/swift"
//	)
//
//	func handler(w http.ResponseWriter, r *http.Request) {
//		ctx := appengine.NewContext(r)
//		tr := urlfetch.Transport{Context: ctx}
//		c := swift.Connection{
//			UserName:  "user",
//			ApiKey:    "key",
//			AuthUrl:   "auth_url",
//			Transport: tr,
//		}
//		_ := c.Authenticate()
//		containers, _ := c.ContainerNames(nil)
//		fmt.Fprintf(w, "containers: %q", containers)
//	}
//
// If you don't supply a Transport, one is made which relies on
// http.ProxyFromEnvironment (http://golang.org/pkg/net/http/#ProxyFromEnvironment).
// This means that the connection will respect the HTTP proxy specified by the
// environment variables $HTTP_PROXY and $NO_PROXY.
type Connection struct {
	// Parameters - fill these in before calling Authenticate
	// They are all optional except UserName, ApiKey and AuthUrl
	Domain         string            // User's domain name
	DomainId       string            // User's domain Id
	UserName       string            // UserName for api
	ApiKey         string            // Key for api access
	AuthUrl        string            // Auth URL
	Retries        int               // Retries on error (default is 3)
	UserAgent      string            // Http User agent (default goswift/1.0)
	ConnectTimeout time.Duration     // Connect channel timeout (default 10s)
	Timeout        time.Duration     // Data channel timeout (default 60s)
	Region         string            // Region to use eg "LON", "ORD" - default is use first region (V2 auth only)
	AuthVersion    int               // Set to 1 or 2 or leave at 0 for autodetect
	Internal       bool              // Set this to true to use the the internal / service network
	Tenant         string            // Name of the tenant (v2 auth only)
	TenantId       string            // Id of the tenant (v2 auth only)
	TrustId        string            // Id of the trust (v3 auth only)
	Transport      http.RoundTripper `json:"-" xml:"-"` // Optional specialised http.Transport (eg. for Google Appengine)
	// These are filled in after Authenticate is called as are the defaults for above
	StorageUrl string
	AuthToken  string
	client     *http.Client
	Auth       Authenticator `json:"-" xml:"-"` // the current authenticator
	authLock   sync.Mutex    // lock when R/W StorageUrl, AuthToken, Auth
}

// Error - all errors generated by this package are of this type.  Other error
// may be passed on from library functions though.
type Error struct {
	StatusCode int // HTTP status code if relevant or 0 if not
	Text       string
}

// Error satisfy the error interface.
func (e *Error) Error() string {
	return e.Text
}

// newError make a new error from a string.
func newError(StatusCode int, Text string) *Error {
	return &Error{
		StatusCode: StatusCode,
		Text:       Text,
	}
}

// newErrorf makes a new error from sprintf parameters.
func newErrorf(StatusCode int, Text string, Parameters ...interface{}) *Error {
	return newError(StatusCode, fmt.Sprintf(Text, Parameters...))
}

// errorMap defines http error codes to error mappings.
type errorMap map[int]error

var (
	// Specific Errors you might want to check for equality
	BadRequest          = newError(400, "Bad Request")
	AuthorizationFailed = newError(401, "Authorization Failed")
	ContainerNotFound   = newError(404, "Container Not Found")
	ContainerNotEmpty   = newError(409, "Container Not Empty")
	ObjectNotFound      = newError(404, "Object Not Found")
	ObjectCorrupted     = newError(422, "Object Corrupted")
	TimeoutError        = newError(408, "Timeout when reading or writing data")
	Forbidden           = newError(403, "Operation forbidden")
	TooLargeObject      = newError(413, "Too Large Object")

	// Mappings for authentication errors
	authErrorMap = errorMap{
		400: BadRequest,
		401: AuthorizationFailed,
		403: Forbidden,
	}

	// Mappings for container errors
	ContainerErrorMap = errorMap{
		400: BadRequest,
		403: Forbidden,
		404: ContainerNotFound,
		409: ContainerNotEmpty,
	}

	// Mappings for object errors
	objectErrorMap = errorMap{
		400: BadRequest,
		403: Forbidden,
		404: ObjectNotFound,
		413: TooLargeObject,
		422: ObjectCorrupted,
	}
)

// checkClose is used to check the return from Close in a defer
// statement.
func checkClose(c io.Closer, err *error) {
	cerr := c.Close()
	if *err == nil {
		*err = cerr
	}
}

// parseHeaders checks a response for errors and translates into
// standard errors if necessary.
func (c *Connection) parseHeaders(resp *http.Response, errorMap errorMap) error {
	if errorMap != nil {
		if err, ok := errorMap[resp.StatusCode]; ok {
			return err
		}
	}
	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		return newErrorf(resp.StatusCode, "HTTP Error: %d: %s", resp.StatusCode, resp.Status)
	}
	return nil
}

// readHeaders returns a Headers object from the http.Response.
//
// If it receives multiple values for a key (which should never
// happen) it will use the first one
func readHeaders(resp *http.Response) Headers {
	headers := Headers{}
	for key, values := range resp.Header {
		headers[key] = values[0]
	}
	return headers
}

// Headers stores HTTP headers (can only have one of each header like Swift).
type Headers map[string]string

// Does an http request using the running timer passed in
func (c *Connection) doTimeoutRequest(timer *time.Timer, req *http.Request) (*http.Response, error) {
	// Do the request in the background so we can check the timeout
	type result struct {
		resp *http.Response
		err  error
	}
	done := make(chan result, 1)
	go func() {
		resp, err := c.client.Do(req)
		done <- result{resp, err}
	}()
	// Wait for the read or the timeout
	select {
	case r := <-done:
		return r.resp, r.err
	case <-timer.C:
		// Kill the connection on timeout so we don't leak sockets or goroutines
		cancelRequest(c.Transport, req)
		return nil, TimeoutError
	}
	panic("unreachable") // For Go 1.0
}

// Set defaults for any unset values
//
// Call with authLock held
func (c *Connection) setDefaults() {
	if c.UserAgent == "" {
		c.UserAgent = DefaultUserAgent
	}
	if c.Retries == 0 {
		c.Retries = DefaultRetries
	}
	if c.ConnectTimeout == 0 {
		c.ConnectTimeout = 10 * time.Second
	}
	if c.Timeout == 0 {
		c.Timeout = 60 * time.Second
	}
	if c.Transport == nil {
		c.Transport = &http.Transport{
			//		TLSClientConfig:    &tls.Config{RootCAs: pool},
			//		DisableCompression: true,
			Proxy:               http.ProxyFromEnvironment,
			MaxIdleConnsPerHost: 2048,
		}
	}
	if c.client == nil {
		c.client = &http.Client{
			//		CheckRedirect: redirectPolicyFunc,
			Transport: c.Transport,
		}
	}
}

// Authenticate connects to the Swift server.
//
// If you don't call it before calling one of the connection methods
// then it will be called for you on the first access.
func (c *Connection) Authenticate() (err error) {
	c.authLock.Lock()
	defer c.authLock.Unlock()
	return c.authenticate()
}

// Internal implementation of Authenticate
//
// Call with authLock held
func (c *Connection) authenticate() (err error) {
	c.setDefaults()

	// Flush the keepalives connection - if we are
	// re-authenticating then stuff has gone wrong
	flushKeepaliveConnections(c.Transport)

	if c.Auth == nil {
		c.Auth, err = newAuth(c)
		if err != nil {
			return
		}
	}

	retries := 1
again:
	var req *http.Request
	req, err = c.Auth.Request(c)
	if err != nil {
		return
	}
	timer := time.NewTimer(c.ConnectTimeout)
	var resp *http.Response
	resp, err = c.doTimeoutRequest(timer, req)
	if err != nil {
		return
	}
	defer func() {
		checkClose(resp.Body, &err)
		// Flush the auth connection - we don't want to keep
		// it open if keepalives were enabled
		flushKeepaliveConnections(c.Transport)
	}()
	if err = c.parseHeaders(resp, authErrorMap); err != nil {
		// Try again for a limited number of times on
		// AuthorizationFailed or BadRequest. This allows us
		// to try some alternate forms of the request
		if (err == AuthorizationFailed || err == BadRequest) && retries > 0 {
			retries--
			goto again
		}
		return
	}
	err = c.Auth.Response(resp)
	if err != nil {
		return
	}
	c.StorageUrl = c.Auth.StorageUrl(c.Internal)
	c.AuthToken = c.Auth.Token()
	if !c.authenticated() {
		err = newError(0, "Response didn't have storage url and auth token")
		return
	}
	return
}

// Get an authToken and url
//
// The Url may be updated if it needed to authenticate using the OnReAuth function
func (c *Connection) getUrlAndAuthToken(targetUrlIn string, OnReAuth func() (string, error)) (targetUrlOut, authToken string, err error) {
	c.authLock.Lock()
	defer c.authLock.Unlock()
	targetUrlOut = targetUrlIn
	if !c.authenticated() {
		err = c.authenticate()
		if err != nil {
			return
		}
		if OnReAuth != nil {
			targetUrlOut, err = OnReAuth()
			if err != nil {
				return
			}
		}
	}
	authToken = c.AuthToken
	return
}

// flushKeepaliveConnections is called to flush pending requests after an error.
func flushKeepaliveConnections(transport http.RoundTripper) {
	if tr, ok := transport.(interface {
		CloseIdleConnections()
	}); ok {
		tr.CloseIdleConnections()
	}
}

// UnAuthenticate removes the authentication from the Connection.
func (c *Connection) UnAuthenticate() {
	c.authLock.Lock()
	c.StorageUrl = ""
	c.AuthToken = ""
	c.authLock.Unlock()
}

// Authenticated returns a boolean to show if the current connection
// is authenticated.
//
// Doesn't actually check the credentials against the server.
func (c *Connection) Authenticated() bool {
	c.authLock.Lock()
	defer c.authLock.Unlock()
	return c.authenticated()
}

// Internal version of Authenticated()
//
// Call with authLock held
func (c *Connection) authenticated() bool {
	return c.StorageUrl != "" && c.AuthToken != ""
}

// RequestOpts contains parameters for Connection.storage.
type RequestOpts struct {
	Container  string
	ObjectName string
	Operation  string
	Parameters url.Values
	Headers    Headers
	ErrorMap   errorMap
	NoResponse bool
	Body       io.Reader
	Retries    int
	// if set this is called on re-authentication to refresh the targetUrl
	OnReAuth func() (string, error)
}

// Call runs a remote command on the targetUrl, returns a
// response, headers and possible error.
//
// operation is GET, HEAD etc
// container is the name of a container
// Any other parameters (if not None) are added to the targetUrl
//
// Returns a response or an error.  If response is returned then
// resp.Body.Close() must be called on it, unless noResponse is set in
// which case the body will be closed in this function
//
// This will Authenticate if necessary, and re-authenticate if it
// receives a 401 error which means the token has expired
//
// This method is exported so extensions can call it.
func (c *Connection) Call(targetUrl string, p RequestOpts) (resp *http.Response, headers Headers, err error) {
	c.authLock.Lock()
	c.setDefaults()
	c.authLock.Unlock()
	retries := p.Retries
	if retries == 0 {
		retries = c.Retries
	}
	var req *http.Request
	for {
		var authToken string
		targetUrl, authToken, err = c.getUrlAndAuthToken(targetUrl, p.OnReAuth)

		var URL *url.URL
		URL, err = url.Parse(targetUrl)
		if err != nil {
			return
		}
		if p.Container != "" {
			URL.Path += "/" + p.Container
			if p.ObjectName != "" {
				URL.Path += "/" + p.ObjectName
			}
		}
		if p.Parameters != nil {
			URL.RawQuery = p.Parameters.Encode()
		}
		timer := time.NewTimer(c.ConnectTimeout)
		reader := p.Body
		if reader != nil {
			reader = newWatchdogReader(reader, c.Timeout, timer)
		}
		req, err = http.NewRequest(p.Operation, URL.String(), reader)
		if err != nil {
			return
		}
		if p.Headers != nil {
			for k, v := range p.Headers {
				req.Header.Add(k, v)
			}
		}
		req.Header.Add("User-Agent", DefaultUserAgent)
		req.Header.Add("X-Auth-Token", authToken)
		resp, err = c.doTimeoutRequest(timer, req)
		if err != nil {
			if p.Operation == "HEAD" || p.Operation == "GET" {
				retries--
				continue
			}
			return
		}
		// Check to see if token has expired
		if resp.StatusCode == 401 && retries > 0 {
			_ = resp.Body.Close()
			c.UnAuthenticate()
			retries--
		} else {
			break
		}
	}

	if err = c.parseHeaders(resp, p.ErrorMap); err != nil {
		_ = resp.Body.Close()
		return nil, nil, err
	}
	headers = readHeaders(resp)
	if p.NoResponse {
		err = resp.Body.Close()
		if err != nil {
			return nil, nil, err
		}
	} else {
		// Cancel the request on timeout
		cancel := func() {
			cancelRequest(c.Transport, req)
		}
		// Wrap resp.Body to make it obey an idle timeout
		resp.Body = newTimeoutReader(resp.Body, c.Timeout, cancel)
	}
	return
}

// storage runs a remote command on a the storage url, returns a
// response, headers and possible error.
//
// operation is GET, HEAD etc
// container is the name of a container
// Any other parameters (if not None) are added to the storage url
//
// Returns a response or an error.  If response is returned then
// resp.Body.Close() must be called on it, unless noResponse is set in
// which case the body will be closed in this function
//
// This will Authenticate if necessary, and re-authenticate if it
// receives a 401 error which means the token has expired
func (c *Connection) storage(p RequestOpts) (resp *http.Response, headers Headers, err error) {
	p.OnReAuth = func() (string, error) {
		return c.StorageUrl, nil
	}
	c.authLock.Lock()
	url := c.StorageUrl
	c.authLock.Unlock()
	return c.Call(url, p)
}

// readLines reads the response into an array of strings.
//
// Closes the response when done
func readLines(resp *http.Response) (lines []string, err error) {
	defer checkClose(resp.Body, &err)
	reader := bufio.NewReader(resp.Body)
	buffer := bytes.NewBuffer(make([]byte, 0, 128))
	var part []byte
	var prefix bool
	for {
		if part, prefix, err = reader.ReadLine(); err != nil {
			break
		}
		buffer.Write(part)
		if !prefix {
			lines = append(lines, buffer.String())
			buffer.Reset()
		}
	}
	if err == io.EOF {
		err = nil
	}
	return
}

// readJson reads the response into the json type passed in
//
// Closes the response when done
func readJson(resp *http.Response, result interface{}) (err error) {
	defer checkClose(resp.Body, &err)
	decoder := json.NewDecoder(resp.Body)
	return decoder.Decode(result)
}

/* ------------------------------------------------------------ */

// ContainersOpts is options for Containers() and ContainerNames()
type ContainersOpts struct {
	Limit     int     // For an integer value n, limits the number of results to at most n values.
	Marker    string  // Given a string value x, return object names greater in value than the specified marker.
	EndMarker string  // Given a string value x, return container names less in value than the specified marker.
	Headers   Headers // Any additional HTTP headers - can be nil
}

// parse the ContainerOpts
func (opts *ContainersOpts) parse() (url.Values, Headers) {
	v := url.Values{}
	var h Headers
	if opts != nil {
		if opts.Limit > 0 {
			v.Set("limit", strconv.Itoa(opts.Limit))
		}
		if opts.Marker != "" {
			v.Set("marker", opts.Marker)
		}
		if opts.EndMarker != "" {
			v.Set("end_marker", opts.EndMarker)
		}
		h = opts.Headers
	}
	return v, h
}

// ContainerNames returns a slice of names of containers in this account.
func (c *Connection) ContainerNames(opts *ContainersOpts) ([]string, error) {
	v, h := opts.parse()
	resp, _, err := c.storage(RequestOpts{
		Operation:  "GET",
		Parameters: v,
		ErrorMap:   ContainerErrorMap,
		Headers:    h,
	})
	if err != nil {
		return nil, err
	}
	lines, err := readLines(resp)
	return lines, err
}

// Container contains information about a container
type Container struct {
	Name  string // Name of the container
	Count int64  // Number of objects in the container
	Bytes int64  // Total number of bytes used in the container
}

// Containers returns a slice of structures with full information as
// described in Container.
func (c *Connection) Containers(opts *ContainersOpts) ([]Container, error) {
	v, h := opts.parse()
	v.Set("format", "json")
	resp, _, err := c.storage(RequestOpts{
		Operation:  "GET",
		Parameters: v,
		ErrorMap:   ContainerErrorMap,
		Headers:    h,
	})
	if err != nil {
		return nil, err
	}
	var containers []Container
	err = readJson(resp, &containers)
	return containers, err
}

// containersAllOpts makes a copy of opts if set or makes a new one and
// overrides Limit and Marker
func containersAllOpts(opts *ContainersOpts) *ContainersOpts {
	var newOpts ContainersOpts
	if opts != nil {
		newOpts = *opts
	}
	if newOpts.Limit == 0 {
		newOpts.Limit = allContainersLimit
	}
	newOpts.Marker = ""
	return &newOpts
}

// ContainersAll is like Containers but it returns all the Containers
//
// It calls Containers multiple times using the Marker parameter
//
// It has a default Limit parameter but you may pass in your own
func (c *Connection) ContainersAll(opts *ContainersOpts) ([]Container, error) {
	opts = containersAllOpts(opts)
	containers := make([]Container, 0)
	for {
		newContainers, err := c.Containers(opts)
		if err != nil {
			return nil, err
		}
		containers = append(containers, newContainers...)
		if len(newContainers) < opts.Limit {
			break
		}
		opts.Marker = newContainers[len(newContainers)-1].Name
	}
	return containers, nil
}

// ContainerNamesAll is like ContainerNamess but it returns all the Containers
//
// It calls ContainerNames multiple times using the Marker parameter
//
// It has a default Limit parameter but you may pass in your own
func (c *Connection) ContainerNamesAll(opts *ContainersOpts) ([]string, error) {
	opts = containersAllOpts(opts)
	containers := make([]string, 0)
	for {
		newContainers, err := c.ContainerNames(opts)
		if err != nil {
			return nil, err
		}
		containers = append(containers, newContainers...)
		if len(newContainers) < opts.Limit {
			break
		}
		opts.Marker = newContainers[len(newContainers)-1]
	}
	return containers, nil
}

/* ------------------------------------------------------------ */

// ObjectOpts is options for Objects() and ObjectNames()
type ObjectsOpts struct {
	Limit     int     // For an integer value n, limits the number of results to at most n values.
	Marker    string  // Given a string value x, return object names greater in value than the  specified marker.
	EndMarker string  // Given a string value x, return object names less in value than the specified marker
	Prefix    string  // For a string value x, causes the results to be limited to object names beginning with the substring x.
	Path      string  // For a string value x, return the object names nested in the pseudo path
	Delimiter rune    // For a character c, return all the object names nested in the container
	Headers   Headers // Any additional HTTP headers - can be nil
}

// parse reads values out of ObjectsOpts
func (opts *ObjectsOpts) parse() (url.Values, Headers) {
	v := url.Values{}
	var h Headers
	if opts != nil {
		if opts.Limit > 0 {
			v.Set("limit", strconv.Itoa(opts.Limit))
		}
		if opts.Marker != "" {
			v.Set("marker", opts.Marker)
		}
		if opts.EndMarker != "" {
			v.Set("end_marker", opts.EndMarker)
		}
		if opts.Prefix != "" {
			v.Set("prefix", opts.Prefix)
		}
		if opts.Path != "" {
			v.Set("path", opts.Path)
		}
		if opts.Delimiter != 0 {
			v.Set("delimiter", string(opts.Delimiter))
		}
		h = opts.Headers
	}
	return v, h
}

// ObjectNames returns a slice of names of objects in a given container.
func (c *Connection) ObjectNames(container string, opts *ObjectsOpts) ([]string, error) {
	v, h := opts.parse()
	resp, _, err := c.storage(RequestOpts{
		Container:  container,
		Operation:  "GET",
		Parameters: v,
		ErrorMap:   ContainerErrorMap,
		Headers:    h,
	})
	if err != nil {
		return nil, err
	}
	return readLines(resp)
}

// Object contains information about an object
type Object struct {
	Name               string    `json:"name"`          // object name
	ContentType        string    `json:"content_type"`  // eg application/directory
	Bytes              int64     `json:"bytes"`         // size in bytes
	ServerLastModified string    `json:"last_modified"` // Last modified time, eg '2011-06-30T08:20:47.736680' as a string supplied by the server
	LastModified       time.Time // Last modified time converted to a time.Time
	Hash               string    `json:"hash"` // MD5 hash, eg "d41d8cd98f00b204e9800998ecf8427e"
	PseudoDirectory    bool      // Set when using delimiter to show that this directory object does not really exist
	SubDir             string    `json:"subdir"` // returned only when using delimiter to mark "pseudo directories"
}

// Objects returns a slice of Object with information about each
// object in the container.
//
// If Delimiter is set in the opts then PseudoDirectory may be set,
// with ContentType 'application/directory'.  These are not real
// objects but represent directories of objects which haven't had an
// object created for them.
func (c *Connection) Objects(container string, opts *ObjectsOpts) ([]Object, error) {
	v, h := opts.parse()
	v.Set("format", "json")
	resp, _, err := c.storage(RequestOpts{
		Container:  container,
		Operation:  "GET",
		Parameters: v,
		ErrorMap:   ContainerErrorMap,
		Headers:    h,
	})
	if err != nil {
		return nil, err
	}
	var objects []Object
	err = readJson(resp, &objects)
	// Convert Pseudo directories and dates
	for i := range objects {
		object := &objects[i]
		if object.SubDir != "" {
			object.Name = object.SubDir
			object.PseudoDirectory = true
			object.ContentType = "application/directory"
		}
		if object.ServerLastModified != "" {
			// 2012-11-11T14:49:47.887250
			//
			// Remove fractional seconds if present. This
			// then keeps it consistent with Object
			// which can only return timestamps accurate
			// to 1 second
			//
			// The TimeFormat will parse fractional
			// seconds if desired though
			datetime := strings.SplitN(object.ServerLastModified, ".", 2)[0]
			object.LastModified, err = time.Parse(TimeFormat, datetime)
			if err != nil {
				return nil, err
			}
		}
	}
	return objects, err
}

// objectsAllOpts makes a copy of opts if set or makes a new one and
// overrides Limit and Marker
func objectsAllOpts(opts *ObjectsOpts, Limit int) *ObjectsOpts {
	var newOpts ObjectsOpts
	if opts != nil {
		newOpts = *opts
	}
	if newOpts.Limit == 0 {
		newOpts.Limit = Limit
	}
	newOpts.Marker = ""
	return &newOpts
}

// A closure defined by the caller to iterate through all objects
//
// Call Objects or ObjectNames from here with the *ObjectOpts passed in
//
// Do whatever is required with the results then return them
type ObjectsWalkFn func(*ObjectsOpts) (interface{}, error)

// ObjectsWalk is uses to iterate through all the objects in chunks as
// returned by Objects or ObjectNames using the Marker and Limit
// parameters in the ObjectsOpts.
//
// Pass in a closure `walkFn` which calls Objects or ObjectNames with
// the *ObjectsOpts passed to it and does something with the results.
//
// Errors will be returned from this function
//
// It has a default Limit parameter but you may pass in your own
func (c *Connection) ObjectsWalk(container string, opts *ObjectsOpts, walkFn ObjectsWalkFn) error {
	opts = objectsAllOpts(opts, allObjectsChanLimit)
	for {
		objects, err := walkFn(opts)
		if err != nil {
			return err
		}
		var n int
		var last string
		switch objects := objects.(type) {
		case []string:
			n = len(objects)
			if n > 0 {
				last = objects[len(objects)-1]
			}
		case []Object:
			n = len(objects)
			if n > 0 {
				last = objects[len(objects)-1].Name
			}
		default:
			panic("Unknown type returned to ObjectsWalk")
		}
		if n < opts.Limit {
			break
		}
		opts.Marker = last
	}
	return nil
}

// ObjectsAll is like Objects but it returns an unlimited number of Objects in a slice
//
// It calls Objects multiple times using the Marker parameter
func (c *Connection) ObjectsAll(container string, opts *ObjectsOpts) ([]Object, error) {
	objects := make([]Object, 0)
	err := c.ObjectsWalk(container, opts, func(opts *ObjectsOpts) (interface{}, error) {
		newObjects, err := c.Objects(container, opts)
		if err == nil {
			objects = append(objects, newObjects...)
		}
		return newObjects, err
	})
	return objects, err
}

// ObjectNamesAll is like ObjectNames but it returns all the Objects
//
// It calls ObjectNames multiple times using the Marker parameter
//
// It has a default Limit parameter but you may pass in your own
func (c *Connection) ObjectNamesAll(container string, opts *ObjectsOpts) ([]string, error) {
	objects := make([]string, 0)
	err := c.ObjectsWalk(container, opts, func(opts *ObjectsOpts) (interface{}, error) {
		newObjects, err := c.ObjectNames(container, opts)
		if err == nil {
			objects = append(objects, newObjects...)
		}
		return newObjects, err
	})
	return objects, err
}

// Account contains information about this account.
type Account struct {
	BytesUsed  int64 // total number of bytes used
	Containers int64 // total number of containers
	Objects    int64 // total number of objects
}

// getInt64FromHeader is a helper function to decode int64 from header.
func getInt64FromHeader(resp *http.Response, header string) (result int64, err error) {
	value := resp.Header.Get(header)
	result, err = strconv.ParseInt(value, 10, 64)
	if err != nil {
		err = newErrorf(0, "Bad Header '%s': '%s': %s", header, value, err)
	}
	return
}

// Account returns info about the account in an Account struct.
func (c *Connection) Account() (info Account, headers Headers, err error) {
	var resp *http.Response
	resp, headers, err = c.storage(RequestOpts{
		Operation:  "HEAD",
		ErrorMap:   ContainerErrorMap,
		NoResponse: true,
	})
	if err != nil {
		return
	}
	// Parse the headers into a dict
	//
	//    {'Accept-Ranges': 'bytes',
	//     'Content-Length': '0',
	//     'Date': 'Tue, 05 Jul 2011 16:37:06 GMT',
	//     'X-Account-Bytes-Used': '316598182',
	//     'X-Account-Container-Count': '4',
	//     'X-Account-Object-Count': '1433'}
	if info.BytesUsed, err = getInt64FromHeader(resp, "X-Account-Bytes-Used"); err != nil {
		return
	}
	if info.Containers, err = getInt64FromHeader(resp, "X-Account-Container-Count"); err != nil {
		return
	}
	if info.Objects, err = getInt64FromHeader(resp, "X-Account-Object-Count"); err != nil {
		return
	}
	return
}

// AccountUpdate adds, replaces or remove account metadata.
//
// Add or update keys by mentioning them in the Headers.
//
// Remove keys by setting them to an empty string.
func (c *Connection) AccountUpdate(h Headers) error {
	_, _, err := c.storage(RequestOpts{
		Operation:  "POST",
		ErrorMap:   ContainerErrorMap,
		NoResponse: true,
		Headers:    h,
	})
	return err
}

// ContainerCreate creates a container.
//
// If you don't want to add Headers just pass in nil
//
// No error is returned if it already exists but the metadata if any will be updated.
func (c *Connection) ContainerCreate(container string, h Headers) error {
	_, _, err := c.storage(RequestOpts{
		Container:  container,
		Operation:  "PUT",
		ErrorMap:   ContainerErrorMap,
		NoResponse: true,
		Headers:    h,
	})
	return err
}

// ContainerDelete deletes a container.
//
// May return ContainerDoesNotExist or ContainerNotEmpty
func (c *Connection) ContainerDelete(container string) error {
	_, _, err := c.storage(RequestOpts{
		Container:  container,
		Operation:  "DELETE",
		ErrorMap:   ContainerErrorMap,
		NoResponse: true,
	})
	return err
}

// Container returns info about a single container including any
// metadata in the headers.
func (c *Connection) Container(container string) (info Container, headers Headers, err error) {
	var resp *http.Response
	resp, headers, err = c.storage(RequestOpts{
		Container:  container,
		Operation:  "HEAD",
		ErrorMap:   ContainerErrorMap,
		NoResponse: true,
	})
	if err != nil {
		return
	}
	// Parse the headers into the struct
	info.Name = container
	if info.Bytes, err = getInt64FromHeader(resp, "X-Container-Bytes-Used"); err != nil {
		return
	}
	if info.Count, err = getInt64FromHeader(resp, "X-Container-Object-Count"); err != nil {
		return
	}
	return
}

// ContainerUpdate adds, replaces or removes container metadata.
//
// Add or update keys by mentioning them in the Metadata.
//
// Remove keys by setting them to an empty string.
//
// Container metadata can only be read with Container() not with Containers().
func (c *Connection) ContainerUpdate(container string, h Headers) error {
	_, _, err := c.storage(RequestOpts{
		Container:  container,
		Operation:  "POST",
		ErrorMap:   ContainerErrorMap,
		NoResponse: true,
		Headers:    h,
	})
	return err
}

// ------------------------------------------------------------

// ObjectCreateFile represents a swift object open for writing
type ObjectCreateFile struct {
	checkHash  bool           // whether we are checking the hash
	pipeReader *io.PipeReader // pipe for the caller to use
	pipeWriter *io.PipeWriter
	hash       hash.Hash      // hash being build up as we go along
	done       chan struct{}  // signals when the upload has finished
	resp       *http.Response // valid when done has signalled
	err        error          // ditto
	headers    Headers        // ditto
}

// Write bytes to the object - see io.Writer
func (file *ObjectCreateFile) Write(p []byte) (n int, err error) {
	n, err = file.pipeWriter.Write(p)
	if err == io.ErrClosedPipe {
		if file.err != nil {
			return 0, file.err
		}
		return 0, newError(500, "Write on closed file")
	}
	if err == nil && file.checkHash {
		_, _ = file.hash.Write(p)
	}
	return
}

// Close the object and checks the md5sum if it was required.
//
// Also returns any other errors from the server (eg container not
// found) so it is very important to check the errors on this method.
func (file *ObjectCreateFile) Close() error {
	// Close the body
	err := file.pipeWriter.Close()
	if err != nil {
		return err
	}

	// Wait for the HTTP operation to complete
	<-file.done

	// Check errors
	if file.err != nil {
		return file.err
	}
	if file.checkHash {
		receivedMd5 := strings.ToLower(file.headers["Etag"])
		calculatedMd5 := fmt.Sprintf("%x", file.hash.Sum(nil))
		if receivedMd5 != calculatedMd5 {
			return ObjectCorrupted
		}
	}
	return nil
}

// Check it satisfies the interface
var _ io.WriteCloser = &ObjectCreateFile{}

// objectPutHeaders create a set of headers for a PUT
//
// It guesses the contentType from the objectName if it isn't set
//
// checkHash may be changed
func objectPutHeaders(objectName string, checkHash *bool, Hash string, contentType string, h Headers) Headers {
	if contentType == "" {
		contentType = mime.TypeByExtension(path.Ext(objectName))
		if contentType == "" {
			contentType = "application/octet-stream"
		}
	}
	// Meta stuff
	extraHeaders := map[string]string{
		"Content-Type": contentType,
	}
	for key, value := range h {
		extraHeaders[key] = value
	}
	if Hash != "" {
		extraHeaders["Etag"] = Hash
		*checkHash = false // the server will do it
	}
	return extraHeaders
}

// ObjectCreate creates or updates the object in the container.  It
// returns an io.WriteCloser you should write the contents to.  You
// MUST call Close() on it and you MUST check the error return from
// Close().
//
// If checkHash is True then it will calculate the MD5 Hash of the
// file as it is being uploaded and check it against that returned
// from the server.  If it is wrong then it will return
// ObjectCorrupted on Close()
//
// If you know the MD5 hash of the object ahead of time then set the
// Hash parameter and it will be sent to the server (as an Etag
// header) and the server will check the MD5 itself after the upload,
// and this will return ObjectCorrupted on Close() if it is incorrect.
//
// If you don't want any error protection (not recommended) then set
// checkHash to false and Hash to "".
//
// If contentType is set it will be used, otherwise one will be
// guessed from objectName using mime.TypeByExtension
func (c *Connection) ObjectCreate(container string, objectName string, checkHash bool, Hash string, contentType string, h Headers) (file *ObjectCreateFile, err error) {
	extraHeaders := objectPutHeaders(objectName, &checkHash, Hash, contentType, h)
	pipeReader, pipeWriter := io.Pipe()
	file = &ObjectCreateFile{
		hash:       md5.New(),
		checkHash:  checkHash,
		pipeReader: pipeReader,
		pipeWriter: pipeWriter,
		done:       make(chan struct{}),
	}
	// Run the PUT in the background piping it data
	go func() {
		file.resp, file.headers, file.err = c.storage(RequestOpts{
			Container:  container,
			ObjectName: objectName,
			Operation:  "PUT",
			Headers:    extraHeaders,
			Body:       pipeReader,
			NoResponse: true,
			ErrorMap:   objectErrorMap,
		})
		// Signal finished
		pipeReader.Close()
		close(file.done)
	}()
	return
}

// ObjectPut creates or updates the path in the container from
// contents.  contents should be an open io.Reader which will have all
// its contents read.
//
// This is a low level interface.
//
// If checkHash is True then it will calculate the MD5 Hash of the
// file as it is being uploaded and check it against that returned
// from the server.  If it is wrong then it will return
// ObjectCorrupted.
//
// If you know the MD5 hash of the object ahead of time then set the
// Hash parameter and it will be sent to the server (as an Etag
// header) and the server will check the MD5 itself after the upload,
// and this will return ObjectCorrupted if it is incorrect.
//
// If you don't want any error protection (not recommended) then set
// checkHash to false and Hash to "".
//
// If contentType is set it will be used, otherwise one will be
// guessed from objectName using mime.TypeByExtension
func (c *Connection) ObjectPut(container string, objectName string, contents io.Reader, checkHash bool, Hash string, contentType string, h Headers) (headers Headers, err error) {
	extraHeaders := objectPutHeaders(objectName, &checkHash, Hash, contentType, h)
	hash := md5.New()
	var body io.Reader = contents
	if checkHash {
		body = io.TeeReader(contents, hash)
	}
	_, headers, err = c.storage(RequestOpts{
		Container:  container,
		ObjectName: objectName,
		Operation:  "PUT",
		Headers:    extraHeaders,
		Body:       body,
		NoResponse: true,
		ErrorMap:   objectErrorMap,
	})
	if err != nil {
		return
	}
	if checkHash {
		receivedMd5 := strings.ToLower(headers["Etag"])
		calculatedMd5 := fmt.Sprintf("%x", hash.Sum(nil))
		if receivedMd5 != calculatedMd5 {
			err = ObjectCorrupted
			return
		}
	}
	return
}

// ObjectPutBytes creates an object from a []byte in a container.
//
// This is a simplified interface which checks the MD5.
func (c *Connection) ObjectPutBytes(container string, objectName string, contents []byte, contentType string) (err error) {
	buf := bytes.NewBuffer(contents)
	_, err = c.ObjectPut(container, objectName, buf, true, "", contentType, nil)
	return
}

// ObjectPutString creates an object from a string in a container.
//
// This is a simplified interface which checks the MD5
func (c *Connection) ObjectPutString(container string, objectName string, contents string, contentType string) (err error) {
	buf := strings.NewReader(contents)
	_, err = c.ObjectPut(container, objectName, buf, true, "", contentType, nil)
	return
}

// ObjectOpenFile represents a swift object open for reading
type ObjectOpenFile struct {
	connection *Connection    // stored copy of Connection used in Open
	container  string         // stored copy of container used in Open
	objectName string         // stored copy of objectName used in Open
	headers    Headers        // stored copy of headers used in Open
	resp       *http.Response // http connection
	body       io.Reader      // read data from this
	checkHash  bool           // true if checking MD5
	hash       hash.Hash      // currently accumulating MD5
	bytes      int64          // number of bytes read on this connection
	eof        bool           // whether we have read end of file
	pos        int64          // current position when reading
	lengthOk   bool           // whether length is valid
	length     int64          // length of the object if read
	seeked     bool           // whether we have seeked this file or not
}

// Read bytes from the object - see io.Reader
func (file *ObjectOpenFile) Read(p []byte) (n int, err error) {
	n, err = file.body.Read(p)
	file.bytes += int64(n)
	file.pos += int64(n)
	if err == io.EOF {
		file.eof = true
	}
	return
}

// Seek sets the offset for the next Read to offset, interpreted
// according to whence: 0 means relative to the origin of the file, 1
// means relative to the current offset, and 2 means relative to the
// end. Seek returns the new offset and an Error, if any.
//
// Seek uses HTTP Range headers which, if the file pointer is moved,
// will involve reopening the HTTP connection.
//
// Note that you can't seek to the end of a file or beyond; HTTP Range
// requests don't support the file pointer being outside the data,
// unlike os.File
//
// Seek(0, 1) will return the current file pointer.
func (file *ObjectOpenFile) Seek(offset int64, whence int) (newPos int64, err error) {
	switch whence {
	case 0: // relative to start
		newPos = offset
	case 1: // relative to current
		newPos = file.pos + offset
	case 2: // relative to end
		if !file.lengthOk {
			return file.pos, newError(0, "Length of file unknown so can't seek from end")
		}
		newPos = file.length + offset
	default:
		panic("Unknown whence in ObjectOpenFile.Seek")
	}
	// If at correct position (quite likely), do nothing
	if newPos == file.pos {
		return
	}
	// Close the file...
	file.seeked = true
	err = file.Close()
	if err != nil {
		return
	}
	// ...and re-open with a Range header
	if file.headers == nil {
		file.headers = Headers{}
	}
	if newPos > 0 {
		file.headers["Range"] = fmt.Sprintf("bytes=%d-", newPos)
	} else {
		delete(file.headers, "Range")
	}
	newFile, _, err := file.connection.ObjectOpen(file.container, file.objectName, false, file.headers)
	if err != nil {
		return
	}
	// Update the file
	file.resp = newFile.resp
	file.body = newFile.body
	file.checkHash = false
	file.pos = newPos
	return
}

// Length gets the objects content length either from a cached copy or
// from the server.
func (file *ObjectOpenFile) Length() (int64, error) {
	if !file.lengthOk {
		info, _, err := file.connection.Object(file.container, file.objectName)
		file.length = info.Bytes
		file.lengthOk = (err == nil)
		return file.length, err
	}
	return file.length, nil
}

// Close the object and checks the length and md5sum if it was
// required and all the object was read
func (file *ObjectOpenFile) Close() (err error) {
	// Close the body at the end
	defer checkClose(file.resp.Body, &err)

	// If not end of file or seeked then can't check anything
	if !file.eof || file.seeked {
		return
	}

	// Check the MD5 sum if requested
	if file.checkHash {
		receivedMd5 := strings.ToLower(file.resp.Header.Get("Etag"))
		calculatedMd5 := fmt.Sprintf("%x", file.hash.Sum(nil))
		if receivedMd5 != calculatedMd5 {
			err = ObjectCorrupted
			return
		}
	}

	// Check to see we read the correct number of bytes
	if file.lengthOk && file.length != file.bytes {
		err = ObjectCorrupted
		return
	}
	return
}

// Check it satisfies the interfaces
var _ io.ReadCloser = &ObjectOpenFile{}
var _ io.Seeker = &ObjectOpenFile{}

// ObjectOpen returns an ObjectOpenFile for reading the contents of
// the object.  This satisfies the io.ReadCloser and the io.Seeker
// interfaces.
//
// You must call Close() on contents when finished
//
// Returns the headers of the response.
//
// If checkHash is true then it will calculate the md5sum of the file
// as it is being received and check it against that returned from the
// server.  If it is wrong then it will return ObjectCorrupted. It
// will also check the length returned. No checking will be done if
// you don't read all the contents.
//
// Note that objects with X-Object-Manifest set won't ever have their
// md5sum's checked as the md5sum reported on the object is actually
// the md5sum of the md5sums of the parts. This isn't very helpful to
// detect a corrupted download as the size of the parts aren't known
// without doing more operations.  If you want to ensure integrity of
// an object with a manifest then you will need to download everything
// in the manifest separately.
//
// headers["Content-Type"] will give the content type if desired.
func (c *Connection) ObjectOpen(container string, objectName string, checkHash bool, h Headers) (file *ObjectOpenFile, headers Headers, err error) {
	var resp *http.Response
	resp, headers, err = c.storage(RequestOpts{
		Container:  container,
		ObjectName: objectName,
		Operation:  "GET",
		ErrorMap:   objectErrorMap,
		Headers:    h,
	})
	if err != nil {
		return
	}
	// Can't check MD5 on an object with X-Object-Manifest set
	if checkHash && headers["X-Object-Manifest"] != "" {
		// log.Printf("swift: turning off md5 checking on object with manifest %v", objectName)
		checkHash = false
	}
	file = &ObjectOpenFile{
		connection: c,
		container:  container,
		objectName: objectName,
		headers:    h,
		resp:       resp,
		checkHash:  checkHash,
		body:       resp.Body,
	}
	if checkHash {
		file.hash = md5.New()
		file.body = io.TeeReader(resp.Body, file.hash)
	}
	// Read Content-Length
	if resp.Header.Get("Content-Length") != "" {
		file.length, err = getInt64FromHeader(resp, "Content-Length")
		file.lengthOk = (err == nil)
	}
	return
}

// ObjectGet gets the object into the io.Writer contents.
//
// Returns the headers of the response.
//
// If checkHash is true then it will calculate the md5sum of the file
// as it is being received and check it against that returned from the
// server.  If it is wrong then it will return ObjectCorrupted.
//
// headers["Content-Type"] will give the content type if desired.
func (c *Connection) ObjectGet(container string, objectName string, contents io.Writer, checkHash bool, h Headers) (headers Headers, err error) {
	file, headers, err := c.ObjectOpen(container, objectName, checkHash, h)
	if err != nil {
		return
	}
	defer checkClose(file, &err)
	_, err = io.Copy(contents, file)
	return
}

// ObjectGetBytes returns an object as a []byte.
//
// This is a simplified interface which checks the MD5
func (c *Connection) ObjectGetBytes(container string, objectName string) (contents []byte, err error) {
	var buf bytes.Buffer
	_, err = c.ObjectGet(container, objectName, &buf, true, nil)
	contents = buf.Bytes()
	return
}

// ObjectGetString returns an object as a string.
//
// This is a simplified interface which checks the MD5
func (c *Connection) ObjectGetString(container string, objectName string) (contents string, err error) {
	var buf bytes.Buffer
	_, err = c.ObjectGet(container, objectName, &buf, true, nil)
	contents = buf.String()
	return
}

// ObjectDelete deletes the object.
//
// May return ObjectNotFound if the object isn't found
func (c *Connection) ObjectDelete(container string, objectName string) error {
	_, _, err := c.storage(RequestOpts{
		Container:  container,
		ObjectName: objectName,
		Operation:  "DELETE",
		ErrorMap:   objectErrorMap,
	})
	return err
}

// ObjectTempUrl returns a temporary URL for an object
func (c *Connection) ObjectTempUrl(container string, objectName string, secretKey string, method string, expires time.Time) string {
	mac := hmac.New(sha1.New, []byte(secretKey))
	prefix, _ := url.Parse(c.StorageUrl)
	body := fmt.Sprintf("%s\n%d\n%s/%s/%s", method, expires.Unix(), prefix.Path, container, objectName)
	mac.Write([]byte(body))
	sig := hex.EncodeToString(mac.Sum(nil))
	return fmt.Sprintf("%s/%s/%s?temp_url_sig=%s&temp_url_expires=%d", c.StorageUrl, container, objectName, sig, expires.Unix())
}

// parseResponseStatus parses string like "200 OK" and returns Error.
//
// For status codes beween 200 and 299, this returns nil.
func parseResponseStatus(resp string, errorMap errorMap) error {
	code := 0
	reason := resp
	t := strings.SplitN(resp, " ", 2)
	if len(t) == 2 {
		ncode, err := strconv.Atoi(t[0])
		if err == nil {
			code = ncode
			reason = t[1]
		}
	}
	if errorMap != nil {
		if err, ok := errorMap[code]; ok {
			return err
		}
	}
	if 200 <= code && code <= 299 {
		return nil
	}
	return newError(code, reason)
}

// BulkDeleteResult stores results of BulkDelete().
//
// Individual errors may (or may not) be returned by Errors.
// Errors is a map whose keys are a full path of where the object was
// to be deleted, and whose values are Error objects.  A full path of
// object looks like "/API_VERSION/USER_ACCOUNT/CONTAINER/OBJECT_PATH".
type BulkDeleteResult struct {
	NumberNotFound int64            // # of objects not found.
	NumberDeleted  int64            // # of deleted objects.
	Errors         map[string]error // Mapping between object name and an error.
	Headers        Headers          // Response HTTP headers.
}

// BulkDelete deletes multiple objectNames from container in one operation.
//
// Some servers may not accept bulk-delete requests since bulk-delete is
// an optional feature of swift - these will return the Forbidden error.
//
// See also:
// * http://docs.openstack.org/trunk/openstack-object-storage/admin/content/object-storage-bulk-delete.html
// * http://docs.rackspace.com/files/api/v1/cf-devguide/content/Bulk_Delete-d1e2338.html
func (c *Connection) BulkDelete(container string, objectNames []string) (result BulkDeleteResult, err error) {
	var buffer bytes.Buffer
	for _, s := range objectNames {
		buffer.WriteString(fmt.Sprintf("/%s/%s\n", container,
			url.QueryEscape(s)))
	}
	resp, headers, err := c.storage(RequestOpts{
		Operation:  "DELETE",
		Parameters: url.Values{"bulk-delete": []string{"1"}},
		Headers: Headers{
			"Accept":       "application/json",
			"Content-Type": "text/plain",
		},
		ErrorMap: ContainerErrorMap,
		Body:     &buffer,
	})
	if err != nil {
		return
	}
	var jsonResult struct {
		NotFound int64  `json:"Number Not Found"`
		Status   string `json:"Response Status"`
		Errors   [][]string
		Deleted  int64 `json:"Number Deleted"`
	}
	err = readJson(resp, &jsonResult)
	if err != nil {
		return
	}

	err = parseResponseStatus(jsonResult.Status, objectErrorMap)
	result.NumberNotFound = jsonResult.NotFound
	result.NumberDeleted = jsonResult.Deleted
	result.Headers = headers
	el := make(map[string]error, len(jsonResult.Errors))
	for _, t := range jsonResult.Errors {
		if len(t) != 2 {
			continue
		}
		el[t[0]] = parseResponseStatus(t[1], objectErrorMap)
	}
	result.Errors = el
	return
}

// BulkUploadResult stores results of BulkUpload().
//
// Individual errors may (or may not) be returned by Errors.
// Errors is a map whose keys are a full path of where an object was
// to be created, and whose values are Error objects.  A full path of
// object looks like "/API_VERSION/USER_ACCOUNT/CONTAINER/OBJECT_PATH".
type BulkUploadResult struct {
	NumberCreated int64            // # of created objects.
	Errors        map[string]error // Mapping between object name and an error.
	Headers       Headers          // Response HTTP headers.
}

// BulkUpload uploads multiple files in one operation.
//
// uploadPath can be empty, a container name, or a pseudo-directory
// within a container.  If uploadPath is empty, new containers may be
// automatically created.
//
// Files are read from dataStream.  The format of the stream is specified
// by the format parameter.  Available formats are:
// * UploadTar       - Plain tar stream.
// * UploadTarGzip   - Gzip compressed tar stream.
// * UploadTarBzip2  - Bzip2 compressed tar stream.
//
// Some servers may not accept bulk-upload requests since bulk-upload is
// an optional feature of swift - these will return the Forbidden error.
//
// See also:
// * http://docs.openstack.org/trunk/openstack-object-storage/admin/content/object-storage-extract-archive.html
// * http://docs.rackspace.com/files/api/v1/cf-devguide/content/Extract_Archive-d1e2338.html
func (c *Connection) BulkUpload(uploadPath string, dataStream io.Reader, format string, h Headers) (result BulkUploadResult, err error) {
	extraHeaders := Headers{"Accept": "application/json"}
	for key, value := range h {
		extraHeaders[key] = value
	}
	// The following code abuses Container parameter intentionally.
	// The best fix might be to rename Container to UploadPath.
	resp, headers, err := c.storage(RequestOpts{
		Container:  uploadPath,
		Operation:  "PUT",
		Parameters: url.Values{"extract-archive": []string{format}},
		Headers:    extraHeaders,
		ErrorMap:   ContainerErrorMap,
		Body:       dataStream,
	})
	if err != nil {
		return
	}
	// Detect old servers which don't support this feature
	if headers["Content-Type"] != "application/json" {
		err = Forbidden
		return
	}
	var jsonResult struct {
		Created int64  `json:"Number Files Created"`
		Status  string `json:"Response Status"`
		Errors  [][]string
	}
	err = readJson(resp, &jsonResult)
	if err != nil {
		return
	}

	err = parseResponseStatus(jsonResult.Status, objectErrorMap)
	result.NumberCreated = jsonResult.Created
	result.Headers = headers
	el := make(map[string]error, len(jsonResult.Errors))
	for _, t := range jsonResult.Errors {
		if len(t) != 2 {
			continue
		}
		el[t[0]] = parseResponseStatus(t[1], objectErrorMap)
	}
	result.Errors = el
	return
}

// Object returns info about a single object including any metadata in the header.
//
// May return ObjectNotFound.
//
// Use headers.ObjectMetadata() to read the metadata in the Headers.
func (c *Connection) Object(container string, objectName string) (info Object, headers Headers, err error) {
	var resp *http.Response
	resp, headers, err = c.storage(RequestOpts{
		Container:  container,
		ObjectName: objectName,
		Operation:  "HEAD",
		ErrorMap:   objectErrorMap,
		NoResponse: true,
	})
	if err != nil {
		return
	}
	// Parse the headers into the struct
	// HTTP/1.1 200 OK
	// Date: Thu, 07 Jun 2010 20:59:39 GMT
	// Server: Apache
	// Last-Modified: Fri, 12 Jun 2010 13:40:18 GMT
	// ETag: 8a964ee2a5e88be344f36c22562a6486
	// Content-Length: 512000
	// Content-Type: text/plain; charset=UTF-8
	// X-Object-Meta-Meat: Bacon
	// X-Object-Meta-Fruit: Bacon
	// X-Object-Meta-Veggie: Bacon
	// X-Object-Meta-Dairy: Bacon
	info.Name = objectName
	info.ContentType = resp.Header.Get("Content-Type")
	if resp.Header.Get("Content-Length") != "" {
		if info.Bytes, err = getInt64FromHeader(resp, "Content-Length"); err != nil {
			return
		}
	}
	info.ServerLastModified = resp.Header.Get("Last-Modified")
	if info.LastModified, err = time.Parse(http.TimeFormat, info.ServerLastModified); err != nil {
		return
	}
	info.Hash = resp.Header.Get("Etag")
	return
}

// ObjectUpdate adds, replaces or removes object metadata.
//
// Add or Update keys by mentioning them in the Metadata.  Use
// Metadata.ObjectHeaders and Headers.ObjectMetadata to convert your
// Metadata to and from normal HTTP headers.
//
// This removes all metadata previously added to the object and
// replaces it with that passed in so to delete keys, just don't
// mention them the headers you pass in.
//
// Object metadata can only be read with Object() not with Objects().
//
// This can also be used to set headers not already assigned such as
// X-Delete-At or X-Delete-After for expiring objects.
//
// You cannot use this to change any of the object's other headers
// such as Content-Type, ETag, etc.
//
// Refer to copying an object when you need to update metadata or
// other headers such as Content-Type or CORS headers.
//
// May return ObjectNotFound.
func (c *Connection) ObjectUpdate(container string, objectName string, h Headers) error {
	_, _, err := c.storage(RequestOpts{
		Container:  container,
		ObjectName: objectName,
		Operation:  "POST",
		ErrorMap:   objectErrorMap,
		NoResponse: true,
		Headers:    h,
	})
	return err
}

// ObjectCopy does a server side copy of an object to a new position
//
// All metadata is preserved.  If metadata is set in the headers then
// it overrides the old metadata on the copied object.
//
// The destination container must exist before the copy.
//
// You can use this to copy an object to itself - this is the only way
// to update the content type of an object.
func (c *Connection) ObjectCopy(srcContainer string, srcObjectName string, dstContainer string, dstObjectName string, h Headers) (headers Headers, err error) {
	// Meta stuff
	extraHeaders := map[string]string{
		"Destination": dstContainer + "/" + dstObjectName,
	}
	for key, value := range h {
		extraHeaders[key] = value
	}
	_, headers, err = c.storage(RequestOpts{
		Container:  srcContainer,
		ObjectName: srcObjectName,
		Operation:  "COPY",
		ErrorMap:   objectErrorMap,
		NoResponse: true,
		Headers:    extraHeaders,
	})
	return
}

// ObjectMove does a server side move of an object to a new position
//
// This is a convenience method which calls ObjectCopy then ObjectDelete
//
// All metadata is preserved.
//
// The destination container must exist before the copy.
func (c *Connection) ObjectMove(srcContainer string, srcObjectName string, dstContainer string, dstObjectName string) (err error) {
	_, err = c.ObjectCopy(srcContainer, srcObjectName, dstContainer, dstObjectName, nil)
	if err != nil {
		return
	}
	return c.ObjectDelete(srcContainer, srcObjectName)
}

// ObjectUpdateContentType updates the content type of an object
//
// This is a convenience method which calls ObjectCopy
//
// All other metadata is preserved.
func (c *Connection) ObjectUpdateContentType(container string, objectName string, contentType string) (err error) {
	h := Headers{"Content-Type": contentType}
	_, err = c.ObjectCopy(container, objectName, container, objectName, h)
	return
}

// ------------------------------------------------------------

// VersionContainerCreate is a helper method for creating and enabling version controlled containers.
//
// It builds the current object container, the non-current object version container, and enables versioning.
//
// If the server doesn't support versioning then it will return
// Forbidden however it will have created both the containers at that point.
func (c *Connection) VersionContainerCreate(current, version string) error {
	if err := c.ContainerCreate(version, nil); err != nil {
		return err
	}
	if err := c.ContainerCreate(current, nil); err != nil {
		return err
	}
	if err := c.VersionEnable(current, version); err != nil {
		return err
	}
	return nil
}

// VersionEnable enables versioning on the current container with version as the tracking container.
//
// May return Forbidden if this isn't supported by the server
func (c *Connection) VersionEnable(current, version string) error {
	h := Headers{"X-Versions-Location": version}
	if err := c.ContainerUpdate(current, h); err != nil {
		return err
	}
	// Check to see if the header was set properly
	_, headers, err := c.Container(current)
	if err != nil {
		return err
	}
	// If failed to set versions header, return Forbidden as the server doesn't support this
	if headers["X-Versions-Location"] != version {
		return Forbidden
	}
	return nil
}

// VersionDisable disables versioning on the current container.
func (c *Connection) VersionDisable(current string) error {
	h := Headers{"X-Versions-Location": ""}
	if err := c.ContainerUpdate(current, h); err != nil {
		return err
	}
	return nil
}

// VersionObjectList returns a list of older versions of the object.
//
// Objects are returned in the format <length><object_name>/<timestamp>
func (c *Connection) VersionObjectList(version, object string) ([]string, error) {
	opts := &ObjectsOpts{
		// <3-character zero-padded hexadecimal character length><object name>/
		Prefix: fmt.Sprintf("%03x", len(object)) + object + "/",
	}
	return c.ObjectNames(version, opts)
}