Changeset 6

Show
Ignore:
Timestamp:
08/19/09 15:46:48 (4 years ago)
Author:
jesus
Message:

robustness in the transport layers (sql and mq) and a skeleton config

Files:

Legend:

Unmodified
Added
Removed
Modified
Copied
Moved

  • trunk/perl/lib/Cornea/ApacheStore.pm

    r4 r6  
    1818  my ($serviceId, $assetId, $repId) = @_; 
    1919  my $config = Cornea::Config->new(); 
    20   my $base = $config->get("Basepath"); 
     20  my $base = $config->get("Storage::path"); 
    2121  $assetId = "$assetId"; # Treat it as a string. 
    2222  if($assetId =~ /^\d+$/ and length $assetId < 4) { 

  • trunk/perl/lib/Cornea/Config.pm

    r4 r6  
    5252    while(<CONF>) { 
    5353        next if /^\s*[;#]/; 
    54         if(/^\s*([^\s=]+)\s*=\s*(.*)$/) { 
     54        if(/^\s*([^\s=]+)\s*=\s*\[(.*)\]\s*$/) { 
     55          my $key = lc($1); 
     56          (my $val = $2) =~ s/\s+$//; 
     57          _g_Config($key, [split /\s+/, $val]); 
     58        } 
     59        elsif(/^\s*([^\s=]+)\s*=\s*(.*)$/) { 
    5560          my $key = lc($1); 
    5661          (my $val = $2) =~ s/\s+$//; 
    …… 
    6267} 
    6368 
    64 sub get { _g_Config(lc($_[1])); } 
     69sub get { my $v = _g_Config(lc($_[1])); return (ref $v eq 'ARRAY') ? $v->[0] : $v; } 
     70sub get_list { my $v = _g_Config(lc($_[1])); return (ref $v eq 'ARRAY') ? $v : [$v]; } 
    6571sub set { _g_Config(lc($_[1]), $_[2]); } 
    6672sub unset { _g_Config_unset(lc($_[1])); } 

  • trunk/perl/lib/Cornea/Queue.pm

    r4 r6  
    1212    delete $self->{stomp}; 
    1313  } 
    14   my $stomp = Net::Stomp->new( { hostname => $config->get("MQ::hostname"), 
    15                                  port => $config->get("MQ::port") }); 
    16   foreach (@{$self->{queues}}) { 
    17     $stomp->subscribe( { destination => $_, ack => 'client' } ); 
     14  my $stomphosts = $config->get_list("MQ::hostname"); 
     15  foreach my $hostname ( @$stomphosts ) { 
     16    eval { 
     17      my $stomp = Net::Stomp->new( { hostname => $hostame, 
     18                                     port => $config->get("MQ::port") }); 
     19      $stomp->connect( { login => $config->get("MQ::login"), 
     20                         passcode => $config->get("MQ::passcode") } ) ||  
     21        die "could not connect to stomp on $hostname\n"; 
     22    }; 
     23    last unless $@; 
     24    $stomp = undef; 
     25  } 
     26  if($stomp) { 
     27    foreach (@{$self->{queues}}) { 
     28      $stomp->subscribe( { destination => $_, ack => 'client' } ); 
     29    } 
    1830  } 
    1931  $self->{stomp} = $stomp; 

  • trunk/perl/lib/Cornea/RecallTable.pm

    r4 r6  
    22use strict; 
    33use Cornea::Config; 
     4use Cornea::Utils; 
    45use DBI; 
    56 
     7sub __connect { 
     8  my $self = shift; 
     9  my $config = Cornea::Config->new(); 
     10  my $dbh; 
     11  my $dsns = $config->get_list("DB::dsn"); 
     12  Cornea::Utils::shuffle($dsns); 
     13  foreach my $dsn (@$dsns) { 
     14    eval { 
     15      $dbh = DBI->connect($config->get("DB::dsn"), 
     16                          $config->get("DB::user"), 
     17                          $config->get("DB::pass"), 
     18                          { PrintError => 0, RaiseError => 1 }, 
     19                         ); 
     20    }; 
     21    last unless ($@); 
     22  } 
     23  $self->{dbh} = shift; 
     24} 
     25sub __reconnect { 
     26  my $self = shift; 
     27  $self->{dbh} = undef; 
     28  $self->__connect(); 
     29} 
    630sub new { 
    731  my $class = shift; 
    8   $class = ref($class) ? ref $class : $class; 
    9  
    10   my $config = Cornea::Config->new(); 
    11   my $dbh = DBI->connect($config->get("DB::dsn"), 
    12                          $config->get("DB::user"), 
    13                          $config->get("DB::pass"), 
    14                          { PrintError => 0, RaiseError => 1 }, 
    15                         ); 
    16  
    17   bless {}, $class; 
     32  my $self = bless { }, $class; 
     33  $self->__connect; 
    1834} 
    1935 
    …… 
    2137  my $self = shift; 
    2238  my ($serviceId, $assetId, $repId, $snl) = @_; 
     39  my $tried = 0; 
    2340  die 'bad parameters' unless UNIVERSAL::ISA($snl, 'Cornea::StorageNodeList'); 
    24   my $sth = $self->{dbh}->prepare("select storeAsset(?,?,?,?)"); 
    25   $sth->execute($serviceId, $assetId, $repId, $snl); 
    26   $sth->finish(); 
     41 again: 
     42  eval { 
     43    my $sth = $self->{dbh}->prepare("select storeAsset(?,?,?,?)"); 
     44    $sth->execute($serviceId, $assetId, $repId, $snl); 
     45    $sth->finish(); 
     46  }; 
     47  if ($@) { 
     48    unless ($tried++) { $self->{dbh}->__reconnect();  goto again; } 
     49    die $@ if $@; 
     50  } 
     51  return 1; 
    2752} 
    2853 
    29 sub getOpenNodes { 
     54sub find { 
    3055  my $self = shift; 
    31   my $snl = Cornea::StorageNodeList->new(); 
    32   my $sth = $self->{dbh}->prepare("select * from getOpenNodes()"); 
    33   $sth->execute(); 
    34   while(my $row = $sth->fetchrow_hashref()) { 
    35     $snl->add(Cornea::StorageNode->new_from_row($row)); 
     56  my ($serviceId, $assetId, $repId) = @_; 
     57  my $sth = $self->{dbh}->prepare("select findAsset(?,?,?)"); 
     58  my $tried = 0; 
     59  my $C; 
     60 again: 
     61  eval { 
     62    $C = Cornea::StorageNodeList->new(); 
     63    $sth->execute($serviceId, $assetId, $repId); 
     64    while(my $node = $sth->fetchrow_hashref()) { 
     65      $C->add(Cornea::StorageNode->new_from_row($node)); 
     66    } 
     67    $sth->finish(); 
     68  }; 
     69  if ($@) { 
     70    unless ($tried++) { $self->{dbh}->__reconnect();  goto again; } 
     71    die $@ if $@; 
    3672  } 
    37   $sth->finish(); 
     73  return $C; 
     74} 
     75 
     76sub getNodes { 
     77  my $self = shift; 
     78  my $type = shift; 
     79  my $tried = 0; 
     80  my $snl; 
     81 again: 
     82  eval { 
     83    $snl = Cornea::StorageNodeList->new(); 
     84    my $sth = $self->{dbh}->prepare("select * from getCorneaNodes(?)"); 
     85    $sth->execute($type); 
     86    while(my $row = $sth->fetchrow_hashref()) { 
     87      $snl->add(Cornea::StorageNode->new_from_row($row)); 
     88    } 
     89    $sth->finish(); 
     90  }; 
     91  if ($@) { 
     92    unless ($tried++) { $self->{dbh}->__reconnect();  goto again; } 
     93    die $@ if $@; 
     94  } 
    3895  return $snl; 
    3996} 
    …… 
    4299  my $self = shift; 
    43100  my ($serviceId, $repId) = @_; 
    44   my $sth = $self->{dbh}->prepare("select * from getRepInfo(?,?)"); 
    45   $sth->execute($serviceId, $repId); 
    46   my $row = $sth->fetchrow_hashref(); 
    47   $sth->finish(); 
     101  my $tried = 0; 
     102  my $row; 
     103 again: 
     104  eval { 
     105    my $sth = $self->{dbh}->prepare("select * from getRepInfo(?,?)"); 
     106    $sth->execute($serviceId, $repId); 
     107    $row = $sth->fetchrow_hashref(); 
     108    $sth->finish(); 
     109  }; 
     110  if ($@) { 
     111    unless ($tried++) { $self->{dbh}->__reconnect();  goto again; } 
     112    die $@ if $@; 
     113  } 
    48114  return Cornea::RepresentationInfo->new_from_row($row); 
    49115} 
    …… 
    52118  my $self = shift; 
    53119  my ($serviceId, $repId) = @_; 
     120  my $tried = 0; 
    54121  my @deps; 
    55122 
    56   my $sth = $self->{dbh}->prepare("select * from getRepInfoDependents(?,?)"); 
    57   $sth->execute($serviceId, $repId); 
    58   while(my $row = $sth->fetchrow_hashref()) { 
    59     push @deps, Cornea::RepresentationInfo->new_from_row($row); 
     123  eval { 
     124    @deps = (); 
     125    my $sth = $self->{dbh}->prepare("select * from getRepInfoDependents(?,?)"); 
     126    $sth->execute($serviceId, $repId); 
     127    while(my $row = $sth->fetchrow_hashref()) { 
     128      push @deps, Cornea::RepresentationInfo->new_from_row($row); 
     129    } 
     130    $sth->finish(); 
     131  }; 
     132  if ($@) { 
     133    unless ($tried++) { $self->{dbh}->__reconnect();  goto again; } 
     134    die $@ if $@; 
    60135  } 
    61   $sth->finish(); 
    62136  return @deps; 
    63137} 

  • trunk/perl/lib/Cornea/StorageNode.pm

    r3 r6  
    33 
    44=pod 
     5 
     6=item id() 
     7 
     8The unique identifier. 
     9 
     10=item state() 
     11 
     12The state of the node: {open,closed,offline,decommissioned} 
     13 
     14=item total_storage() 
     15 
     16The total storage in kilobytes. 
     17 
     18=item used_storage() 
     19 
     20The amount of storage used in kilobytes. 
     21 
     22=item fqdn() 
     23 
     24The fully-qualified domain name of the node. 
    525 
    626=item location() 
    …… 
    1434=cut 
    1535 
    16 sub location() { 
     36sub new_from_row { 
     37  my $class = shift; 
     38  my $hash = shift; 
     39  bless $hash, $class; 
    1740} 
     41sub id { shift->{id}; } 
     42sub state { shift->{state}; } 
     43sub total_storage { shift->{total_storage}; } 
     44sub used_storage { shift->{used_storage}; } 
     45sub fqdn { shift->{fqdn}; } 
     46sub location { shift->{location}; } 
    1847 
    1948sub distance() {