lib/mogilefs/new_file/content_range.rb

   1 # -*- encoding: binary -*-
   2 # here are internal implementation details, do not rely on them in your code
   3 require 'net/http'
   4 require 'mogilefs/new_file/writer'
   5
   6 # an IO-like object
   7 class MogileFS::NewFile::ContentRange
   8   include MogileFS::NewFile::Writer
   9   include MogileFS::NewFile::Common
  10   attr_reader :md5
  11
  12   # :stopdoc:
  13   begin
  14     require 'net/http/persistent'
  15     NHP = Net::HTTP::Persistent.new('mogilefs')
  16
  17     def hit(uri, req)
  18       NHP.request(uri, req).value
  19     end
  20   rescue LoadError
  21     def hit(uri, req)
  22       Net::HTTP.start(uri.host, uri.port) { |h| h.request(req).value }
  23     end
  24   end
  25   # :startdoc:
  26
  27   def initialize(dests, opts) # :nodoc:
  28     @dests = dests
  29     @opts = opts
  30     @devid = @uri = @md5 = nil
  31     @bytes_uploaded = 0
  32     @errors = []
  33   end
  34
  35   def get_dest # :nodoc:
  36     return [ @devid, @uri ] if @uri
  37     rv = @dests.shift or no_nodes!
  38     rv[1] = URI.parse(rv[1])
  39     rv
  40   end
  41
  42   def no_nodes! # :nodoc:
  43     raise NoStorageNodesError,
  44           "all paths failed with PUT: #{@errors.join(', ')}", []
  45   end
  46
  47   def request_for(uri, buf) # :nodoc:
  48     put = Net::HTTP::Put.new(uri.path)
  49     put["Content-Type"] = "application/octet-stream"
  50     put["Content-MD5"] = [ Digest::MD5.digest(buf) ].pack("m").chomp!
  51     if @bytes_uploaded > 0
  52       last_byte = @bytes_uploaded + buf.bytesize - 1
  53       put["Content-Range"] = "bytes #@bytes_uploaded-#{last_byte}/*"
  54     end
  55     put.body = buf
  56
  57     put
  58   end
  59
  60   # see IO#write
  61   def write(buf)
  62     buf = String buf
  63     len = buf.bytesize
  64     return 0 if 0 == len
  65
  66     devid, uri = get_dest
  67     put = request_for(uri, buf)
  68     begin
  69       hit(uri, put) # raises on error
  70     rescue => e
  71       raise if @bytes_uploaded > 0
  72
  73       # nothing uploaded, try another dest
  74       @errors << "#{uri.to_s} - #{e.message} (#{e.class})"
  75       devid, uri = get_dest
  76       put = request_for(uri, buf)
  77       retry
  78     end
  79
  80     @uri, @devid = uri, devid if 0 == @bytes_uploaded
  81     @bytes_uploaded += len
  82     len
  83   end
  84
  85   # called on close, do not use
  86   def commit # :nodoc:
  87     zero_byte_special if @bytes_uploaded == 0
  88
  89     create_close(@devid, @uri, @bytes_uploaded)
  90   end
  91
  92   # special case for zero-byte files :<
  93   def zero_byte_special # :nodoc:
  94     @devid, @uri = get_dest
  95     put = request_for(@uri, "")
  96     begin
  97       hit(@uri, put) # raises on error
  98     rescue => e
  99       @errors << "#{@uri.to_s} - #{e.message} (#{e.class})"
 100       @devid, @uri = get_dest
 101       put = request_for(@uri, "")
 102       retry
 103     end
 104   end
 105 end