Open-ILS/src/perlmods/OpenILS/Application/Storage/Driver/Pg/storage.pm

   1
   2 {
   3         package OpenILS::Application::Storage;
   4         use OpenSRF::Utils::Logger;
   5
   6         our $NOPRIMARY = 0;
   7         my $log = 'OpenSRF::Utils::Logger';
   8         my $pg = 'OpenILS::Application::Storage::Driver::Pg';
   9
  10         sub child_exit {
  11                 $_->disconnect for $pg->db_Handles;
  12         }
  13
  14         sub current_xact {
  15                 my $self = shift;
  16                 my $client = shift;
  17                 return $pg->current_xact_id;
  18         }
  19         __PACKAGE__->register_method(
  20                 method          => 'current_xact',
  21                 api_name        => 'open-ils.storage.transaction.current',
  22                 api_level       => 1,
  23                 argc            => 0,
  24         );
  25
  26
  27         sub pg_begin_xaction {
  28                 my $self = shift;
  29                 my $client = shift;
  30
  31                 local $OpenILS::Application::Storage::WRITE = 1;
  32
  33                 if (my $old_xact = $pg->current_xact_session) {
  34                         if ($pg->current_xact_is_auto) {
  35                                 $log->debug("Commiting old autocommit transaction with Open-ILS XACT-ID [$old_xact]", INFO);
  36                                 $self->method_lookup("open-ils.storage.transaction.commit")->run();
  37                         } else {
  38                                 $log->debug("Rolling back old NON-autocommit transaction with Open-ILS XACT-ID [$old_xact]", INFO);
  39                                 $self->method_lookup("open-ils.storage.transaction.rollback")->run();
  40                                 throw OpenSRF::DomainObject::oilsException->new(
  41                                                 statusCode => 500,
  42                                                 status => "Previous transaction rolled back!",
  43                                 );
  44                         }
  45                 }
  46
  47                 $pg->set_xact_session( $client->session );
  48                 my $xact_id = $pg->current_xact_id;
  49
  50                 $log->debug("Beginning a new transaction with Open-ILS XACT-ID [$xact_id]", INFO);
  51
  52                 my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
  53
  54                 try {
  55                         $dbh->begin_work;
  56
  57                 } catch Error with {
  58                         my $e = shift;
  59                         $log->debug("Failed to begin a new transaction with Open-ILS XACT-ID [$xact_id]: ".$e, INFO);
  60                         throw $e;
  61                 };
  62
  63
  64                 my $death_cb = $client->session->register_callback(
  65                         death => sub {
  66                                 __PACKAGE__->pg_rollback_xaction;
  67                         }
  68                 );
  69
  70                 $log->debug("Registered 'death' callback [$death_cb] for new transaction with Open-ILS XACT-ID [$xact_id]", DEBUG);
  71
  72                 $client->session->session_data( death_cb => $death_cb );
  73
  74                 if ($self->api_name =~ /autocommit$/o) {
  75                         $pg->current_xact_is_auto(1);
  76                         my $dc_cb = $client->session->register_callback(
  77                                 disconnect => sub {
  78                                         my $ses = shift;
  79                                         $ses->unregister_callback(death => $death_cb);
  80                                         __PACKAGE__->pg_commit_xaction;
  81                                 }
  82                         );
  83                         $log->debug("Registered 'disconnect' callback [$dc_cb] for new transaction with Open-ILS XACT-ID [$xact_id]", DEBUG);
  84                         if ($client and $client->session) {
  85                                 $client->session->session_data( disconnect_cb => $dc_cb );
  86                         }
  87                 }
  88
  89                 return 1;
  90
  91         }
  92         __PACKAGE__->register_method(
  93                 method          => 'pg_begin_xaction',
  94                 api_name        => 'open-ils.storage.transaction.begin',
  95                 api_level       => 1,
  96                 argc            => 0,
  97         );
  98         __PACKAGE__->register_method(
  99                 method          => 'pg_begin_xaction',
 100                 api_name        => 'open-ils.storage.transaction.begin.autocommit',
 101                 api_level       => 1,
 102                 argc            => 0,
 103         );
 104
 105         sub pg_commit_xaction {
 106                 my $self = shift;
 107
 108                 local $OpenILS::Application::Storage::WRITE = 1;
 109
 110                 my $xact_id = $pg->current_xact_id;
 111
 112                 my $success = 1;
 113                 try {
 114                         $log->debug("Committing transaction with Open-ILS XACT-ID [$xact_id]", INFO) if ($xact_id);
 115                         my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 116                         $dbh->commit;
 117
 118                 } catch Error with {
 119                         my $e = shift;
 120                         $log->debug("Failed to commit transaction with Open-ILS XACT-ID [$xact_id]: ".$e, INFO);
 121                         $success = 0;
 122                 };
 123
 124                 $pg->current_xact_session->unregister_callback( death =>
 125                         $pg->current_xact_session->session_data( 'death_cb' )
 126                 ) if ($pg->current_xact_session);
 127
 128                 if ($pg->current_xact_is_auto) {
 129                         $pg->current_xact_session->unregister_callback( disconnect =>
 130                                 $pg->current_xact_session->session_data( 'disconnect_cb' )
 131                         );
 132                 }
 133
 134                 $pg->unset_xact_session;
 135
 136                 return $success;
 137
 138         }
 139         __PACKAGE__->register_method(
 140                 method          => 'pg_commit_xaction',
 141                 api_name        => 'open-ils.storage.transaction.commit',
 142                 api_level       => 1,
 143                 argc            => 0,
 144         );
 145
 146         sub pg_rollback_xaction {
 147                 my $self = shift;
 148
 149                 local $OpenILS::Application::Storage::WRITE = 1;
 150
 151                 my $xact_id = $pg->current_xact_id;
 152
 153                 my $success = 1;
 154                 try {
 155                         my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 156                         $log->debug("Rolling back a transaction with Open-ILS XACT-ID [$xact_id]", INFO);
 157                         $dbh->rollback;
 158
 159                 } catch Error with {
 160                         my $e = shift;
 161                         $log->debug("Failed to roll back transaction with Open-ILS XACT-ID [$xact_id]: ".$e, INFO);
 162                         $success = 0;
 163                 };
 164
 165                 $pg->current_xact_session->unregister_callback( death =>
 166                         $pg->current_xact_session->session_data( 'death_cb' )
 167                 ) if ($pg->current_xact_session);
 168
 169                 if ($pg->current_xact_is_auto) {
 170                         $pg->current_xact_session->unregister_callback( disconnect =>
 171                                 $pg->current_xact_session->session_data( 'disconnect_cb' )
 172                         );
 173                 }
 174
 175                 $pg->unset_xact_session;
 176
 177                 return $success;
 178         }
 179         __PACKAGE__->register_method(
 180                 method          => 'pg_rollback_xaction',
 181                 api_name        => 'open-ils.storage.transaction.rollback',
 182                 api_level       => 1,
 183                 argc            => 0,
 184         );
 185
 186         sub set_savepoint {
 187                 my $self = shift;
 188                 my $client = shift;
 189                 my $sp = shift || 'osrf_savepoint';
 190                 return OpenILS::Application::Storage::CDBI->db_Main->pg_savepoint($sp);
 191         }
 192         __PACKAGE__->register_method(
 193                 method          => 'set_savepoint',
 194                 api_name        => 'open-ils.storage.savepoint.set',
 195                 api_level       => 1,
 196                 argc            => 1,
 197         );
 198
 199         sub release_savepoint {
 200                 my $self = shift;
 201                 my $client = shift;
 202                 my $sp = shift || 'osrf_savepoint';
 203                 return OpenILS::Application::Storage::CDBI->db_Main->pg_release($sp);
 204         }
 205         __PACKAGE__->register_method(
 206                 method          => 'release_savepoint',
 207                 api_name        => 'open-ils.storage.savepoint.release',
 208                 api_level       => 1,
 209                 argc            => 1,
 210         );
 211
 212         sub rollback_to_savepoint {
 213                 my $self = shift;
 214                 my $client = shift;
 215                 my $sp = shift || 'osrf_savepoint';
 216                 return OpenILS::Application::Storage::CDBI->db_Main->pg_rollback_to($sp);
 217         }
 218         __PACKAGE__->register_method(
 219                 method          => 'rollback_to_savepoint',
 220                 api_name        => 'open-ils.storage.savepoint.rollback',
 221                 api_level       => 1,
 222                 argc            => 1,
 223         );
 224
 225
 226         sub copy_create_start {
 227                 my $self = shift;
 228                 my $client = shift;
 229
 230                 local $OpenILS::Application::Storage::WRITE = 1;
 231
 232                 #return undef unless ($pg->current_xact_session);
 233
 234                 my @cols = $self->{cdbi}->columns('Essential');
 235                 if ($NOPRIMARY) {
 236                         my ($p) = $self->{cdbi}->columns('Primary');
 237                         @cols = grep { $_ ne $p } @cols;
 238                 }
 239
 240                 my $col_list = join ',', @cols;
 241
 242                 $log->debug('Starting COPY import for '.$self->{cdbi}->table." ($col_list)", DEBUG);
 243                 $self->{cdbi}->sql_copy_start($self->{cdbi}->table, $col_list)->execute;
 244
 245                 return 1;
 246         }
 247
 248         sub copy_create_push {
 249                 my $self = shift;
 250                 my $client = shift;
 251                 my @fm_nodes = @_;
 252
 253                 local $OpenILS::Application::Storage::WRITE = 1;
 254
 255                 #return undef unless ($pg->current_xact_session);
 256
 257                 my @cols = $self->{cdbi}->columns('Essential');
 258                 if ($NOPRIMARY) {
 259                         my ($p) = $self->{cdbi}->columns('Primary');
 260                         @cols = grep { $_ ne $p } @cols;
 261                 }
 262
 263                 my $dbh = $self->{cdbi}->db_Main;
 264                 for my $node ( @fm_nodes ) {
 265                         next unless ($node);
 266                         my $line = join("\t", map { defined($node->$_()) ? $node->$_() : '\N' } @cols);
 267                         $log->debug("COPY line: [$line]",DEBUG);
 268                         $dbh->pg_putline($line."\n");
 269                 }
 270
 271                 return scalar(@fm_nodes);
 272         }
 273
 274         sub copy_create_finish {
 275                 my $self = shift;
 276                 my $client = shift;
 277                 my @fm_nodes = @_;
 278
 279                 local $OpenILS::Application::Storage::WRITE = 1;
 280
 281                 #return undef unless ($pg->current_xact_session);
 282
 283                 my $dbh = $self->{cdbi}->db_Main;
 284
 285                 $dbh->pg_endcopy || $log->debug("Could not end COPY with pg_endcopy", WARN);
 286
 287                 $log->debug('COPY import for '.$self->{cdbi}->table." ($col_list) complete", DEBUG);
 288
 289                 return 1;
 290         }
 291
 292         sub copy_create {
 293                 my $self = shift;
 294                 my $client = shift;
 295                 my @fm_nodes = @_;
 296
 297                 local $NOPRIMARY = 1;
 298
 299                 copy_create_start(  $self => $client );
 300                 copy_create_push(   $self => $client => @fm_nodes );
 301                 copy_create_finish( $self => $client );
 302
 303                 return scalar(@fm_nodes);
 304         }
 305
 306         sub autoprimary {
 307                 my $class = shift;
 308                 my $val = shift;
 309                 $NOPRIMARY = $val if (defined $val);
 310                 return $NOPRIMARY;
 311         }
 312
 313 }
 314
 315 1;