lib/mogilefs/httpfile.rb

   1 # -*- encoding: binary -*-
   2 require 'stringio'
   3 require 'uri'
   4 require 'mogilefs/backend'
   5 require 'mogilefs/util'
   6
   7 ##
   8 # HTTPFile wraps up the new file operations for storing files onto an HTTP
   9 # storage node.
  10 #
  11 # You really don't want to create an HTTPFile by hand.  Instead you want to
  12 # create a new file using MogileFS::MogileFS.new_file.
  13 #
  14 class MogileFS::HTTPFile < StringIO
  15   include MogileFS::Util
  16
  17   class EmptyResponseError < MogileFS::Error; end
  18   class BadResponseError < MogileFS::Error; end
  19   class UnparseableResponseError < MogileFS::Error; end
  20   class NoStorageNodesError < MogileFS::Error
  21     def message; 'Unable to open socket to storage node'; end
  22   end
  23
  24   ##
  25   # The URI this file will be stored to.
  26
  27   attr_reader :uri
  28
  29   ##
  30   # The key for this file.  This key won't represent a real file until you've
  31   # called #close.
  32
  33   attr_reader :key
  34
  35   ##
  36   # The class of this file.
  37
  38   attr_reader :class
  39
  40   ##
  41   # The big_io name in case we have file > 256M
  42
  43   attr_accessor :big_io
  44
  45   attr_accessor :streaming_io
  46
  47   ##
  48   # Works like File.open.  Use MogileFS::MogileFS#new_file instead of this
  49   # method.
  50
  51   def self.open(*args)
  52     fp = new(*args)
  53     fp.set_encoding(Encoding::BINARY) if fp.respond_to?(:set_encoding)
  54
  55     return fp unless block_given?
  56
  57     begin
  58       yield fp
  59     ensure
  60       fp.close
  61     end
  62   end
  63
  64   ##
  65   # Creates a new HTTPFile with MogileFS-specific data.  Use
  66   # MogileFS::MogileFS#new_file instead of this method.
  67
  68   def initialize(mg, fid, klass, key, dests, content_length)
  69     super ''
  70     @mg = mg
  71     @fid = fid
  72     @uri = @devid = nil
  73     @klass = klass
  74     @key = key
  75     @big_io = nil
  76     @streaming_io = nil
  77
  78     @dests = dests
  79     @tried = {}
  80
  81     @socket = nil
  82   end
  83
  84   ##
  85   # Writes an HTTP PUT request to +sock+ to upload the file and
  86   # returns file size if the socket finished writing
  87   def upload(devid, uri)
  88     file_size = length
  89     sock = MogileFS::Socket.tcp(uri.host, uri.port)
  90
  91     if @streaming_io
  92       file_size = @streaming_io.length
  93       sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
  94                  "Content-Length: #{file_size}\r\n\r\n")
  95       @streaming_io.call(Proc.new do |data_to_write|
  96         sock.write(data_to_write)
  97       end)
  98     elsif @big_io
  99       # Don't try to run out of memory
 100       File.open(@big_io, "rb") do |fp|
 101         file_size = fp.stat.size
 102         sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
 103                    "Content-Length: #{file_size}\r\n\r\n")
 104         copy_stream(fp, sock)
 105       end
 106     else
 107       sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
 108                  "Content-Length: #{length}\r\n\r\n#{string}")
 109     end
 110
 111     line = sock.gets or
 112       raise EmptyResponseError, 'Unable to read response line from server'
 113
 114     if line =~ %r%^HTTP/\d+\.\d+\s+(\d+)% then
 115       case $1.to_i
 116       when 200..299 then # success!
 117       else
 118         raise BadResponseError, "HTTP response status from upload: #{$1}"
 119       end
 120     else
 121       raise UnparseableResponseError, "Response line not understood: #{line}"
 122     end
 123
 124     @mg.backend.create_close(:fid => @fid, :devid => devid,
 125                              :domain => @mg.domain, :key => @key,
 126                              :path => uri.to_s, :size => file_size)
 127     file_size
 128   end # def upload
 129
 130   def close
 131     try_dests = @dests.dup
 132     last_err = nil
 133
 134     loop do
 135       devid, url = try_dests.shift
 136       devid && url or break
 137
 138       uri = URI.parse(url)
 139       begin
 140         bytes = upload(devid, uri)
 141         @devid, @uri = devid, uri
 142         return bytes
 143       rescue SystemCallError, Errno::ECONNREFUSED, MogileFS::Timeout,
 144              EmptyResponseError, BadResponseError,
 145              UnparseableResponseError => err
 146         last_err = @tried[url] = err
 147       end
 148     end
 149
 150     raise last_err ? last_err : NoStorageNodesError
 151   end
 152
 153 end
 154