class S3

# Represents a multipart upload to an S3 object. See

# {S3Object#multipart_upload} for a convenient way to initiate a

# multipart upload.

#

# Note: After you initiate multipart upload and upload one or more

# parts, you must either complete or abort multipart upload in order

# to stop getting charged for storage of the uploaded parts. Only

# after you either complete or abort multipart upload, Amazon S3

# frees up the parts storage and stops charging you for the parts

# storage.

class MultipartUpload

include Core::Model

class EmptyUploadError < StandardError; end

# @api private

def initialize(object, id, options = {})

@id = id

@object = object

super

@completed_parts = {}

@increment_mutex = Mutex.new

@completed_mutex = Mutex.new

@last_part = 0

end

def bucket

object.bucket

end

def inspect

"<#{self.class}:#{object.bucket.name}/#{object.key}:#{id}>"

end

# @return [String] Returns the upload id.

attr_reader :id

alias_method :upload_id, :id

# @return [S3Object] Returns the object this upload is intended for.

attr_reader :object

# @return [Boolean] Returns true if both multipart uploads

# represent the same object and upload.

def ==(other)

other.kind_of?(MultipartUpload) and

other.object == object and

other.id == id

end

alias_method :eql?, :==

# @return [Boolean] True if the upload exists.

def exists?

client.list_parts(base_opts)

rescue Errors::NoSuchUpload => e

false

else

true

end

# @return The upload initiator. This object will have `:id`

# and `:display_name` methods; if the initiator is an IAM

# user, the `:id` method will return the ARN of the user, and

# if the initiator is an AWS account, this method will return

# the same data as {#owner}.

def initiator

client.list_parts(base_opts).initiator

end

# @return The upload owner. This object will have `:id`

# and `:display_name` methods.

def owner

client.list_parts(base_opts).owner

end

# @return [Symbol] The class of storage used to store the

# uploaded object. Possible values:

#

# * `:standard`

# * `:reduced_redundancy?`

def storage_class

client.list_parts(base_opts).storage_class.downcase.to_sym

end

# @return [Boolean] True if the uploaded object will be stored

# with reduced redundancy.

def reduced_redundancy?

storage_class == :reduced_redundancy

end

# Aborts the upload. After a multipart upload is aborted, no

# additional parts can be uploaded using that upload ID. The

# storage consumed by any previously uploaded parts will be

# freed. However, if any part uploads are currently in

# progress, those part uploads might or might not succeed. As

# a result, it might be necessary to abort a given multipart

# upload multiple times in order to completely free all

# storage consumed by all parts.

# @return [nil]

def abort

unless aborted?

client.abort_multipart_upload(base_opts)

@aborted = true

end

nil

end

alias_method :delete, :abort

alias_method :cancel, :abort

# @return [Boolean] True if the upload has been aborted.

# @see #abort

def aborted?

@aborted

end

# Uploads a part.

#

# @overload add_part(data, options = {})

#

# @param data The data to upload.

# Valid values include:

#

# * A string

# * A Pathname object

# * Any object responding to `read` and `eof?`; the object

# must support the following access methods:

#

# read # all at once

# read(length) until eof? # in chunks

#

# If you specify data this way, you must also include

# the `:content_length` option.

#

# @param [Hash] options Additional options for the upload.

#

# @option options [Integer] :content_length If provided,

# this option must match the total number of bytes written

# to S3 during the operation. This option is required if

# `:data` is an IO-like object without a `size` method.

#

# @overload add_part(options)

#

# @param [Hash] options Options for the upload. Either

# `:data` or `:file` is required.

#

# @option options :data The data to upload. Valid values

# include:

#

# * A string

# * A Pathname object

# * Any object responding to `read` and `eof?`; the object

# must support the following access methods:

#

# read # all at once

# read(length) until eof? # in chunks

#

# If you specify data this way, you must also include

# the `:content_length` option.

#

# @option options [String] :file Can be specified instead of

# `:data`; its value specifies the path of a file to

# upload.

#

# @option options [Integer] :content_length If provided,

# this option must match the total number of bytes written

