Open-ILS/src/perlmods/lib/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         if ($ENV{TZ}) {
  64             try {
  65                 $dbh->do('SET LOCAL timezone TO ?;',{},$ENV{TZ});
  66
  67             } catch Error with {
  68                 my $e = shift;
  69                 $log->debug("Failed to set timezone: $ENV{TZ}", WARN);
  70             };
  71         }
  72
  73
  74         if ($client->session) { # not a subrequest
  75             my $death_cb = $client->session->register_callback(
  76                 death => sub {
  77                     __PACKAGE__->pg_rollback_xaction;
  78                 }
  79             );
  80
  81             $log->debug("Registered 'death' callback [$death_cb] for new transaction with Open-ILS XACT-ID [$xact_id]", DEBUG);
  82
  83             $client->session->session_data( death_cb => $death_cb );
  84
  85             if ($self->api_name =~ /autocommit$/o) {
  86                 $pg->current_xact_is_auto(1);
  87                 my $dc_cb = $client->session->register_callback(
  88                     disconnect => sub {
  89                         my $ses = shift;
  90                         $ses->unregister_callback(death => $death_cb);
  91                         __PACKAGE__->pg_commit_xaction;
  92                     }
  93                 );
  94                 $log->debug("Registered 'disconnect' callback [$dc_cb] for new transaction with Open-ILS XACT-ID [$xact_id]", DEBUG);
  95                 if ($client and $client->session) {
  96                     $client->session->session_data( disconnect_cb => $dc_cb );
  97                 }
  98             }
  99         }
 100
 101         return 1;
 102
 103     }
 104     __PACKAGE__->register_method(
 105         method      => 'pg_begin_xaction',
 106         api_name    => 'open-ils.storage.transaction.begin',
 107         api_level   => 1,
 108         argc        => 0,
 109     );
 110     __PACKAGE__->register_method(
 111         method      => 'pg_begin_xaction',
 112         api_name    => 'open-ils.storage.transaction.begin.autocommit',
 113         api_level   => 1,
 114         argc        => 0,
 115     );
 116
 117     sub pg_commit_xaction {
 118         my $self = shift;
 119
 120         local $OpenILS::Application::Storage::WRITE = 1;
 121
 122         my $xact_id = $pg->current_xact_id;
 123
 124         my $success = 1;
 125         try {
 126             $log->debug("Committing transaction with Open-ILS XACT-ID [$xact_id]", INFO) if ($xact_id);
 127             my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 128             $dbh->commit;
 129
 130         } catch Error with {
 131             my $e = shift;
 132             $log->debug("Failed to commit transaction with Open-ILS XACT-ID [$xact_id]: ".$e, INFO);
 133             $success = 0;
 134         };
 135
 136         if ($pg->current_xact_session) { # not a subrequest
 137             $pg->current_xact_session->unregister_callback( death =>
 138                 $pg->current_xact_session->session_data( 'death_cb' )
 139             ) if ($pg->current_xact_session);
 140
 141             if ($pg->current_xact_is_auto) {
 142                 $pg->current_xact_session->unregister_callback( disconnect =>
 143                     $pg->current_xact_session->session_data( 'disconnect_cb' )
 144                 );
 145             }
 146         }
 147
 148         $pg->unset_xact_session;
 149
 150         return $success;
 151
 152     }
 153     __PACKAGE__->register_method(
 154         method      => 'pg_commit_xaction',
 155         api_name    => 'open-ils.storage.transaction.commit',
 156         api_level   => 1,
 157         argc        => 0,
 158     );
 159
 160     sub pg_rollback_xaction {
 161         my $self = shift;
 162
 163         local $OpenILS::Application::Storage::WRITE = 1;
 164
 165         my $xact_id = $pg->current_xact_id;
 166
 167         my $success = 1;
 168         try {
 169             my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 170             $log->debug("Rolling back a transaction with Open-ILS XACT-ID [$xact_id]", INFO);
 171             $dbh->rollback;
 172
 173         } catch Error with {
 174             my $e = shift;
 175             $log->debug("Failed to roll back transaction with Open-ILS XACT-ID [$xact_id]: ".$e, INFO);
 176             $success = 0;
 177         };
 178
 179         if ($pg->current_xact_session) { # not a subrequest
 180             $pg->current_xact_session->unregister_callback( death =>
 181                 $pg->current_xact_session->session_data( 'death_cb' )
 182             ) if ($pg->current_xact_session);
 183
 184             if ($pg->current_xact_is_auto) {
 185                 $pg->current_xact_session->unregister_callback( disconnect =>
 186                     $pg->current_xact_session->session_data( 'disconnect_cb' )
 187                 );
 188             }
 189         }
 190
 191         $pg->unset_xact_session;
 192
 193         return $success;
 194     }
 195     __PACKAGE__->register_method(
 196         method      => 'pg_rollback_xaction',
 197         api_name    => 'open-ils.storage.transaction.rollback',
 198         api_level   => 1,
 199         argc        => 0,
 200     );
 201
 202     sub set_savepoint {
 203         my $self = shift;
 204         my $client = shift;
 205         my $sp = shift || 'osrf_savepoint';
 206         return OpenILS::Application::Storage::CDBI->db_Main->pg_savepoint($sp);
 207     }
 208     __PACKAGE__->register_method(
 209             method          => 'set_savepoint',
 210             api_name        => 'open-ils.storage.savepoint.set',
 211             api_level       => 1,
 212             argc            => 1,
 213     );
 214
 215     sub release_savepoint {
 216         my $self = shift;
 217         my $client = shift;
 218         my $sp = shift || 'osrf_savepoint';
 219         return OpenILS::Application::Storage::CDBI->db_Main->pg_release($sp);
 220     }
 221     __PACKAGE__->register_method(
 222             method          => 'release_savepoint',
 223             api_name        => 'open-ils.storage.savepoint.release',
 224             api_level       => 1,
 225             argc            => 1,
 226     );
 227
 228     sub rollback_to_savepoint {
 229         my $self = shift;
 230         my $client = shift;
 231         my $sp = shift || 'osrf_savepoint';
 232         return OpenILS::Application::Storage::CDBI->db_Main->pg_rollback_to($sp);
 233     }
 234     __PACKAGE__->register_method(
 235             method          => 'rollback_to_savepoint',
 236             api_name        => 'open-ils.storage.savepoint.rollback',
 237             api_level       => 1,
 238             argc            => 1,
 239     );
 240
 241     sub pg_set_audit_info {
 242         my $self = shift;
 243         my $client = shift;
 244         my $authtoken = shift;
 245         my $user_id = shift;
 246         my $ws_id = shift;
 247
 248         local $OpenILS::Application::Storage::WRITE = 1;
 249
 250         $log->debug("Setting auditor information", INFO);
 251
 252         if($pg->current_audit_session) {
 253             $log->debug("Already sent audit data.", INFO);
 254             return 1;
 255         }
 256
 257         my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 258
 259         try {
 260             if(!$user_id) {
 261                 my $ses = OpenSRF::AppSession->create('open-ils.auth');
 262                 my $content = $ses->request('open-ils.auth.session.retrieve', $authtoken, 1)->gather(1);
 263                 if(!$content or !$content->{userObj}) {
 264                     return 0;
 265                 }
 266                 $user_id = $content->{userObj}->id;
 267                 $ws_id = $content->{userObj}->wsid;
 268             }
 269             $ws_id = 'NULL' unless $ws_id;
 270             $dbh->do("SELECT auditor.set_audit_info($user_id, $ws_id);");
 271         } catch Error with {
 272             my $e = shift;
 273             $log->debug("Failed to set auditor information: ".$e, INFO);
 274             throw $e;
 275         };
 276
 277         $pg->set_audit_session( $client->session );
 278
 279         if ($client->session) { # not a subrequest
 280             my $death_cb = $client->session->register_callback(
 281                 death => sub {
 282                     __PACKAGE__->pg_clear_audit_info;
 283                 }
 284             );
 285
 286             $log->debug("Registered 'death' callback [$death_cb] for clearing audit information", DEBUG);
 287
 288             $client->session->session_data( death_cb_ai => $death_cb );
 289         }
 290
 291         return 1;
 292
 293     }
 294     __PACKAGE__->register_method(
 295         method      => 'pg_set_audit_info',
 296         api_name    => 'open-ils.storage.set_audit_info',
 297         api_level   => 1,
 298         argc        => 3,
 299     );
 300
 301     sub pg_clear_audit_info {
 302         my $self = shift;
 303
 304         try {
 305             my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 306             $log->debug("Clearing Audit Information", INFO);
 307             $dbh->do("SELECT auditor.clear_audit_info();");
 308         } catch Error with {
 309             my $e = shift;
 310             $log->debug("Failed to clear audit information: ".$e, INFO);
 311         };
 312
 313         if ($pg->current_audit_session) { # not a subrequest
 314             $pg->current_audit_session->unregister_callback( death =>
 315                 $pg->current_audit_session->session_data( 'death_cb_ai' )
 316             ) if ($pg->current_audit_session);
 317         }
 318
 319         $pg->unset_audit_session;
 320     }
 321
 322
 323
 324     sub copy_create_start {
 325         my $self = shift;
 326         my $client = shift;
 327
 328         local $OpenILS::Application::Storage::WRITE = 1;
 329
 330         #return undef unless ($pg->current_xact_session);
 331
 332         my @cols = $self->{cdbi}->columns('Essential');
 333         if ($NOPRIMARY) {
 334             my ($p) = $self->{cdbi}->columns('Primary');
 335             @cols = grep { $_ ne $p } @cols;
 336         }
 337
 338         my $col_list = join ',', @cols;
 339
 340         $log->debug('Starting COPY import for '.$self->{cdbi}->table." ($col_list)", DEBUG);
 341         $self->{cdbi}->sql_copy_start($self->{cdbi}->table, $col_list)->execute;
 342
 343         return 1;
 344     }
 345
 346     sub copy_create_push {
 347         my $self = shift;
 348         my $client = shift;
 349         my @fm_nodes = @_;
 350
 351         local $OpenILS::Application::Storage::WRITE = 1;
 352
 353         #return undef unless ($pg->current_xact_session);
 354
 355         my @cols = $self->{cdbi}->columns('Essential');
 356         if ($NOPRIMARY) {
 357             my ($p) = $self->{cdbi}->columns('Primary');
 358             @cols = grep { $_ ne $p } @cols;
 359         }
 360
 361         my $dbh = $self->{cdbi}->db_Main;
 362         for my $node ( @fm_nodes ) {
 363             next unless ($node);
 364             my $line = join("\t", map { defined($node->$_()) ? $node->$_() : '\N' } @cols);
 365             $log->debug("COPY line: [$line]",DEBUG);
 366             $dbh->pg_putline($line."\n");
 367         }
 368
 369         return scalar(@fm_nodes);
 370     }
 371
 372     sub copy_create_finish {
 373         my $self = shift;
 374         my $client = shift;
 375         my @fm_nodes = @_;
 376
 377         local $OpenILS::Application::Storage::WRITE = 1;
 378
 379         #return undef unless ($pg->current_xact_session);
 380
 381         my $dbh = $self->{cdbi}->db_Main;
 382
 383         $dbh->pg_endcopy || $log->debug("Could not end COPY with pg_endcopy", WARN);
 384
 385         $log->debug('COPY import for '.$self->{cdbi}->table." ($col_list) complete", DEBUG);
 386
 387         return 1;
 388     }
 389
 390     sub copy_create {
 391         my $self = shift;
 392         my $client = shift;
 393         my @fm_nodes = @_;
 394
 395         local $NOPRIMARY = 1;
 396
 397         copy_create_start(  $self => $client );
 398         copy_create_push(   $self => $client => @fm_nodes );
 399         copy_create_finish( $self => $client );
 400
 401         return scalar(@fm_nodes);
 402     }
 403
 404     sub autoprimary {
 405         my $class = shift;
 406         my $val = shift;
 407         $NOPRIMARY = $val if (defined $val);
 408         return $NOPRIMARY;
 409     }
 410
 411 }
 412
 413 1;