1 # -*- encoding: binary -*-
3 # \MogileFS file manipulation client.
5 # Create a new instance that will communicate with these trackers:
6 # hosts = %w[192.168.1.69:6001 192.168.1.70:6001]
7 # mg = MogileFS::MogileFS.new(:domain => 'test', :hosts => hosts)
9 # # Stores "A bunch of text to store" into 'some_key' with a class of 'text'.
10 # mg.store_content('some_key', 'text', "A bunch of text to store")
12 # # Retrieve data from 'some_key' as a string
13 # data = mg.get_file_data('some_key')
15 # # Store the contents of 'image.jpeg' into the key 'my_image' with a
17 # mg.store_file('my_image', 'image', 'image.jpeg')
19 # # Store the contents of 'image.jpeg' into the key 'my_image' with a
20 # # class of 'image' using an open IO object.
21 # File.open('image.jpeg') { |fp| mg.store_file('my_image', 'image', fp) }
23 # # Retrieve the contents of 'my_image' into '/path/to/huge_file'
24 # # without slurping the entire contents into memory:
25 # File.open('/path/to/huge_file', 'w') do |fp|
26 # mg.get_file_data('my_image', fp)
29 # # Remove the key 'my_image' and 'some_key'.
30 # mg.delete('my_image')
31 # mg.delete('some_key')
33 class MogileFS::MogileFS < MogileFS::Client
34 include MogileFS::Bigfile
36 # The domain of keys for this MogileFS client.
39 # The timeout for get_file_data. Defaults to five seconds.
40 attr_accessor :get_file_data_timeout
42 # Creates a new MogileFS::MogileFS instance. +args+ must include a key
43 # :domain specifying the domain of this client.
44 def initialize(args = {})
45 @domain = args[:domain]
47 @get_file_data_timeout = 5
49 raise ArgumentError, "you must specify a domain" unless @domain
51 if @backend = args[:db_backend]
58 # Enumerates keys, limited by optional +prefix+
59 def each_key(prefix = "", &block)
62 keys, after = list_keys(prefix, after, 1000, &block)
63 end while keys && keys[0]
67 # Retrieves the contents of +key+. If +dst+ is specified, +dst+
68 # should be an IO-like object capable of receiving the +write+ method
69 # or a path name. +copy_length+ may be specified to limit the number of
70 # bytes to retrieve, and +src_offset+ can be specified to specified the
71 # start position of the copy.
72 def get_file_data(key, dst = nil, copy_length = nil, src_offset = nil)
73 paths = get_paths(key)
74 sock = MogileFS::HTTPReader.first(paths, @get_file_data_timeout,
75 copy_length, src_offset)
84 sock.close if sock && ! sock.closed?
87 # Get the paths (URLs as strings) for +key+. If +args+ is specified,
89 # - :noverify -> boolean, whether or not the tracker checks (default: true)
90 # - :pathcount -> a positive integer of URLs to retrieve (default: 2)
91 # - :zone -> "alt" or nil (default: nil)
93 # :noverify defaults to false because this client library is capable of
94 # verifying paths for readability itself. It is also faster and more
95 # reliable to verify paths on the client.
96 def get_paths(key, *args)
100 :noverify => args[0],
105 opts[:noverify] = args[:noverify]
106 opts[:zone] = args[:zone]
107 pathcount = args[:pathcount] and opts[:pathcount] = pathcount.to_i
110 opts[:noverify] = false == opts[:noverify] ? 0 : 1
111 @backend.respond_to?(:_get_paths) and return @backend._get_paths(opts)
112 res = @backend.get_paths(opts)
113 (1..res['paths'].to_i).map { |i| res["path#{i}"] }
116 # Get the URIs for +key+ (paths) as URI::HTTP objects
117 def get_uris(key, *args)
118 get_paths(key, *args).map! { |path| URI.parse(path) }
121 # Creates a new file +key+ in +klass+. +bytes+ is currently unused.
122 # Consider using store_file instead of this method for large files.
123 # This requires a block passed to it and operates like File.open.
124 # This atomically replaces existing data stored as +key+ when
125 def new_file(key, klass = nil, bytes = 0) # :yields: file
126 raise MogileFS::ReadOnlyError if readonly?
127 opts = { :domain => @domain, :key => key, :multi_dest => 1 }
128 opts[:class] = klass if klass && klass != "default"
129 res = @backend.create_open(opts)
131 dests = if dev_count = res['dev_count'] # multi_dest succeeded
132 (1..dev_count.to_i).map do |i|
133 [res["devid_#{i}"], res["path_#{i}"]]
135 else # single destination returned
136 # 0x0040: d0e4 4f4b 2064 6576 6964 3d31 2666 6964 ..OK.devid=1&fid
137 # 0x0050: 3d33 2670 6174 683d 6874 7470 3a2f 2f31 =3&path=http://1
138 # 0x0060: 3932 2e31 3638 2e31 2e37 323a 3735 3030 92.168.1.72:7500
139 # 0x0070: 2f64 6576 312f 302f 3030 302f 3030 302f /dev1/0/000/000/
140 # 0x0080: 3030 3030 3030 3030 3033 2e66 6964 0d0a 0000000003.fid..
142 [[res['devid'], res['path']]]
145 case (dests[0][1] rescue nil)
146 when /^http:\/\// then
147 http_file = MogileFS::HTTPFile.new(dests, bytes)
149 rv = http_file.commit
150 @backend.create_close(:fid => res['fid'],
151 :devid => http_file.devid,
154 :path => http_file.uri.to_s,
158 raise MogileFS::EmptyPathError,
159 "Empty path for mogile upload res=#{res.inspect}"
161 raise MogileFS::UnsupportedPathError,
162 "paths '#{dests.inspect}' returned by backend is not supported"
166 # Copies the contents of +file+ into +key+ in class +klass+. +file+ can be
167 # either a path name (String or Pathname object) or an IO-like object that
168 # responds to #read or #readpartial. Returns size of +file+ stored.
169 # This atomically replaces existing data stored as +key+
170 def store_file(key, klass, file)
171 raise MogileFS::ReadOnlyError if readonly?
173 new_file(key, klass) { |mfp| mfp.big_io = file }
176 # Stores +content+ into +key+ in class +klass+, where +content+ is a String
177 # This atomically replaces existing data stored as +key+
178 def store_content(key, klass, content)
179 raise MogileFS::ReadOnlyError if readonly?
181 new_file key, klass do |mfp|
182 if content.is_a?(MogileFS::Util::StoreContent)
183 mfp.streaming_io = content
192 raise MogileFS::ReadOnlyError if readonly?
194 @backend.delete :domain => @domain, :key => key
198 # Sleeps +duration+, only used for testing
199 def sleep(duration) # :nodoc:
200 @backend.sleep :duration => duration
203 # Renames a key +from+ to key +to+.
205 raise MogileFS::ReadOnlyError if readonly?
207 @backend.rename :domain => @domain, :from_key => from, :to_key => to
211 # Returns the size of +key+.
213 @backend.respond_to?(:_size) and return @backend._size(domain, key)
215 file_info(key)["length"].to_i
216 rescue MogileFS::Backend::UnknownCommandError
217 paths_size(get_paths(key))
221 def paths_size(paths) # :nodoc:
222 require "mogilefs/paths_size"
223 MogileFS::PathsSize.call(paths)
226 # Lists keys starting with +prefix+ following +after+ up to +limit+. If
227 # +after+ is nil the list starts at the beginning.
228 def list_keys(prefix = "", after = nil, limit = 1000, &block)
229 @backend.respond_to?(:_list_keys) and
230 return @backend._list_keys(domain, prefix, after, limit, &block)
233 res = @backend.list_keys(:domain => domain, :prefix => prefix,
234 :after => after, :limit => limit)
235 rescue MogileFS::Backend::NoneMatchError
239 keys = (1..res['key_count'].to_i).map { |i| res["key_#{i}"] }
242 keys.each { |key| block.call(key) }
244 list_keys_verbose(keys, block)
248 [ keys, res['next_after'] ]
251 def list_keys_verbose(keys, block) # :nodoc:
252 # emulate the MogileFS::Mysql interface, slowly...
255 on_file_info = lambda do |info|
256 Hash === info or raise info
257 file_info_cleanup(info)
259 # deal with trackers with multiple queryworkers responding out-of-order
260 ready[info["key"]] = info
261 while info = ready.delete(ordered[0])
262 block.call(ordered.shift, info["length"], info["devcount"])
265 opts = { :domain => @domain }
269 @backend.pipeline_dispatch(:file_info, opts, &on_file_info)
271 @backend.pipeline_wait
272 rescue MogileFS::Backend::UnknownCommandError # MogileFS < 2.45
273 @backend.shutdown # reset the socket
274 args = { :pathcount => 0x7fffffff }
276 paths = get_paths(key, args)
277 block.call(key, paths_size(paths), paths.size)
279 rescue MogileFS::PipelineError, SystemCallError,
280 MogileFS::RequestTruncatedError,
281 MogileFS::UnreadableSocketError,
282 MogileFS::InvalidResponseError, # truncated response
285 keys = ordered - ready.keys
293 # Return metadata about a file as a hash.
294 # Returns the domain, class, expected length, devcount, etc.
295 # Optionally device ids (not paths) can be returned as
296 # well if :devices is specified and +true+.
298 # This should only be used for informational purposes, and not usually
299 # for dynamically serving files.
300 def file_info(key, args = nil)
301 opts = { :domain => @domain, :key => key }
302 args and devices = args[:devices] and opts[:devices] = devices ? 1 : 0
303 file_info_cleanup(@backend.file_info(opts))
306 def file_info_cleanup(rv) # :nodoc:
307 %w(fid length devcount).each { |f| rv[f] = rv[f].to_i }
308 devids = rv["devids"] and
309 rv["devids"] = devids.split(/,/).map! { |x| x.to_i }
313 # Given an Integer +fid+ or String +key+ and domain, thorougly search
314 # the database for all occurences of a particular fid.
316 # Use this sparingly, this command hits the master database numerous
317 # times and is very expensive. This is not for production use, only
318 # troubleshooting and debugging.
320 # Searches for fid=666:
322 # client.file_debug(666)
324 # Search for key=foo using the default domain for this object:
326 # client.file_debug("foo")
328 # Search for key=foo in domain="bar":
330 # client.file_debug(:key => "foo", :domain => "bar")
334 when Integer then args = { "fid" => args }
335 when String then args = { "key" => args }
337 opts = { :domain => args[:domain] || @domain }.merge!(args)
339 rv = @backend.file_debug(opts)
342 when /_(?:classid|devcount|dmid|fid|length|
343 nexttry|fromdevid|failcount|flags|devid|type)\z/x
346 rv[k] = v.split(/,/).map! { |x| x.to_i }