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   # Creates a new file +key+ in +klass+.  +bytes+ is currently unused.
  90   #
  91   # The +block+ operates like File.open.
  92
  93   def new_file(key, klass = nil, bytes = 0, &block) # :yields: file
  94     raise MogileFS::ReadOnlyError if readonly?
  95     opts = { :domain => @domain, :key => key, :multi_dest => 1 }
  96     opts[:class] = klass if klass
  97     res = @backend.create_open(opts)
  98
  99     dests = if dev_count = res['dev_count'] # multi_dest succeeded
 100       (1..dev_count.to_i).map do |i|
 101         [res["devid_#{i}"], res["path_#{i}"]]
 102       end
 103     else # single destination returned
 104       # 0x0040:  d0e4 4f4b 2064 6576 6964 3d31 2666 6964  ..OK.devid=1&fid
 105       # 0x0050:  3d33 2670 6174 683d 6874 7470 3a2f 2f31  =3&path=http://1
 106       # 0x0060:  3932 2e31 3638 2e31 2e37 323a 3735 3030  92.168.1.72:7500
 107       # 0x0070:  2f64 6576 312f 302f 3030 302f 3030 302f  /dev1/0/000/000/
 108       # 0x0080:  3030 3030 3030 3030 3033 2e66 6964 0d0a  0000000003.fid..
 109
 110       [[res['devid'], res['path']]]
 111     end
 112
 113     case (dests[0][1] rescue nil)
 114     when nil, '' then
 115       raise MogileFS::EmptyPathError
 116     when /^http:\/\// then
 117       MogileFS::HTTPFile.open(self, res['fid'], klass, key,
 118                               dests, bytes, &block)
 119     else
 120       raise MogileFS::UnsupportedPathError,
 121             "paths '#{dests.inspect}' returned by backend is not supported"
 122     end
 123   end
 124
 125   ##
 126   # Copies the contents of +file+ into +key+ in class +klass+.  +file+ can be
 127   # either a file name or an object that responds to #read.
 128
 129   def store_file(key, klass, file)
 130     raise MogileFS::ReadOnlyError if readonly?
 131
 132     new_file key, klass do |mfp|
 133       if file.respond_to? :sysread then
 134         return sysrwloop(file, mfp)
 135       else
 136         if File.size(file) > 0x10000 # Bigass file, handle differently
 137           mfp.big_io = file
 138           return
 139         else
 140           return File.open(file, "rb") { |fp| sysrwloop(fp, mfp) }
 141         end
 142       end
 143     end
 144   end
 145
 146   ##
 147   # Stores +content+ into +key+ in class +klass+.
 148
 149   def store_content(key, klass, content)
 150     raise MogileFS::ReadOnlyError if readonly?
 151
 152     new_file key, klass do |mfp|
 153       if content.is_a?(MogileFS::Util::StoreContent)
 154         mfp.streaming_io = content
 155       else
 156         mfp << content
 157       end
 158     end
 159
 160     content.length
 161   end
 162
 163   ##
 164   # Removes +key+.
 165
 166   def delete(key)
 167     raise MogileFS::ReadOnlyError if readonly?
 168
 169     @backend.delete :domain => @domain, :key => key
 170   end
 171
 172   ##
 173   # Sleeps +duration+.
 174
 175   def sleep(duration)
 176     @backend.sleep :duration => duration
 177   end
 178
 179   ##
 180   # Renames a key +from+ to key +to+.
 181
 182   def rename(from, to)
 183     raise MogileFS::ReadOnlyError if readonly?
 184
 185     @backend.rename :domain => @domain, :from_key => from, :to_key => to
 186     nil
 187   end
 188
 189   ##
 190   # Returns the size of +key+.
 191   def size(key)
 192     @backend.respond_to?(:_size) and return @backend._size(domain, key)
 193     paths = get_paths(key) or return nil
 194     paths_size(paths)
 195   end
 196
 197   def paths_size(paths)
 198     paths.each do |path|
 199       begin
 200         return http_read_sock(URI.parse(path), "HEAD").mogilefs_size
 201       rescue MogileFS::InvalidResponseError, MogileFS::Timeout,
 202              Errno::ECONNREFUSED, EOFError, SystemCallError => err
 203         next
 204       end
 205     end
 206     nil
 207   end
 208
 209   ##
 210   # Lists keys starting with +prefix+ follwing +after+ up to +limit+.  If
 211   # +after+ is nil the list starts at the beginning.
 212
 213   def list_keys(prefix, after = nil, limit = 1000, &block)
 214     if @backend.respond_to?(:_list_keys)
 215       return @backend._list_keys(domain, prefix, after, limit, &block)
 216     end
 217
 218     res = begin
 219       @backend.list_keys(:domain => domain, :prefix => prefix,
 220                          :after => after, :limit => limit)
 221     rescue MogileFS::Backend::NoneMatchError
 222       return nil
 223     end
 224
 225     keys = (1..res['key_count'].to_i).map { |i| res["key_#{i}"] }
 226     if block_given?
 227       # emulate the MogileFS::Mysql interface, slowly...
 228       keys.each do |key|
 229         paths = get_paths(key) or next
 230         length = paths_size(paths) or next
 231         yield key, length, paths.size
 232       end
 233     end
 234
 235     [ keys, res['next_after'] ]
 236   end
 237
 238   protected
 239
 240     # given a URI, this returns a readable socket with ready data from the
 241     # body of the response.
 242     def http_read_sock(uri, http_method = "GET")
 243       sock = Socket.mogilefs_new_request(uri.host, uri.port,
 244                     "#{http_method} #{uri.request_uri} HTTP/1.0\r\n\r\n",
 245                     @get_file_data_timeout)
 246       buf = sock.recv_nonblock(4096, Socket::MSG_PEEK)
 247       head, body = buf.split(/\r\n\r\n/, 2)
 248
 249       # we're dealing with a seriously slow/stupid HTTP server if we can't
 250       # get the header in a single read(2) syscall.
 251       if head =~ %r{\AHTTP/\d+\.\d+\s+200\s*} &&
 252          head =~ %r{^Content-Length:\s*(\d+)}i
 253         sock.mogilefs_size = $1.to_i
 254         case http_method
 255         when "HEAD" then sock.close
 256         when "GET" then sock.recv(head.size + 4, 0)
 257         end
 258         return sock
 259       end
 260       sock.close rescue nil
 261       raise MogileFS::InvalidResponseError,
 262             "#{http_method} on #{uri} returned: #{head.inspect}"
 263     end # def http_read_sock
 264
 265 end