lib/mogilefs/mogilefs.rb

   1 require 'mogilefs/client'
   2 require 'mogilefs/util'
   3
   4 ##
   5 # MogileFS File manipulation client.
   6
   7 class MogileFS::MogileFS < MogileFS::Client
   8
   9   include MogileFS::Util
  10   include MogileFS::Bigfile
  11
  12   ##
  13   # The domain of keys for this MogileFS client.
  14
  15   attr_reader :domain
  16
  17   ##
  18   # The timeout for get_file_data.  Defaults to five seconds.
  19
  20   attr_accessor :get_file_data_timeout
  21
  22   ##
  23   # Creates a new MogileFS::MogileFS instance.  +args+ must include a key
  24   # :domain specifying the domain of this client.
  25
  26   def initialize(args = {})
  27     @domain = args[:domain]
  28
  29     @get_file_data_timeout = 5
  30
  31     raise ArgumentError, "you must specify a domain" unless @domain
  32
  33     if @backend = args[:db_backend]
  34       @readonly = true
  35     else
  36       super
  37     end
  38   end
  39
  40   ##
  41   # Enumerates keys starting with +key+.
  42
  43   def each_key(prefix)
  44     after = nil
  45
  46     keys, after = list_keys prefix
  47
  48     until keys.nil? or keys.empty? do
  49       keys.each { |k| yield k }
  50       keys, after = list_keys prefix, after
  51     end
  52
  53     nil
  54   end
  55
  56   ##
  57   # Retrieves the contents of +key+.
  58
  59   def get_file_data(key, &block)
  60     paths = get_paths(key) or return nil
  61     paths.each do |path|
  62       begin
  63         sock = http_read_sock(URI.parse(path))
  64         begin
  65           return yield(sock) if block_given?
  66           return sysread_full(sock, sock.mogilefs_size, @get_file_data_timeout)
  67         ensure
  68           sock.close rescue nil
  69         end
  70       rescue MogileFS::Timeout, MogileFS::InvalidResponseError,
  71              Errno::ECONNREFUSED, EOFError, SystemCallError
  72       end
  73     end
  74     nil
  75   end
  76
  77   ##
  78   # Get the paths for +key+.
  79
  80   def get_paths(key, noverify = true, zone = nil)
  81     opts = { :domain => @domain, :key => key,
  82              :noverify => noverify ? 1 : 0, :zone => zone }
  83     @backend.respond_to?(:_get_paths) and return @backend._get_paths(opts)
  84     res = @backend.get_paths(opts)
  85     (1..res['paths'].to_i).map { |i| res["path#{i}"] }.compact
  86   end
  87
  88   ##
  89   # Get the URIs for +key+.
  90
  91   def get_uris(key, noverify = true, zone = nil)
  92     get_paths(key, noverify, zone).map { |path| URI.parse(path) }
  93   end
  94
  95   ##
  96   # Creates a new file +key+ in +klass+.  +bytes+ is currently unused.
  97   #
  98   # The +block+ operates like File.open.
  99
 100   def new_file(key, klass = nil, bytes = 0, &block) # :yields: file
 101     raise MogileFS::ReadOnlyError if readonly?
 102     opts = { :domain => @domain, :key => key, :multi_dest => 1 }
 103     opts[:class] = klass if klass
 104     res = @backend.create_open(opts)
 105
 106     dests = if dev_count = res['dev_count'] # multi_dest succeeded
 107       (1..dev_count.to_i).map do |i|
 108         [res["devid_#{i}"], res["path_#{i}"]]
 109       end
 110     else # single destination returned
 111       # 0x0040:  d0e4 4f4b 2064 6576 6964 3d31 2666 6964  ..OK.devid=1&fid
 112       # 0x0050:  3d33 2670 6174 683d 6874 7470 3a2f 2f31  =3&path=http://1
 113       # 0x0060:  3932 2e31 3638 2e31 2e37 323a 3735 3030  92.168.1.72:7500
 114       # 0x0070:  2f64 6576 312f 302f 3030 302f 3030 302f  /dev1/0/000/000/
 115       # 0x0080:  3030 3030 3030 3030 3033 2e66 6964 0d0a  0000000003.fid..
 116
 117       [[res['devid'], res['path']]]
 118     end
 119
 120     case (dests[0][1] rescue nil)
 121     when nil, '' then
 122       raise MogileFS::EmptyPathError
 123     when /^http:\/\// then
 124       MogileFS::HTTPFile.open(self, res['fid'], klass, key,
 125                               dests, bytes, &block)
 126     else
 127       raise MogileFS::UnsupportedPathError,
 128             "paths '#{dests.inspect}' returned by backend is not supported"
 129     end
 130   end
 131
 132   ##
 133   # Copies the contents of +file+ into +key+ in class +klass+.  +file+ can be
 134   # either a file name or an object that responds to #read.
 135
 136   def store_file(key, klass, file)
 137     raise MogileFS::ReadOnlyError if readonly?
 138
 139     new_file key, klass do |mfp|
 140       if file.respond_to? :sysread then
 141         return sysrwloop(file, mfp)
 142       else
 143         if File.size(file) > 0x10000 # Bigass file, handle differently
 144           mfp.big_io = file
 145           return
 146         else
 147           return File.open(file, "rb") { |fp| sysrwloop(fp, mfp) }
 148         end
 149       end
 150     end
 151   end
 152
 153   ##
 154   # Stores +content+ into +key+ in class +klass+.
 155
 156   def store_content(key, klass, content)
 157     raise MogileFS::ReadOnlyError if readonly?
 158
 159     new_file key, klass do |mfp|
 160       if content.is_a?(MogileFS::Util::StoreContent)
 161         mfp.streaming_io = content
 162       else
 163         mfp << content
 164       end
 165     end
 166
 167     content.length
 168   end
 169
 170   ##
 171   # Removes +key+.
 172
 173   def delete(key)
 174     raise MogileFS::ReadOnlyError if readonly?
 175
 176     @backend.delete :domain => @domain, :key => key
 177   end
 178
 179   ##
 180   # Sleeps +duration+.
 181
 182   def sleep(duration)
 183     @backend.sleep :duration => duration
 184   end
 185
 186   ##
 187   # Renames a key +from+ to key +to+.
 188
 189   def rename(from, to)
 190     raise MogileFS::ReadOnlyError if readonly?
 191
 192     @backend.rename :domain => @domain, :from_key => from, :to_key => to
 193     nil
 194   end
 195
 196   ##
 197   # Returns the size of +key+.
 198   def size(key)
 199     @backend.respond_to?(:_size) and return @backend._size(domain, key)
 200     paths = get_paths(key) or return nil
 201     paths_size(paths)
 202   end
 203
 204   def paths_size(paths)
 205     paths.each do |path|
 206       begin
 207         return http_read_sock(URI.parse(path), "HEAD").mogilefs_size
 208       rescue MogileFS::InvalidResponseError, MogileFS::Timeout,
 209              Errno::ECONNREFUSED, EOFError, SystemCallError => err
 210         next
 211       end
 212     end
 213     nil
 214   end
 215
 216   ##
 217   # Lists keys starting with +prefix+ follwing +after+ up to +limit+.  If
 218   # +after+ is nil the list starts at the beginning.
 219
 220   def list_keys(prefix, after = nil, limit = 1000, &block)
 221     if @backend.respond_to?(:_list_keys)
 222       return @backend._list_keys(domain, prefix, after, limit, &block)
 223     end
 224
 225     res = begin
 226       @backend.list_keys(:domain => domain, :prefix => prefix,
 227                          :after => after, :limit => limit)
 228     rescue MogileFS::Backend::NoneMatchError
 229       return nil
 230     end
 231
 232     keys = (1..res['key_count'].to_i).map { |i| res["key_#{i}"] }
 233     if block_given?
 234       # emulate the MogileFS::Mysql interface, slowly...
 235       keys.each do |key|
 236         paths = get_paths(key) or next
 237         length = paths_size(paths) or next
 238         yield key, length, paths.size
 239       end
 240     end
 241
 242     [ keys, res['next_after'] ]
 243   end
 244
 245   protected
 246
 247     # given a URI, this returns a readable socket with ready data from the
 248     # body of the response.
 249     def http_read_sock(uri, http_method = "GET")
 250       sock = Socket.mogilefs_new_request(uri.host, uri.port,
 251                     "#{http_method} #{uri.request_uri} HTTP/1.0\r\n\r\n",
 252                     @get_file_data_timeout)
 253       buf = sock.recv_nonblock(4096, Socket::MSG_PEEK)
 254       head, body = buf.split(/\r\n\r\n/, 2)
 255
 256       # we're dealing with a seriously slow/stupid HTTP server if we can't
 257       # get the header in a single read(2) syscall.
 258       if head =~ %r{\AHTTP/\d+\.\d+\s+200\s*} &&
 259          head =~ %r{^Content-Length:\s*(\d+)}i
 260         sock.mogilefs_size = $1.to_i
 261         case http_method
 262         when "HEAD" then sock.close
 263         when "GET" then sock.recv(head.size + 4, 0)
 264         end
 265         return sock
 266       end
 267       sock.close rescue nil
 268       raise MogileFS::InvalidResponseError,
 269             "#{http_method} on #{uri} returned: #{head.inspect}"
 270     end # def http_read_sock
 271
 272 end