# to S3 during the operation. This option is required if

# `:data` is an IO-like object without a `size` method.

#

# @option options [Integer] :part_number The part number.

def add_part(data_or_options, options = {})

if data_or_options.kind_of?(Hash)

part_options = base_opts.merge(data_or_options)

else

part_options = base_opts.merge(:data => data_or_options)

end

part_options.merge!(options)

unless part_options[:part_number]

@increment_mutex.synchronize do

part_options[:part_number] = (@last_part += 1)

end

end

part_number = part_options[:part_number]

resp = client.upload_part(part_options)

@completed_mutex.synchronize do

@completed_parts[part_number] = {

:part_number => part_number,

:etag => resp[:etag]

}

end

UploadedPart.new(self, part_number, :etag => resp[:etag])

end

# Copies a part.

#

# @param [string] copy_source Full S3 name of source, ie bucket/key

#

# @param [Hash] options Additional options for the copy.

#

# @option options [Integer] :part_number The part number.

#

# @option options [Integer] :copy_source_range Range of bytes to copy, ie bytes=0-45687

def copy_part(copy_source, options = {})

part_options = base_opts.merge(options)

part_options.merge!(:copy_source => copy_source)

unless part_options[:part_number]

@increment_mutex.synchronize do

part_options[:part_number] = (@last_part += 1)

end

end

part_number = part_options[:part_number]

resp = client.copy_part(part_options)

@completed_mutex.synchronize do

@completed_parts[part_number] = {

:part_number => part_number,

:etag => resp[:etag]

}

end

UploadedPart.new(self, part_number, :etag => resp[:etag])

end

# Completes the upload by assembling previously uploaded

# parts.

#

# @return [S3Object, ObjectVersion] If the bucket has versioning

# enabled, returns the {ObjectVersion} representing the

# version that was uploaded. If versioning is disabled,

# returns the object.

def complete(*parts)

parts = parts.flatten

case parts.first

when :remote_parts

complete_opts = get_complete_opts

when :local_parts, nil

complete_opts = base_opts.merge(:parts => completed_parts)

else

part_numbers = parts.map do |part|

case part

when Integer

part

when UploadedPart

raise ArgumentError.new("cannot complete an upload with parts "+

"from a different upload") unless

part.upload == self

part.part_number

else

raise ArgumentError.new("expected number or UploadedPart")

end

end

complete_opts = get_complete_opts(part_numbers)

end

raise EmptyUploadError.new("Unable to complete an empty upload.") if complete_opts[:parts].empty?

resp = client.complete_multipart_upload(complete_opts)

if resp.data[:version_id]

ObjectVersion.new(object, resp.data[:version_id])

else

object

end

end

# Completes the upload or aborts it if no parts have been

# uploaded yet. Does nothing if the upload has already been

# aborted.

#

# @return [S3Object, ObjectVersion] If the bucket has versioning

# enabled, returns the {ObjectVersion} representing the

# version that was uploaded. If versioning is disabled,

# returns the object. If no upload was attempted (e.g. if it

# was aborted or if no parts were uploaded), returns `nil`.

def close

if aborted?

nil

elsif completed_parts.empty?

abort

else

complete

end

end

# @return [UploadedPartCollection] A collection representing

# the parts that have been uploaded to S3 for this upload.

def parts

UploadedPartCollection.new(self)

end

# @api private

def completed_parts

@completed_parts.values.

sort { |a, b| a[:part_number] <=> b[:part_number] }

end

# @api private

def inspect

"<#{self.class}:#{object.bucket.name}/#{object.key}:#{id}>"

end

private

def get_complete_opts(part_numbers = nil)

parts = []

self.parts.each do |part|

parts << { :part_number => part.part_number, :etag => part.etag }

end

if part_numbers

parts.reject! do |part|

!part_numbers.include?(part[:part_number])

end

end

base_opts.merge(:parts => parts)

end

def base_opts

opts = {

:bucket_name => object.bucket.name,

:key => object.key

}

opts[:upload_id] = upload_id if upload_id

opts

end

end

end