diff --git a/lib/qiniu/resumable_upload.rb b/lib/qiniu/resumable_upload.rb new file mode 100755 index 0000000..290458f --- /dev/null +++ b/lib/qiniu/resumable_upload.rb @@ -0,0 +1,307 @@ +# -*- encoding: utf-8 -*- + +require 'zlib' +require 'yaml' +require 'tmpdir' +require 'fileutils' +require 'mime/types' +require 'digest/sha1' +require 'qiniu/abstract' +require 'qiniu/exceptions' +require 'qiniu/io' + +module Qiniu + module Storage + + module AbstractClass + class ChunkProgressNotifier + include Qiniu::Abstract + abstract_methods :notify + # def notify(block_index, block_put_progress); end + end + + class BlockProgressNotifier + include Qiniu::Abstract + abstract_methods :notify + # def notify(block_index, checksum); end + end + end # module AbstractClass + + class ChunkProgressNotifier < AbstractClass::ChunkProgressNotifier + def notify(index, progress) + logmsg = "chunk #{progress[:offset]/Config.settings[:chunk_size]} in block #{index} successfully uploaded.\n" + progress.to_s + Utils.debug(logmsg) + end + end # class ChunkProgressNotifier + + class BlockProgressNotifier < AbstractClass::BlockProgressNotifier + def notify(index, checksum) + Utils.debug "block #{index}: {ctx: #{checksum}} successfully uploaded." + Utils.debug "block #{index}: {checksum: #{checksum}} successfully uploaded." + end + end # class BlockProgressNotifier + + class << self + include Utils + + def resumable_upload_with_token(uptoken, + local_file, + bucket, + key = nil, + mime_type = nil, + custom_meta = nil, + customer = nil, + callback_params = nil, + rotate = nil) + begin + ifile = File.open(local_file, 'rb') + fh = FileData.new(ifile) + fsize = fh.data_size + key = Digest::SHA1.hexdigest(local_file + fh.mtime.to_s) if key.nil? + if mime_type.nil? || mime_type.empty? + mime = MIME::Types.type_for local_file + mime_type = mime.empty? ? 'application/octet-stream' : mime[0].content_type + end + code, data = _resumable_upload(uptoken, fh, fsize, bucket, key, mime_type, custom_meta, customer, callback_params, rotate) + [code, data] + ensure + ifile.close unless ifile.nil? + end + end # resumable_upload_with_token + + private + + class FileData + attr_accessor :fh + def initialize(fh) + @fh = fh + end + def data_size + @fh.stat.size + end + def get_data(offset, length) + @fh.seek(offset) + @fh.read(length) + end + def path + @fh.path + end + def mtime + @fh.mtime + end + #delegate :path, :mtime, :to => :fh + end # class FileData + + def _new_block_put_progress_data + {:ctx => nil, :offset => 0, :restsize => nil, :status_code => nil, :host => nil} + end # _new_block_put_progress_data + + def _call_binary_with_token(uptoken, url, data, content_type = nil, retry_times = 0) + options = { + :method => :post, + :content_type => 'application/octet-stream', + :upload_signature_token => uptoken + } + options[:content_type] = content_type if !content_type.nil? && !content_type.empty? + code, data = http_request url, data, options + unless Utils.is_response_ok?(code) + retry_times += 1 + if Config.settings[:auto_reconnect] && retry_times < Config.settings[:max_retry_times] + return _call_binary_with_token(uptoken, url, data, options[:content_type], retry_times) + end + end + [code, data] + end # _call_binary_with_token + + def _mkblock(uptoken, block_size, body) + url = Config.settings[:up_host] + "/mkblk/#{block_size}" + _call_binary_with_token(uptoken, url, body) + end # _mkblock + + def _putblock(uphost, uptoken, ctx, offset, body) + url = uphost + "/bput/#{ctx}/#{offset}" + _call_binary_with_token(uptoken, url, body) + end # _putblock + + def _resumable_put_block(uptoken, + fh, + block_index, + block_size, + chunk_size, + progress, + retry_times, + notifier) + code, data = 0, {} + fpath = fh.path + + # this block has never been uploaded. + if progress[:ctx] == nil || progress[:ctx].empty? + progress[:offset] = 0 + progress[:restsize] = block_size + # choose the smaller one + body_length = [block_size, chunk_size].min + for i in 1..retry_times + seek_pos = block_index*Config.settings[:block_size] + body = fh.get_data(seek_pos, body_length) + result_length = body.length + if result_length != body_length + raise FileSeekReadError.new(fpath, block_index, seek_pos, body_length, result_length) + end + code, data = _mkblock(uptoken, block_size, body) + body_crc32 = Zlib.crc32(body) + if Utils.is_response_ok?(code) && data["crc32"] == body_crc32 + progress[:ctx] = data["ctx"] + progress[:offset] = body_length + progress[:restsize] = block_size - body_length + progress[:status_code] = code + progress[:host] = data["host"] + if !notifier.nil? && notifier.respond_to?("notify") + notifier.notify(block_index, progress) + end + break + elsif i == retry_times && data["crc32"] != body_crc32 + Log.logger.error %Q(Uploading block error. Expected crc32: #{body_crc32}, but got: #{data["crc32"]}) + end + end + elsif progress[:offset] + progress[:restsize] != block_size + raise BlockSizeNotMathchError.new(fpath, block_index, progress[:offset], progress[:restsize], block_size) + end + + # loop uploading other chunks except the first one + while progress[:restsize].to_i > 0 && progress[:restsize] < block_size + # choose the smaller one + body_length = [progress[:restsize], chunk_size].min + for i in 1..retry_times + seek_pos = block_index*Config.settings[:block_size] + progress[:offset] + body = fh.get_data(seek_pos, body_length) + result_length = body.length + if result_length != body_length + raise FileSeekReadError.new(fpath, block_index, seek_pos, body_length, result_length) + end + code, data = _putblock(progress[:host], uptoken, progress[:ctx], progress[:offset], body) + body_crc32 = Zlib.crc32(body) + if Utils.is_response_ok?(code) && data["crc32"] == body_crc32 + progress[:ctx] = data["ctx"] + progress[:offset] += body_length + progress[:restsize] -= body_length + progress[:status_code] = code + progress[:host] = data["host"] + if !notifier.nil? && notifier.respond_to?("notify") + notifier.notify(block_index, progress) + end + break + elsif i == retry_times && data["crc32"] != body_crc32 + Log.logger.error %Q(Uploading block error. Expected crc32: #{body_crc32}, but got: #{data["crc32"]}) + end + end + end + # return + return [code, data] + end # _resumable_put_block + + def _block_count(fsize) + ((fsize + Config.settings[:block_size] - 1) / Config.settings[:block_size]).to_i + end # _block_count + + def _resumable_put(uptoken, + fh, + checksums, + progresses, + block_notifier = nil, + chunk_notifier = nil) + code, data = 0, {} + fsize = fh.data_size + block_count = _block_count(fsize) + checksum_count = checksums.length + progress_count = progresses.length + if checksum_count != block_count || progress_count != block_count + raise BlockCountNotMathchError.new(fh.path, block_count, checksum_count, progress_count) + end + 0.upto(block_count-1).each do |block_index| + if checksums[block_index].nil? || checksums[block_index].empty? + block_size = Config.settings[:block_size] + if block_index == block_count - 1 + block_size = fsize - block_index*Config.settings[:block_size] + end + if progresses[block_index].nil? + progresses[block_index] = _new_block_put_progress_data + end + #code, data = _resumable_put_block(uptoken, fh, block_index, block_size, Config.settings[:chunk_size], progresses[block_index], Config.settings[:max_retry_times], chunk_notifier) + # Put the whole block as a chunk + code, data = _resumable_put_block(uptoken, fh, block_index, block_size, block_size, progresses[block_index], Config.settings[:max_retry_times], chunk_notifier) + if Utils.is_response_ok?(code) + #checksums[block_index] = data["checksum"] + checksums[block_index] = data["ctx"] + if !block_notifier.nil? && block_notifier.respond_to?("notify") + block_notifier.notify(block_index, checksums[block_index]) + end + end + end + end + return [code, data] + end # _resumable_put + + def _mkfile(uphost, + uptoken, + entry_uri, + fsize, + checksums, + mime_type = nil, + custom_meta = nil, + customer = nil, + callback_params = nil, + rotate = nil) + path = '/rs-mkfile/' + Utils.urlsafe_base64_encode(entry_uri) + "/fsize/#{fsize}" + path += '/mimeType/' + Utils.urlsafe_base64_encode(mime_type) if !mime_type.nil? && !mime_type.empty? + path += '/meta/' + Utils.urlsafe_base64_encode(custom_meta) if !custom_meta.nil? && !custom_meta.empty? + path += '/customer/' + customer if !customer.nil? && !customer.empty? + callback_query_string = Utils.generate_query_string(callback_params) if !callback_params.nil? && !callback_params.empty? + path += '/params/' + Utils.urlsafe_base64_encode(callback_query_string) if !callback_query_string.nil? && !callback_query_string.empty? + path += '/rotate/' + rotate if !rotate.nil? && rotate.to_i >= 0 + url = uphost + path + #body = '' + #checksums.each do |checksum| + # body += Utils.urlsafe_base64_decode(checksum) + #end + body = checksums.join(',') + _call_binary_with_token(uptoken, url, body, 'text/plain') + end # _mkfile + + def _resumable_upload(uptoken, + fh, + fsize, + bucket, + key, + mime_type = nil, + custom_meta = nil, + customer = nil, + callback_params = nil, + rotate = nil) + + block_count = _block_count(fsize) + + chunk_notifier = ChunkProgressNotifier.new() + block_notifier = BlockProgressNotifier.new() + + progresses = [] + block_count.times{progresses << _new_block_put_progress_data} + checksums = [] + block_count.times{checksums << ''} + + code, data = _resumable_put(uptoken, fh, checksums, progresses, block_notifier, chunk_notifier) + + if Utils.is_response_ok?(code) + uphost = data["host"] + entry_uri = bucket + ':' + key + code, data = _mkfile(uphost, uptoken, entry_uri, fsize, checksums, mime_type, custom_meta, customer, callback_params, rotate) + end + + if Utils.is_response_ok?(code) + Utils.debug "File #{fh.path} {size: #{fsize}} successfully uploaded." + end + + [code, data] + end # _resumable_upload + end # self class + end # module Storage +end # module Qiniu diff --git a/lib/qiniu/storage.rb b/lib/qiniu/storage.rb new file mode 100755 index 0000000..c10da81 --- /dev/null +++ b/lib/qiniu/storage.rb @@ -0,0 +1,4 @@ +# -*- encoding: utf-8 -*- + +require 'qiniu/upload' +require 'qiniu/resumable_upload' diff --git a/lib/qiniu/upload.rb b/lib/qiniu/upload.rb new file mode 100755 index 0000000..73f9d98 --- /dev/null +++ b/lib/qiniu/upload.rb @@ -0,0 +1,114 @@ +# -*- encoding: utf-8 -*- + +module Qiniu + module Storage + class << self + include Utils + + def put_file(local_file, + bucket, + key = nil, + mime_type = nil, + custom_meta = nil, + enable_crc32_check = false) + action_params = _generate_action_params( + local_file, + bucket, + key, + mime_type, + custom_meta, + enable_crc32_check + ) + + url = Config.settings[:io_host] + action_params + options = {:content_type => 'application/octet-stream'} + + Auth.request url, ::IO.read(local_file), options + end # put_file + + def upload_with_token(uptoken, + local_file, + bucket, + key = nil, + mime_type = nil, + custom_meta = nil, + callback_params = nil, + enable_crc32_check = false, + rotate = nil) + action_params = _generate_action_params( + local_file, + bucket, + key, + mime_type, + custom_meta, + enable_crc32_check, + rotate + ) + + if callback_params.nil? + callback_params = {:bucket => bucket, :key => key, :mime_type => mime_type} + end + callback_query_string = Utils.generate_query_string(callback_params) + url = Config.settings[:up_host] + '/upload' + + Utils.upload_multipart_data(url, local_file, action_params, callback_query_string, uptoken) + end # upload_with_token + + def upload_with_token_2(uptoken, + local_file, + key = nil, + x_vars = nil) + ### 构造URL + url = Config.settings[:up_host] + url[/\/*$/] = '' + url += '/' + + ### 构造HTTP Body + post_data = { + :file => File.new(local_file, 'rb'), + :multipart => true, + } + if not uptoken.nil? + post_data[:token] = uptoken + end + if not key.nil? + post_data[:key] = key + end + if x_vars.is_a?(Hash) + post_data.merge!(x_vars) + end + + ### 发送请求 + Utils.http_request url, post_data + end # upload_with_token_2 + + private + def _generate_action_params(local_file, + bucket, + key = nil, + mime_type = nil, + custom_meta = nil, + enable_crc32_check = false, + rotate = nil) + raise NoSuchFileError, local_file unless File.exist?(local_file) + + if key.nil? + key = Digest::SHA1.hexdigest(local_file + Time.now.to_s) + end + + entry_uri = bucket + ':' + key + if mime_type.nil? || mime_type.empty? + mime = MIME::Types.type_for local_file + mime_type = mime.empty? ? 'application/octet-stream' : mime[0].content_type + end + + action_params = '/rs-put/' + Utils.urlsafe_base64_encode(entry_uri) + '/mimeType/' + Utils.urlsafe_base64_encode(mime_type) + action_params += '/meta/' + Utils.urlsafe_base64_encode(custom_meta) unless custom_meta.nil? + action_params += '/crc32/' + Utils.crc32checksum(local_file).to_s if enable_crc32_check + action_params += '/rotate/' + rotate if !rotate.nil? && rotate.to_i >= 0 + action_params + end # _generate_action_params + + end # class << self + end # module Storage +end # module Qiniu diff --git a/spec/qiniu/storage_spec.rb b/spec/qiniu/storage_spec.rb new file mode 100755 index 0000000..412d109 --- /dev/null +++ b/spec/qiniu/storage_spec.rb @@ -0,0 +1,184 @@ +# -*- encoding: utf-8 -*- + +require 'spec_helper' +require 'qiniu/auth' +require 'qiniu/storage' +require 'digest/sha1' + +module Qiniu + module Storage + describe Storage do + + before :all do + @bucket = 'RubySdkTest' + (Time.now.to_i+rand(1000)).to_s + @key = Digest::SHA1.hexdigest((Time.now.to_i+rand(100)).to_s) + + @localfile1 = "bigfile.txt" + File.open(@localfile1, "w"){|f| 5242888.times{ f.write(rand(9).to_s) }} + @key1 = Digest::SHA1.hexdigest(@localfile1+Time.now.to_s) + + @localfile2 = "bigfile2.txt" + File.open(@localfile2, "w"){|f| (1 << 22).times{ f.write(rand(9).to_s) }} + @key2 = Digest::SHA1.hexdigest(@localfile2+Time.now.to_s) + + @localfile3 = "bigfile3.txt" + File.open(@localfile3, "w"){|f| (1 << 23).times{ f.write(rand(9).to_s) }} + @key3 = Digest::SHA1.hexdigest(@localfile3+Time.now.to_s) + + @localfile4 = "smallfile.txt" + File.open(@localfile4, "w"){|f| (1 << 20).times{ f.write(rand(9).to_s) }} + @key4 = Digest::SHA1.hexdigest(@localfile4+Time.now.to_s) + + result = Qiniu.mkbucket(@bucket) + puts result.inspect + result.should be_true + end + + after :all do + File.unlink(@localfile1) if File.exists?(@localfile1) + File.unlink(@localfile2) if File.exists?(@localfile2) + File.unlink(@localfile3) if File.exists?(@localfile3) + File.unlink(@localfile4) if File.exists?(@localfile4) + + result = Qiniu.drop(@bucket) + puts result.inspect + result.should_not be_false + end + + context ".put_file" do + it "should works" do + code, data = Qiniu::Storage.put_file(__FILE__, @bucket, @key, 'application/x-ruby', 'customMeta', true) + code.should == 200 + puts data.inspect + end + end + + context ".upload_with_token" do + it "should works" do + upopts = {:scope => @bucket, :expires_in => 3600, :customer => "why404@gmail.com"} + uptoken = Qiniu.generate_upload_token(upopts) + code, data = Qiniu::Storage.upload_with_token(uptoken, __FILE__, @bucket, @key, nil, nil, nil, true) + code.should == 200 + puts data.inspect + end + end + + context ".upload_with_token_2" do + it "should works" do + upopts = {:scope => @bucket, :expires_in => 3600, :endUser => "why404@gmail.com"} + uptoken = Qiniu.generate_upload_token(upopts) + + code, data = Qiniu::Storage.upload_with_token_2(uptoken, __FILE__, @key) + + code.should == 200 + puts data.inspect + end + end # .upload_with_token_2 + + context ".resumable_upload_with_token" do + it "should works" do + upopts = {:scope => @bucket, :expires_in => 3600, :customer => "why404@gmail.com"} + uptoken = Qiniu.generate_upload_token(upopts) + code, data = Qiniu::Storage.resumable_upload_with_token(uptoken, @localfile1, @bucket, @key1) + puts data.inspect + (code/100).should == 2 + end + end + + context ".stat" do + it "should exists" do + code, data = Qiniu::RS.stat(@bucket, @key1) + puts data.inspect + code.should == 200 + end + end + + context ".delete" do + it "should works" do + code, data = Qiniu::RS.delete(@bucket, @key1) + puts data.inspect + code.should == 200 + end + end + + context ".resumable_upload_with_token2" do + it "should works" do + upopts = {:scope => @bucket, :expires_in => 3600, :customer => "why404@gmail.com"} + uptoken = Qiniu.generate_upload_token(upopts) + code, data = Qiniu::Storage.resumable_upload_with_token(uptoken, @localfile2, @bucket, @key2) + puts data.inspect + (code/100).should == 2 + end + end + + context ".stat" do + it "should exists" do + code, data = Qiniu::RS.stat(@bucket, @key2) + puts data.inspect + code.should == 200 + end + end + + context ".delete" do + it "should works" do + code, data = Qiniu::RS.delete(@bucket, @key2) + puts data.inspect + code.should == 200 + end + end + + context ".resumable_upload_with_token3" do + it "should works" do + upopts = {:scope => @bucket, :expires_in => 3600, :customer => "why404@gmail.com"} + uptoken = Qiniu.generate_upload_token(upopts) + code, data = Qiniu::Storage.resumable_upload_with_token(uptoken, @localfile3, @bucket, @key3) + puts data.inspect + (code/100).should == 2 + end + end + + context ".stat" do + it "should exists" do + code, data = Qiniu::RS.stat(@bucket, @key3) + puts data.inspect + code.should == 200 + end + end + + context ".delete" do + it "should works" do + code, data = Qiniu::RS.delete(@bucket, @key3) + puts data.inspect + code.should == 200 + end + end + + context ".resumable_upload_with_token4" do + it "should works" do + upopts = {:scope => @bucket, :expires_in => 3600, :customer => "why404@gmail.com"} + uptoken = Qiniu.generate_upload_token(upopts) + code, data = Qiniu::Storage.resumable_upload_with_token(uptoken, @localfile4, @bucket, @key4) + puts data.inspect + (code/100).should == 2 + end + end + + context ".stat" do + it "should exists" do + code, data = Qiniu::RS.stat(@bucket, @key4) + puts data.inspect + code.should == 200 + end + end + + context ".delete" do + it "should works" do + code, data = Qiniu::RS.delete(@bucket, @key4) + puts data.inspect + code.should == 200 + end + end + + end + end # module Storage +end # module Qiniu