lib/mogilefs/mogilefs.rb

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