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