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
  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     sub pg_set_audit_info {
 226         my $self = shift;
 227         my $client = shift;
 228         my $authtoken = shift;
 229         my $user_id = shift;
 230         my $ws_id = shift;
 231
 232         local $OpenILS::Application::Storage::WRITE = 1;
 233
 234         $log->debug("Setting auditor information", INFO);
 235
 236         if($pg->current_audit_session) {
 237             $log->debug("Already sent audit data.", INFO);
 238             return 1;
 239         }
 240
 241         my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 242
 243         try {
 244             if(!$user_id) {
 245                 my $ses = OpenSRF::AppSession->create('open-ils.auth');
 246                 my $content = $ses->request('open-ils.auth.session.retrieve', $authtoken, 1)->gather(1);
 247                 if(!$content or !$content->{userObj}) {
 248                     return 0;
 249                 }
 250                 $user_id = $content->{userObj}->id;
 251                 $ws_id = $content->{userObj}->wsid;
 252             }
 253             $ws_id = 'NULL' unless $ws_id;
 254             $dbh->do("SELECT auditor.set_audit_info($user_id, $ws_id);");
 255         } catch Error with {
 256             my $e = shift;
 257             $log->debug("Failed to set auditor information: ".$e, INFO);
 258             throw $e;
 259         };
 260
 261         $pg->set_audit_session( $client->session );
 262
 263         my $death_cb = $client->session->register_callback(
 264             death => sub {
 265                 __PACKAGE__->pg_clear_audit_info;
 266             }
 267         );
 268
 269         $log->debug("Registered 'death' callback [$death_cb] for clearing audit information", DEBUG);
 270
 271         $client->session->session_data( death_cb_ai => $death_cb );
 272
 273         return 1;
 274
 275     }
 276     __PACKAGE__->register_method(
 277         method      => 'pg_set_audit_info',
 278         api_name    => 'open-ils.storage.set_audit_info',
 279         api_level   => 1,
 280         argc        => 3,
 281     );
 282
 283     sub pg_clear_audit_info {
 284         my $self = shift;
 285
 286         try {
 287             my $dbh = OpenILS::Application::Storage::CDBI->db_Main;
 288             $log->debug("Clearing Audit Information", INFO);
 289             $dbh->do("SELECT auditor.clear_audit_info();");
 290         } catch Error with {
 291             my $e = shift;
 292             $log->debug("Failed to clear audit information: ".$e, INFO);
 293         };
 294
 295         $pg->current_audit_session->unregister_callback( death =>
 296             $pg->current_audit_session->session_data( 'death_cb_ai' )
 297         ) if ($pg->current_audit_session);
 298
 299         $pg->unset_audit_session;
 300     }
 301
 302
 303
 304     sub copy_create_start {
 305         my $self = shift;
 306         my $client = shift;
 307
 308         local $OpenILS::Application::Storage::WRITE = 1;
 309
 310         #return undef unless ($pg->current_xact_session);
 311
 312         my @cols = $self->{cdbi}->columns('Essential');
 313         if ($NOPRIMARY) {
 314             my ($p) = $self->{cdbi}->columns('Primary');
 315             @cols = grep { $_ ne $p } @cols;
 316         }
 317
 318         my $col_list = join ',', @cols;
 319
 320         $log->debug('Starting COPY import for '.$self->{cdbi}->table." ($col_list)", DEBUG);
 321         $self->{cdbi}->sql_copy_start($self->{cdbi}->table, $col_list)->execute;
 322
 323         return 1;
 324     }
 325
 326     sub copy_create_push {
 327         my $self = shift;
 328         my $client = shift;
 329         my @fm_nodes = @_;
 330
 331         local $OpenILS::Application::Storage::WRITE = 1;
 332
 333         #return undef unless ($pg->current_xact_session);
 334
 335         my @cols = $self->{cdbi}->columns('Essential');
 336         if ($NOPRIMARY) {
 337             my ($p) = $self->{cdbi}->columns('Primary');
 338             @cols = grep { $_ ne $p } @cols;
 339         }
 340
 341         my $dbh = $self->{cdbi}->db_Main;
 342         for my $node ( @fm_nodes ) {
 343             next unless ($node);
 344             my $line = join("\t", map { defined($node->$_()) ? $node->$_() : '\N' } @cols);
 345             $log->debug("COPY line: [$line]",DEBUG);
 346             $dbh->pg_putline($line."\n");
 347         }
 348
 349         return scalar(@fm_nodes);
 350     }
 351
 352     sub copy_create_finish {
 353         my $self = shift;
 354         my $client = shift;
 355         my @fm_nodes = @_;
 356
 357         local $OpenILS::Application::Storage::WRITE = 1;
 358
 359         #return undef unless ($pg->current_xact_session);
 360
 361         my $dbh = $self->{cdbi}->db_Main;
 362
 363         $dbh->pg_endcopy || $log->debug("Could not end COPY with pg_endcopy", WARN);
 364
 365         $log->debug('COPY import for '.$self->{cdbi}->table." ($col_list) complete", DEBUG);
 366
 367         return 1;
 368     }
 369
 370     sub copy_create {
 371         my $self = shift;
 372         my $client = shift;
 373         my @fm_nodes = @_;
 374
 375         local $NOPRIMARY = 1;
 376
 377         copy_create_start(  $self => $client );
 378         copy_create_push(   $self => $client => @fm_nodes );
 379         copy_create_finish( $self => $client );
 380
 381         return scalar(@fm_nodes);
 382     }
 383
 384     sub autoprimary {
 385         my $class = shift;
 386         my $val = shift;
 387         $NOPRIMARY = $val if (defined $val);
 388         return $NOPRIMARY;
 389     }
 390
 391 }
 392
 393 1;