Implement get and purge

* glacier: Register purge and get
* lib/App/Glacier/Command.pm (getyn): New method.
* lib/App/Glacier/Command/DeleteFile.pm: Don't invalidate directory on
completion.
* lib/App/Glacier/Command/Get.pm: New file.
* lib/App/Glacier/Command/Purge.pm: New file.
* lib/App/Glacier/Command/ListVault.pm (get_vault_inventory): Only check
for DIR_PENDING.  DIR_OUTDATED is mostly irrelevant, because the inventory
can be as much as 24 hours old.
* lib/App/Glacier/Command/Put.pm: Fix argument checking.
Don't invalidate directory on completion.
* lib/App/Glacier/Job.pm: Implement debugging.
* lib/App/Glacier/Job/FileRetrieval.pm: Bugfixes.

1 files changed, 179 insertions, 0 deletions

diff --git a/lib/App/Glacier/Command/Get.pm b/lib/App/Glacier/Command/Get.pm
new file mode 100644
index 0000000..dd3ca37
--- /dev/null
+++ b/lib/App/Glacier/Command/Get.pm
@@ -0,0 +1,179 @@
+package App::Glacier::Command::Get;
+use strict;
+use warnings;
+use App::Glacier::Command;
+use App::Glacier::Job::FileRetrieval;
+use App::Glacier::DateTime;
+use parent qw(App::Glacier::Command);
+use Carp;
+
+=head1 NAME
+
+glacier get - download file from a vault
+
+=head1 SYNOPSIS
+
+B<glacier put>
+[B<-fit>]
+[B<-s> I<N>]    
+[B<--force>]    
+[B<--interactive>]    
+[B<-j> I<NJOBS>]
+[B<--jobs=>I<NJOBS>]
+[B<--no-clobber>]
+[B<--part-size=>I<N>]    
+[B<--test>]    
+I<VAULT>
+I<FILE>
+[I<LOCALNAME>]
+
+=cut
+
+use constant {
+    IFEXISTS_OVERWRITE => 0,
+    IFEXISTS_KEEP => 1,
+    IFEXISTS_ASK => 2,
+};
+    
+sub getopt {
+    my ($self, %opts) = @_;
+    $self->{_options}{ifexists} = IFEXISTS_OVERWRITE; # Default
+    $self->SUPER::getopt(
+	'interactive|i' => sub { $self->{_options}{ifexists} = IFEXISTS_ASK },
+	'force|f' => sub { $self->{_options}{ifexists} = IFEXISTS_OVERWRITE },
+	'no-clobber|f' => sub { $self->{_options}{ifexists} = IFEXISTS_KEEP },
+	'part-size|s=i' => \$self->{_options}{part_size},
+	'jobs|j=i' => \$self->{_options}{jobs},
+	'test|t' => \$self->{_options}{test},
+	%opts);
+}
+
+sub run {
+    my $self = shift;
+    $self->abend(EX_USAGE, "two or three arguments expected")
+	unless @_ == 2 || @_ == 3;
+    my ($vaultname, $filespec, $localname) = @_;
+    $filespec =~ /^(?<file>.+?)(?:(?<!\\);(?<ver>\d+))?$/
+	or die "unexpected failure";
+
+    my ($filename, $ver) = ($+{file}, $+{ver});
+    
+    $localname = $filename unless defined($localname);
+
+    if (-e $localname) {
+	if ($self->{_options}{ifexists} == IFEXISTS_ASK) {
+	    $self->{_options}{ifexists} =
+		$self->getyn("\"$localname\" already exists, overwrite")
+		  ? IFEXISTS_OVERWRITE : IFEXISTS_KEEP;
+	}
+	if ($self->{_options}{ifexists} == IFEXISTS_KEEP) {
+	    exit(EX_NOPERM);
+	}
+    }
+
+    my $job = new App::Glacier::Job::FileRetrieval($self, $vaultname,
+						   $filename, $ver);
+
+    if ($self->{_options}{test}) {
+	$self->error("downloading file $filename initialized on",
+		     $job->get('CreationDate')->canned_format('full-iso'));
+	$self->error("job id:", $job->id);
+	$self->error("current status:", $job->get('StatusCode'));
+	if ($job->is_completed) {
+	    $self->error("completed on",
+			 $job->get('CompletionDate')->canned_format('full-iso'));
+	}
+	exit(0);
+    }
+    
+    if ($job->is_completed) {
+	$self->download($job, $localname);
+    } else {
+	$self->abend(EX_TEMPFAIL,
+		     "archive retrieval job for $vaultname:$filespec initiated at " .
+		     $job->get('CreationDate')->canned_format
+		     . "; please retry later to download the file");
+    }
+}
+
+sub download {
+    my ($self, $job, $localname) = @_;
+
+    use threads;
+    use threads::shared;
+
+    my $fd :shared;    
+    open($fd, '>', $localname)
+	or $self->abort(EX_FAILURE, "can't open $localname: $!");
+    binmode($fd);
+    truncate($fd, 0);
+    
+    my $archive_size = $job->get('ArchiveSizeInBytes');
+    my $njobs = $self->{_options}{jobs} || 1;
+    my $part_size = $self->{_options}{part_size} || 10*1024*1024*1024; # FIXME
+
+    my $glacier = $self->{_glacier};
+
+    my $tree_hash;
+    
+    if ($njobs <= 1 || $archive_size < $part_size) {
+	# simple download 
+	my $res;
+	($res, $tree_hash) = $glacier->get_job_output($job->vault, $job->id);
+	syswrite($fd, $res);
+    } else {
+	use Fcntl qw(SEEK_SET);
+	
+	$self->debug(1,
+		     "downloading ".$job->file_name(1)." to $localname in chunks of $part_size bytes, in $njobs jobs");
+
+	my @part_hashes = ();
+	my $read_bytes;
+	my $rest_size = $archive_size;
+	my $off = 0;
+	my $part_idx = 0;
+	while ($rest_size) {
+	    for (my $i = 0; $i < $njobs && $rest_size; $i++) {
+		if ($rest_size < $part_size) {
+		    $part_size = $rest_size;
+		}
+		my ($thr) = threads->create(
+		    sub {
+			my ($part_idx, $off) = @_;
+			my $range = $off . '-' . ($off + $part_size);
+			my ($res, $hash) =
+			    $glacier->get_job_output($job->vault,
+						     $job->id, $range);
+			lock $fd;
+			seek($fd, $off, SEEK_SET);
+			syswrite($fd, $res);
+			return ($part_idx, $hash);
+		    },
+		    $part_idx, $off);
+		$part_idx++;
+		$off += $part_size;
+		$rest_size -= $part_size;
+	    }
+
+	    $self->debug(2, "waiting for the bunch to finish");
+	    while (threads->list()) {
+		foreach my $thr (threads->list(threads::joinable)) {
+		    # FIXME: error handling
+		    my ($idx, $hash) = $thr->join()
+			or croak "thread $thr failed";
+		    $part_hashes[$idx] = $hash;
+		}
+	    }
+	}
+	$tree_hash = $glacier->_tree_hash_from_array_ref(\@part_hashes);
+    }
+
+    close($fd);
+    if ($tree_hash ne $job->get('ArchiveSHA256TreeHash')) {
+	unlink $localname;
+	$self->abend(EX_SOFTWARE, "downloaded file is corrupt");
+    }
+}
+    
+1;
+