Class: Arv::Collection::CollectionStream

Inherits:
CollectionItem show all
Defined in:
lib/arvados/collection.rb

Direct Known Subclasses

CollectionRoot

Instance Attribute Summary

Attributes inherited from CollectionItem

#name, #path

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path) ⇒ CollectionStream

Returns a new instance of CollectionStream.



272
273
274
275
 
# File 'lib/arvados/collection.rb', line 272

def initialize(path)
  super
  @items = {}
end

Class Method Details

.human_name ⇒ Object 



277
278
279
 
# File 'lib/arvados/collection.rb', line 277

def self.human_name
  "stream"
end

Instance Method Details

#[](key) ⇒ Object 



289
290
291
292
 
# File 'lib/arvados/collection.rb', line 289

def [](key)
  items[key] or
    raise Errno::ENOENT.new("%p not found in %p" % [key, path])
end

#add_copy(src_item, key) ⇒ Object 



371
372
373
374
375
376
377
 
# File 'lib/arvados/collection.rb', line 371

def add_copy(src_item, key)
  if key == "."
    self[key] = src_item.copy_named("#{path}")
  else
    self[key] = src_item.copy_named("#{path}/#{key}")
  end
end

#check_can_add_copy(src_item, key) ⇒ Object 



358
359
360
361
362
 
# File 'lib/arvados/collection.rb', line 358

def check_can_add_copy(src_item, key)
  if existing = check_can_merge(src_item, key) and not existing.leaf?
    raise Errno::ENOTEMPTY.new(existing.path)
  end
end

#check_can_merge(src_item, key) ⇒ Object 



364
365
366
367
368
369
 
# File 'lib/arvados/collection.rb', line 364

def check_can_merge(src_item, key)
  if existing = items[key] and (existing.class != src_item.class)
    raise Errno::ENOTDIR.new(existing.path)
  end
  existing
end

#copy_named(copy_path) ⇒ Object 



405
406
407
408
409
410
411
 
# File 'lib/arvados/collection.rb', line 405

def copy_named(copy_path)
  copy = self.class.new(copy_path)
  items.each_pair do |key, item|
    copy.add_copy(item, key)
  end
  copy
end

#delete(name, opts = {}) ⇒ Object 



294
295
296
297
298
299
300
301
 
# File 'lib/arvados/collection.rb', line 294

def delete(name, opts={})
  item = self[name]
  if item.file? or opts[:recursive]
    items.delete(name)
  else
    raise Errno::EISDIR.new(path)
  end
end

#each_file_path ⇒ Object 



303
304
305
306
307
308
309
310
311
312
 
# File 'lib/arvados/collection.rb', line 303

def each_file_path
  return to_enum(__method__) unless block_given?
  items.each_value do |item|
    if item.file?
      yield item.path
    else
      item.each_file_path { |path| yield path }
    end
  end
end

#file? ⇒ Boolean

Returns:

  • (Boolean) 


281
282
283
 
# File 'lib/arvados/collection.rb', line 281

def file?
  false
end

#file_at(find_path) ⇒ Object 



333
334
335
336
337
338
339
340
 
# File 'lib/arvados/collection.rb', line 333

def file_at(find_path)
  stream_path, _, file_name = find_path.rpartition("/")
  if stream_path.empty?
    get_or_new(file_name, CollectionFile, Errno::EISDIR)
  else
    stream_at(stream_path).file_at(file_name)
  end
end

#find(find_path) ⇒ Object 



314
315
316
317
318
319
320
321
 
# File 'lib/arvados/collection.rb', line 314

def find(find_path)
  # Given a POSIX-style path, return the CollectionStream that
  # contains the object at that path, and the name of the object
  # inside it.
  components = find_path.split("/")
  tail = components.pop
  [components.reduce(self, :[]), tail]
end

#leaf? ⇒ Boolean

Returns:

  • (Boolean) 


285
286
287
 
# File 'lib/arvados/collection.rb', line 285

def leaf?
  items.empty?
end

#manifest_text ⇒ Object 



342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
 
# File 'lib/arvados/collection.rb', line 342

def manifest_text
  # Return a string with the normalized manifest text for this stream,
  # including all substreams.
  file_keys, stream_keys = items.keys.sort.partition do |key|
    items[key].file?
  end
  my_line = StreamManifest.new(path)
  file_keys.each do |file_name|
    my_line.add_file(items[file_name])
  end
  sub_lines = stream_keys.map do |sub_name|
    items[sub_name].manifest_text
  end
  my_line.to_s + sub_lines.join("")
end

#merge(src_item, key) ⇒ Object 



379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
 
# File 'lib/arvados/collection.rb', line 379

def merge(src_item, key)
  # Do a recursive copy of the collection item `src_item` to destination
  # `key`.  If a simple copy is safe, do that; otherwise, recursively
  # merge the contents of the stream `src_item` into the stream at
  # `key`.
  begin
    check_can_add_copy(src_item, key)
    add_copy(src_item, key)
  rescue Errno::ENOTEMPTY
    dest = self[key]
    error = nil
    # Copy as much as possible, then raise any error encountered.
    # Start with streams for a depth-first merge.
    src_items = src_item.items.each_pair.sort_by do |_, sub_item|
      (sub_item.file?) ? 1 : 0
    end
    src_items.each do |sub_key, sub_item|
      begin
        dest.merge(sub_item, sub_key)
      rescue Errno::ENOTDIR => error
      end
    end
    raise error unless error.nil?
  end
end

#stream_at(find_path) ⇒ Object 



323
324
325
326
327
328
329
330
331
 
# File 'lib/arvados/collection.rb', line 323

def stream_at(find_path)
  key, rest = find_path.split("/", 2)
  next_stream = get_or_new(key, CollectionStream, Errno::ENOTDIR)
  if rest.nil?
    next_stream
  else
    next_stream.stream_at(rest)
  end
end