Open-ILS/src/perlmods/OpenILS/Application/Trigger.pm

   1 package OpenILS::Application::Trigger;
   2 use strict; use warnings;
   3 use OpenILS::Application;
   4 use base qw/OpenILS::Application/;
   5
   6 use OpenSRF::EX qw/:try/;
   7 use OpenSRF::Utils::JSON;
   8
   9 use OpenSRF::AppSession;
  10 use OpenSRF::MultiSession;
  11 use OpenSRF::Utils::SettingsClient;
  12 use OpenSRF::Utils::Logger qw/$logger/;
  13 use OpenSRF::Utils qw/:datetime/;
  14
  15 use DateTime;
  16 use DateTime::Format::ISO8601;
  17
  18 use OpenILS::Utils::Fieldmapper;
  19 use OpenILS::Utils::CStoreEditor q/:funcs/;
  20 use OpenILS::Application::Trigger::Event;
  21 use OpenILS::Application::Trigger::EventGroup;
  22
  23
  24 my $log = 'OpenSRF::Utils::Logger';
  25 my $parallel_collect;
  26 my $parallel_react;
  27
  28 sub initialize {
  29
  30     my $conf = OpenSRF::Utils::SettingsClient->new;
  31     $parallel_collect = $conf->config_value( apps => 'open-ils.trigger' => app_settings => parallel => 'collect') || 1;
  32     $parallel_react = $conf->config_value( apps => 'open-ils.trigger' => app_settings => parallel => 'react') || 1;
  33
  34 }
  35 sub child_init {}
  36
  37 sub create_active_events_for_object {
  38     my $self = shift;
  39     my $client = shift;
  40     my $key = shift;
  41     my $target = shift;
  42     my $location = shift;
  43     my $granularity = shift;
  44     my $user_data = shift;
  45
  46     my $ident = $target->Identity;
  47     my $ident_value = $target->$ident();
  48
  49     my $editor = new_editor(xact=>1);
  50
  51     my $hooks = $editor->search_action_trigger_hook(
  52         { key       => $key,
  53           core_type => $target->json_hint
  54         }
  55     );
  56
  57     unless(@$hooks) {
  58         $editor->rollback;
  59         return undef;
  60     }
  61
  62     my %hook_hash = map { ($_->key, $_) } @$hooks;
  63
  64     my $orgs = $editor->json_query({ from => [ 'actor.org_unit_ancestors' => $location ] });
  65     my $defs = $editor->search_action_trigger_event_definition(
  66         { hook   => [ keys %hook_hash ],
  67           owner  => [ map { $_->{id} } @$orgs  ],
  68           active => 't'
  69         }
  70     );
  71
  72     for my $def ( @$defs ) {
  73         next if ($granularity && $def->granularity ne $granularity );
  74
  75         if ($def->usr_field && $def->opt_in_setting) {
  76             my $ufield = $def->usr_field;
  77             my $uid = $target->$ufield;
  78             $uid = $uid->id if (ref $uid); # fleshed user object, unflesh it
  79
  80             my $opt_in_setting = $editor->search_actor_user_setting(
  81                 { usr   => $uid,
  82                   name  => $def->opt_in_setting,
  83                   value => 'true'
  84                 }
  85             );
  86
  87             next unless (@$opt_in_setting);
  88         }
  89
  90         my $date = DateTime->now;
  91
  92         if ($hook_hash{$def->hook}->passive eq 'f') {
  93
  94             if (my $dfield = $def->delay_field) {
  95                 if ($target->$dfield()) {
  96                     $date = DateTime::Format::ISO8601->new->parse_datetime( cleanse_ISO8601($target->$dfield) );
  97                 } else {
  98                     next;
  99                 }
 100             }
 101
 102             $date->add( seconds => interval_to_seconds($def->delay) );
 103         }
 104
 105         my $event = Fieldmapper::action_trigger::event->new();
 106         $event->target( $ident_value );
 107         $event->event_def( $def->id );
 108         $event->run_time( $date->strftime( '%F %T%z' ) );
 109         $event->user_data( OpenSRF::Utils::JSON->perl2JSON($user_data) ) if (defined($user_data));
 110
 111         $editor->create_action_trigger_event( $event );
 112
 113         $client->respond( $event->id );
 114     }
 115
 116     $editor->commit;
 117
 118     return undef;
 119 }
 120 __PACKAGE__->register_method(
 121     api_name => 'open-ils.trigger.event.autocreate',
 122     method   => 'create_active_events_for_object',
 123     api_level=> 1,
 124     stream   => 1,
 125     argc     => 3
 126 );
 127
 128 sub create_event_for_object_and_def {
 129     my $self = shift;
 130     my $client = shift;
 131     my $definitions = shift;
 132     my $target = shift;
 133     my $location = shift;
 134     my $user_data = shift;
 135
 136     my $ident = $target->Identity;
 137     my $ident_value = $target->$ident();
 138
 139     my @active = ($self->api_name =~ /inactive/o) ? () : ( active => 't' );
 140
 141     my $editor = new_editor(xact=>1);
 142
 143     my $orgs = $editor->json_query({ from => [ 'actor.org_unit_ancestors' => $location ] });
 144     my $defs = $editor->search_action_trigger_event_definition(
 145         { id => $definitions,
 146           owner  => [ map { $_->{id} } @$orgs  ],
 147           @active
 148         }
 149     );
 150
 151     my $hooks = $editor->search_action_trigger_hook(
 152         { key       => [ map { $_->hook } @$defs ],
 153           core_type => $target->json_hint
 154         }
 155     );
 156
 157     my %hook_hash = map { ($_->key, $_) } @$hooks;
 158
 159     for my $def ( @$defs ) {
 160
 161         if ($def->usr_field && $def->opt_in_setting) {
 162             my $ufield = $def->usr_field;
 163             my $uid = $target->$ufield;
 164             $uid = $uid->id if (ref $uid); # fleshed user object, unflesh it
 165
 166             my $opt_in_setting = $editor->search_actor_user_setting(
 167                 { usr   => $uid,
 168                   name  => $def->opt_in_setting,
 169                   value => 'true'
 170                 }
 171             );
 172
 173             next unless (@$opt_in_setting);
 174         }
 175
 176         my $date = DateTime->now;
 177
 178         if ($hook_hash{$def->hook}->passive eq 'f') {
 179
 180             if (my $dfield = $def->delay_field) {
 181                 if ($target->$dfield()) {
 182                     $date = DateTime::Format::ISO8601->new->parse_datetime( cleanse_ISO8601($target->$dfield) );
 183                 } else {
 184                     next;
 185                 }
 186             }
 187
 188             $date->add( seconds => interval_to_seconds($def->delay) );
 189         }
 190
 191         my $event = Fieldmapper::action_trigger::event->new();
 192         $event->target( $ident_value );
 193         $event->event_def( $def->id );
 194         $event->run_time( $date->strftime( '%F %T%z' ) );
 195         $event->user_data( OpenSRF::Utils::JSON->perl2JSON($user_data) ) if (defined($user_data));
 196
 197         $editor->create_action_trigger_event( $event );
 198
 199         $client->respond( $event->id );
 200     }
 201
 202     $editor->commit;
 203
 204     return undef;
 205 }
 206 __PACKAGE__->register_method(
 207     api_name => 'open-ils.trigger.event.autocreate.by_definition',
 208     method   => 'create_event_for_object_and_def',
 209     api_level=> 1,
 210     stream   => 1,
 211     argc     => 3
 212 );
 213 __PACKAGE__->register_method(
 214     api_name => 'open-ils.trigger.event.autocreate.by_definition.include_inactive',
 215     method   => 'create_event_for_object_and_def',
 216     api_level=> 1,
 217     stream   => 1,
 218     argc     => 3
 219 );
 220
 221
 222 # Retrieves events by object, or object type + filter
 223 #  $object : a target object or object type (class hint)
 224 #
 225 #  $filter : an optional hash of filters ... top level keys:
 226 #     event
 227 #        filters on the atev objects, such as states or null-ness of timing
 228 #        fields. contains the effective default of:
 229 #          { state => 'pending' }
 230 #        an example, which overrides the default, and will find
 231 #        stale 'found' events:
 232 #          { state => 'found', update_time => { '<' => 'yesterday' } }
 233 #
 234 #      event_def
 235 #        filters on the atevdef object. contains the effective default of:
 236 #          { active => 't' }
 237 #
 238 #      hook
 239 #        filters on the hook object. no defaults, but there is a pinned,
 240 #        unchangeable filter based on the passed hint or object type (see
 241 #        $object above). an example for finding passive events:
 242 #          { passive => 't' }
 243 #
 244 #     target
 245 #        filters against the target field on the event. this can contain
 246 #        either an array of target ids (if you passed an object type, and
 247 #        not an object) or can contain a json_query that will return exactly
 248 #        a list of target-type ids.  If you pass an object, the pkey value of
 249 #        that object will be used as a filter in addition to the filter passed
 250 #        in here.  example filter for circs of user 1234 that are open:
 251 #          { select => { circ => ['id'] },
 252 #            from => 'circ',
 253 #            where => {
 254 #              usr => 1234,
 255 #              checkin_time => undef,
 256 #              '-or' => [
 257 #                { stop_fines => undef },
 258 #                { stop_fines => { 'not in' => ['LOST','LONGOVERDUE','CLAIMSRETURNED'] } }
 259 #              ]
 260 #            }
 261
 262 sub events_by_target {
 263     my $self = shift;
 264     my $client = shift;
 265     my $object = shift;
 266     my $filter = shift || {};
 267     my $flesh_fields = shift || {};
 268     my $flesh_depth = shift || 1;
 269
 270     my $obj_class = ref($object) || _fm_class_by_hint($object);
 271     my $obj_hint = ref($object) ? _fm_hint_by_class(ref($object)) : $object;
 272
 273     my $object_ident_field = $obj_class->Identity;
 274
 275     my $query = {
 276         select => { atev => ["id"] },
 277         from   => {
 278             atev => {
 279                 atevdef => {
 280                     field => "id",
 281                     fkey => "event_def",
 282                     join => {
 283                         ath => { field => "key", fkey => "hook" }
 284                     }
 285                 }
 286             }
 287         },
 288         where  => {
 289             "+ath"  => { core_type => $obj_hint },
 290             "+atevdef" => { active => 't' },
 291             "+atev" => { state => 'pending' }
 292         },
 293         order_by => { "atev" => [ 'run_time', 'add_time' ] }
 294     };
 295
 296     $query->{limit} = $filter->{limit} if defined $filter->{limit};
 297     $query->{offset} = $filter->{offset} if defined $filter->{offset};
 298     $query->{order_by} = $filter->{order_by} if defined $filter->{order_by};
 299
 300
 301     # allow multiple 'target' filters
 302     $query->{where}->{'+atev'}->{'-and'} = [];
 303
 304     # if we got a real object, filter on its pkey value
 305     if (ref($object)) { # pass an object, require that target
 306         push @{ $query->{where}->{'+atev'}->{'-and'} },
 307             { target => $object->$object_ident_field }
 308     }
 309
 310     # we have a fancy complex target filter or a list of target ids
 311     if ($$filter{target}) {
 312         push @{ $query->{where}->{'+atev'}->{'-and'} },
 313             { target => {in => $$filter{target} } };
 314     }
 315
 316     # pass no target filter or object, you get no events
 317     if (!@{ $query->{where}->{'+atev'}->{'-and'} }) {
 318         return undef;
 319     }
 320
 321     # any hook filters, other than the required core_type filter
 322     if ($$filter{hook}) {
 323         $query->{where}->{'+ath'}->{$_} = $$filter{hook}{$_}
 324             for (grep { $_ ne 'core_type' } keys %{$$filter{hook}});
 325     }
 326
 327     # any event_def filters.  defaults to { active => 't' }
 328     if ($$filter{event_def}) {
 329         $query->{where}->{'+atevdef'}->{$_} = $$filter{event_def}{$_}
 330             for (keys %{$$filter{event_def}});
 331     }
 332
 333     # any event filters.  defaults to { state => 'pending' }.
 334     # don't overwrite '-and' used for multiple target filters above
 335     if ($$filter{event}) {
 336         $query->{where}->{'+atev'}->{$_} = $$filter{event}{$_}
 337             for (grep { $_ ne '-and' } keys %{$$filter{event}});
 338     }
 339
 340     my $e = new_editor(xact=>1);
 341
 342     my $events = $e->json_query($query);
 343
 344     $flesh_fields->{atev} = ['event_def'] unless $flesh_fields->{atev};
 345
 346     for my $id (@$events) {
 347         my $event = $e->retrieve_action_trigger_event([
 348             $id->{id},
 349             {flesh => $flesh_depth, flesh_fields => $flesh_fields}
 350         ]);
 351
 352         (my $meth = $obj_class) =~ s/^Fieldmapper:://o;
 353         $meth =~ s/::/_/go;
 354         $meth = 'retrieve_'.$meth;
 355
 356         $event->target($e->$meth($event->target));
 357         $client->respond($event);
 358     }
 359
 360     return undef;
 361 }
 362 __PACKAGE__->register_method(
 363     api_name => 'open-ils.trigger.events_by_target',
 364     method   => 'events_by_target',
 365     api_level=> 1,
 366     stream   => 1,
 367     argc     => 2
 368 );
 369
 370 sub _fm_hint_by_class {
 371     my $class = shift;
 372     return Fieldmapper->publish_fieldmapper->{$class}->{hint};
 373 }
 374
 375 sub _fm_class_by_hint {
 376     my $hint = shift;
 377
 378     my ($class) = grep {
 379         Fieldmapper->publish_fieldmapper->{$_}->{hint} eq $hint
 380     } keys %{ Fieldmapper->publish_fieldmapper };
 381
 382     return $class;
 383 }
 384
 385 sub create_batch_events {
 386     my $self = shift;
 387     my $client = shift;
 388     my $key = shift;
 389     my $location_field = shift; # where to look for event_def.owner filtering ... circ_lib, for instance, where hook.core_type = circ
 390     my $filter = shift || {};
 391     my $granularity = shift;
 392     my $user_data = shift;
 393
 394     my $active = ($self->api_name =~ /active/o) ? 1 : 0;
 395     if ($active && !keys(%$filter)) {
 396         $log->info("Active batch event creation requires a target filter but none was supplied to create_batch_events");
 397         return undef;
 398     }
 399
 400     return undef unless ($key && $location_field);
 401
 402     my $editor = new_editor(xact=>1);
 403     my $hooks = $editor->search_action_trigger_hook(
 404         { passive => $active ? 'f' : 't', key => $key }
 405     );
 406
 407     my %hook_hash = map { ($_->key, $_) } @$hooks;
 408
 409     my $defs = $editor->search_action_trigger_event_definition(
 410         { hook   => [ keys %hook_hash ], active => 't' },
 411     );
 412
 413     my $orig_filter_and = [];
 414     if ($$filter{'-and'}) {
 415         for my $f ( @{ $$filter{'-and'} } ) {
 416             push @$orig_filter_and, $f;
 417         }
 418     }
 419
 420     for my $def ( @$defs ) {
 421         next if ($granularity && $def->granularity ne $granularity );
 422
 423         my $date = DateTime->now->subtract( seconds => interval_to_seconds($def->delay) );
 424
 425         # we may need to do some work to backport this to 1.2
 426         $filter->{ $location_field } = { 'in' =>
 427             {
 428                 select  => { aou => [{ column => 'id', transform => 'actor.org_unit_descendants', result_field => 'id' }] },
 429                 from    => 'aou',
 430                 where   => { id => $def->owner }
 431             }
 432         };
 433
 434         my $run_time = 'now';
 435         if ($active) {
 436             $run_time =
 437                 DateTime
 438                     ->now
 439                     ->add( seconds => interval_to_seconds($def->delay) )
 440                     ->strftime( '%F %T%z' );
 441         } else {
 442             if ($def->max_delay) {
 443                 my @times = sort {$a <=> $b} interval_to_seconds($def->delay), interval_to_seconds($def->max_delay);
 444                 $filter->{ $def->delay_field } = {
 445                     'between' => [
 446                         DateTime->now->subtract( seconds => $times[1] )->strftime( '%F %T%z' ),
 447                         DateTime->now->subtract( seconds => $times[0] )->strftime( '%F %T%z' )
 448                     ]
 449                 };
 450             } else {
 451                 $filter->{ $def->delay_field } = {
 452                     '<=' => DateTime->now->subtract( seconds => interval_to_seconds($def->delay) )->strftime( '%F %T%z' )
 453                 };
 454             }
 455         }
 456
 457         my $class = _fm_class_by_hint($hook_hash{$def->hook}->core_type);
 458
 459         # filter where this target has an event (and it's pending, for active hooks)
 460         $$filter{'-and'} = [];
 461         for my $f ( @$orig_filter_and ) {
 462             push @{ $$filter{'-and'} }, $f;
 463         }
 464
 465         my $join = { 'join' => {
 466             atev => {
 467                 field => 'target',
 468                 fkey => $class->Identity,
 469                 type => 'left',
 470                 filter => { event_def => $def->id }
 471             }
 472         }};
 473
 474         push @{ $filter->{'-and'} }, { '+atev' => { id => undef } };
 475
 476         if ($def->usr_field && $def->opt_in_setting) {
 477             push @{ $filter->{'-and'} }, {
 478                 '-exists' => {
 479                     from  => 'aus',
 480                     where => {
 481                         name => $def->opt_in_setting,
 482                         usr  => { '=' => { '+' . $hook_hash{$def->hook}->core_type => $def->usr_field } },
 483                         value=> 'true'
 484                     }
 485                 }
 486             };
 487         }
 488
 489         $class =~ s/^Fieldmapper:://o;
 490         $class =~ s/::/_/go;
 491         my $method = 'search_'. $class;
 492
 493         # for cleaner logging
 494         my $def_id = $def->id;
 495         my $hook = $def->hook;
 496
 497         $logger->info("trigger: create_batch_events() collecting object IDs for def=$def_id / hook=$hook");
 498
 499         my $object_ids = $editor->$method( [$filter, $join], {idlist => 1, timeout => 10800} );
 500
 501         if($object_ids) {
 502             $logger->info("trigger: create_batch_events() fetched ".scalar(@$object_ids)." object IDs for def=$def_id / hook=$hook");
 503         } else {
 504             $logger->warn("trigger: create_batch_events() timeout occurred collecting object IDs for def=$def_id / hook=$hook");
 505         }
 506
 507         for my $o_id (@$object_ids) {
 508
 509             my $event = Fieldmapper::action_trigger::event->new();
 510             $event->target( $o_id );
 511             $event->event_def( $def->id );
 512             $event->run_time( $run_time );
 513             $event->user_data( OpenSRF::Utils::JSON->perl2JSON($user_data) ) if (defined($user_data));
 514
 515             $editor->create_action_trigger_event( $event );
 516
 517             $client->respond( $event->id );
 518         }
 519
 520         $logger->info("trigger: create_batch_events() successfully created events for def=$def_id / hook=$hook");
 521     }
 522
 523     $logger->info("trigger: create_batch_events() done creating events");
 524
 525     $editor->commit;
 526
 527     return undef;
 528 }
 529 __PACKAGE__->register_method(
 530     api_name => 'open-ils.trigger.passive.event.autocreate.batch',
 531     method   => 'create_batch_events',
 532     api_level=> 1,
 533     stream   => 1,
 534     argc     => 2
 535 );
 536
 537 __PACKAGE__->register_method(
 538     api_name => 'open-ils.trigger.active.event.autocreate.batch',
 539     method   => 'create_batch_events',
 540     api_level=> 1,
 541     stream   => 1,
 542     argc     => 2
 543 );
 544
 545 sub fire_single_event {
 546     my $self = shift;
 547     my $client = shift;
 548     my $event_id = shift;
 549
 550     my $e = OpenILS::Application::Trigger::Event->new($event_id);
 551
 552     if ($e->validate->valid) {
 553         $logger->info("trigger: Event is valid, reacting...");
 554         $e->react->cleanup;
 555     }
 556
 557     $e->editor->disconnect;
 558     OpenILS::Application::Trigger::Event->ClearObjectCache();
 559
 560     return {
 561         valid     => $e->valid,
 562         reacted   => $e->reacted,
 563         cleanedup => $e->cleanedup,
 564         event     => $e->event
 565     };
 566 }
 567 __PACKAGE__->register_method(
 568     api_name => 'open-ils.trigger.event.fire',
 569     method   => 'fire_single_event',
 570     api_level=> 1,
 571     argc     => 1
 572 );
 573
 574 sub fire_event_group {
 575     my $self = shift;
 576     my $client = shift;
 577     my $events = shift;
 578
 579     my $e = OpenILS::Application::Trigger::EventGroup->new(@$events);
 580
 581     if ($e->validate->valid) {
 582         $logger->info("trigger: Event group is valid, reacting...");
 583         $e->react->cleanup;
 584     }
 585
 586     $e->editor->disconnect;
 587     OpenILS::Application::Trigger::Event->ClearObjectCache();
 588
 589     return {
 590         valid     => $e->valid,
 591         reacted   => $e->reacted,
 592         cleanedup => $e->cleanedup,
 593         events    => [map { $_->event } @{$e->events}]
 594     };
 595 }
 596 __PACKAGE__->register_method(
 597     api_name => 'open-ils.trigger.event_group.fire',
 598     method   => 'fire_event_group',
 599     api_level=> 1,
 600     argc     => 1
 601 );
 602
 603 sub pending_events {
 604     my $self = shift;
 605     my $client = shift;
 606     my $granularity = shift;
 607     my $granflag = shift;
 608
 609     my $query = [{ state => 'pending', run_time => {'<' => 'now'} }, { order_by => { atev => [ qw/run_time add_time/] }, 'join' => 'atevdef' }];
 610
 611     if (defined $granularity) {
 612         if ($granflag) {
 613             $query->[0]->{'+atevdef'} = {granularity => $granularity};
 614         } else {
 615             $query->[0]->{'+atevdef'} = {'-or' => [ {granularity => $granularity}, {granularity => undef} ] };
 616         }
 617     } else {
 618         $query->[0]->{'+atevdef'} = {granularity => undef};
 619     }
 620
 621     return new_editor(xact=>1)->search_action_trigger_event(
 622         $query, { idlist=> 1, timeout => 7200, substream => 1 }
 623     );
 624 }
 625 __PACKAGE__->register_method(
 626     api_name => 'open-ils.trigger.event.find_pending',
 627     method   => 'pending_events',
 628     api_level=> 1
 629 );
 630
 631 sub gather_events {
 632     my $self = shift;
 633     my $client = shift;
 634     my $e_ids = shift;
 635
 636     $e_ids = [$e_ids] if (!ref($e_ids));
 637
 638     my @events;
 639     for my $e_id (@$e_ids) {
 640         my $e;
 641         try {
 642            $e = OpenILS::Application::Trigger::Event->new($e_id);
 643         } catch Error with {
 644             $logger->error("trigger: Event creation failed with ".shift());
 645         };
 646
 647         next if !$e or $e->event->state eq 'invalid';
 648
 649         try {
 650             $e->build_environment;
 651         } catch Error with {
 652             $logger->error("trigger: Event environment building failed with ".shift());
 653         };
 654
 655         $e->editor->disconnect;
 656         $e->environment->{EventProcessor} = undef; # remove circular ref for json encoding
 657         $client->respond($e);
 658     }
 659
 660     OpenILS::Application::Trigger::Event->ClearObjectCache();
 661
 662     return undef;
 663 }
 664 __PACKAGE__->register_method(
 665     api_name => 'open-ils.trigger.event.gather',
 666     method   => 'gather_events',
 667     api_level=> 1
 668 );
 669
 670 sub grouped_events {
 671     my $self = shift;
 672     my $client = shift;
 673     my $granularity = shift;
 674     my $granflag = shift;
 675
 676     my ($events) = $self->method_lookup('open-ils.trigger.event.find_pending')->run($granularity, $granflag);
 677
 678     my %groups = ( '*' => [] );
 679
 680     if($events) {
 681         $logger->info("trigger: grouped_events found ".scalar(@$events)." pending events to process");
 682     } else {
 683         $logger->warn("trigger: grouped_events timed out loading pending events");
 684         return \%groups;
 685     }
 686
 687     my @fleshed_events;
 688
 689     if ($parallel_collect == 1 or @$events == 1) { # use method lookup
 690         @fleshed_events = $self->method_lookup('open-ils.trigger.event.gather')->run($events);
 691     } else {
 692         my $self_multi = OpenSRF::MultiSession->new(
 693             app                 => 'open-ils.trigger',
 694             cap                 => $parallel_collect,
 695             success_handler     => sub {
 696                 my $self = shift;
 697                 my $req = shift;
 698
 699                 push @fleshed_events,
 700                     map { OpenILS::Application::Trigger::Event->new($_) }
 701                     map { $_->content }
 702                     @{ $req->{response} };
 703             },
 704         );
 705
 706         $self_multi->request( 'open-ils.trigger.event.gather' => $_ ) for ( @$events );
 707         $client->status( new OpenSRF::DomainObject::oilsContinueStatus );
 708
 709         $self_multi->session_wait(1);
 710         $client->status( new OpenSRF::DomainObject::oilsContinueStatus );
 711     }
 712
 713     for my  $e (@fleshed_events) {
 714         if (my $group = $e->event->event_def->group_field) {
 715
 716             # split the grouping link steps
 717             my @steps = split /\./, $group;
 718             my $group_field = pop(@steps); # we didn't flesh to this, it's a field not an object
 719
 720             my $node;
 721             eval {
 722                 $node = $e->target;
 723                 $node = $node->$_() for ( @steps );
 724             };
 725
 726             unless($node) { # should not get here, but to be safe..
 727                 $e->update_state('invalid');
 728                 next;
 729             }
 730
 731             # get the grouping value for the grouping object on this event
 732             my $ident_value = $node->$group_field();
 733             if(ref $ident_value) {
 734                 my $ident_field = $ident_value->Identity;
 735                 $ident_value = $ident_value->$ident_field()
 736             }
 737
 738             # push this event onto the event+grouping_value stack
 739             $groups{$e->event->event_def->id}{$ident_value} ||= [];
 740             push @{ $groups{$e->event->event_def->id}{$ident_value} }, $e;
 741         } else {
 742             # it's a non-grouped event
 743             push @{ $groups{'*'} }, $e;
 744         }
 745     }
 746
 747
 748     return \%groups;
 749 }
 750 __PACKAGE__->register_method(
 751     api_name => 'open-ils.trigger.event.find_pending_by_group',
 752     method   => 'grouped_events',
 753     api_level=> 1
 754 );
 755
 756 sub run_all_events {
 757     my $self = shift;
 758     my $client = shift;
 759     my $granularity = shift;
 760     my $granflag = shift;
 761
 762     my ($groups) = $self->method_lookup('open-ils.trigger.event.find_pending_by_group')->run($granularity, $granflag);
 763     $client->respond({"status" => "found"}) if (keys(%$groups) > 1 || @{$$groups{'*'}});
 764
 765     my $self_multi;
 766     if ($parallel_react > 1 and (keys(%$groups) > 1 || @{$$groups{'*'}} > 1)) {
 767         $self_multi = OpenSRF::MultiSession->new(
 768             app                   => 'open-ils.trigger',
 769             cap                   => $parallel_react,
 770             session_hash_function => sub {
 771                 my $args = shift;
 772                 return $args->{target_id};
 773             },
 774             success_handler       => sub {
 775                 my $me = shift;
 776                 my $req = shift;
 777                 $client->respond( $req->{response}->[0]->content );
 778             }
 779         );
 780     }
 781
 782     for my $def ( keys %$groups ) {
 783         if ($def eq '*') {
 784             $logger->info("trigger: run_all_events firing un-grouped events");
 785             for my $event ( @{ $$groups{'*'} } ) {
 786                 try {
 787                     if ($self_multi) {
 788                         $event->environment->{EventProcessor} = undef; # remove circular ref for json encoding
 789                         $self_multi->request({target_id => $event->id}, 'open-ils.trigger.event.fire', $event);
 790                     } else {
 791                         $client->respond(
 792                             $self
 793                                 ->method_lookup('open-ils.trigger.event.fire')
 794                                 ->run($event)
 795                         );
 796                     }
 797                 } catch Error with {
 798                     $logger->error("trigger: event firing failed with ".shift());
 799                 };
 800             }
 801             $logger->info("trigger: run_all_events completed queuing un-grouped events");
 802             $client->status( new OpenSRF::DomainObject::oilsContinueStatus );
 803
 804         } else {
 805             my $defgroup = $$groups{$def};
 806             $logger->info("trigger: run_all_events firing events for grouped event def=$def");
 807             for my $ident ( keys %$defgroup ) {
 808                 $logger->info("trigger: run_all_events firing group for grouped event def=$def and grp ident $ident");
 809                 try {
 810                     if ($self_multi) {
 811                         $_->environment->{EventProcessor} = undef for @{$$defgroup{$ident}}; # remove circular ref for json encoding
 812                         $self_multi->request({target_id => $ident}, 'open-ils.trigger.event_group.fire', $$defgroup{$ident});
 813                     } else {
 814                         $client->respond(
 815                             $self
 816                                 ->method_lookup('open-ils.trigger.event_group.fire')
 817                                 ->run($$defgroup{$ident})
 818                         );
 819                     }
 820                     $client->status( new OpenSRF::DomainObject::oilsContinueStatus );
 821                 } catch Error with {
 822                     $logger->error("trigger: event firing failed with ".shift());
 823                 };
 824             }
 825             $logger->info("trigger: run_all_events completed queuing events for grouped event def=$def");
 826         }
 827     }
 828
 829     $self_multi->session_wait(1) if ($self_multi);
 830     $logger->info("trigger: run_all_events completed firing events");
 831
 832     $client->respond_complete();
 833     return undef;
 834 }
 835 __PACKAGE__->register_method(
 836     api_name => 'open-ils.trigger.event.run_all_pending',
 837     method   => 'run_all_events',
 838     api_level=> 1
 839 );
 840
 841
 842 1;