tags/0.4.3.1-pre2/SrSv/Message.pm

   1 #       This file is part of SurrealServices.
   2 #
   3 #       SurrealServices is free software; you can redistribute it and/or modify
   4 #       it under the terms of the GNU General Public License as published by
   5 #       the Free Software Foundation; either version 2 of the License, or
   6 #       (at your option) any later version.
   7 #
   8 #       SurrealServices is distributed in the hope that it will be useful,
   9 #       but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 #       MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 #       GNU General Public License for more details.
  12 #
  13 #       You should have received a copy of the GNU General Public License
  14 #       along with SurrealServices; if not, write to the Free Software
  15 #       Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  16
  17 package SrSv::Message;
  18
  19 use strict;
  20
  21 use Exporter 'import';
  22 BEGIN { our @EXPORT_OK = qw(add_callback message call_callback unit_finished current_message) }
  23
  24 use Carp;
  25 use Storable qw(fd_retrieve store_fd);
  26
  27 use SrSv::Debug;
  28 BEGIN {
  29         if(DEBUG) {
  30                 require Data::Dumper; import Data::Dumper ();
  31         }
  32 }
  33
  34 use SrSv::Process::Call qw(safe_call);
  35 use SrSv::Process::Worker qw(ima_worker get_socket multi call_in_parent call_all_child do_callback_in_child);
  36
  37 our %callbacks_by_trigger_class;
  38 our %callbacks_by_after;
  39 our %callbacks_by_name;
  40
  41 our $current_message;
  42
  43 ### Public functions
  44
  45 sub add_callback($) {
  46         my ($callback) = @_;
  47
  48         if(multi) {
  49                 croak "Callbacks cannot be added at runtime";
  50         }
  51
  52         if(my $after = $callback->{AFTER}) {
  53                 push @{$callbacks_by_after{$after}}, $callback;
  54         }
  55
  56         $callback->{NAME} = $callback->{CALL} unless $callback->{NAME};
  57         if(my $name = $callback->{NAME}) {
  58                 push @{$callbacks_by_name{$name}}, $callback;
  59         }
  60
  61         if(my $trigger = $callback->{TRIGGER_COND}{CLASS}) {
  62                 push @{$callbacks_by_trigger_class{$trigger}}, $callback;
  63         }
  64
  65         if(DEBUG()) {
  66                 print "Added callback: $callback->{NAME}\n";
  67         }
  68 }
  69
  70 sub message($) {
  71         my ($message) = @_;
  72
  73         if(ima_worker()) {
  74                 if($message->{SYNC}) {
  75                         print "Triggered a sync callback!\n" if DEBUG();
  76                         trigger_callbacks($message);
  77                 } else {
  78                         store_fd($message, get_socket());
  79                         fd_retrieve(get_socket());
  80                 }
  81                 return;
  82         }
  83
  84         trigger_callbacks($message);
  85 }
  86
  87 ### Semi-private ###
  88
  89 sub call_callback {
  90         my ($callback, $message) = @_;
  91
  92         local $current_message = $message;
  93
  94         if(my $call = $callback->{CALL}) {
  95                 safe_call($call, [$message, $callback]);
  96         }
  97 }
  98
  99 sub unit_finished($$) {
 100         my ($callback, $message) = @_;
 101
 102         if(DEBUG()) {
 103                 print "--- Finished unit\nCallback: $callback->{NAME}\nMessage: $message->{CLASS}\n";
 104         }
 105
 106         safe_call($callback->{ON_FINISH}, [$callback, $message]) if $callback->{ON_FINISH};
 107
 108         $message->{_CB_COUNTDOWN}--;
 109         print "_CB_COUNTDOWN is $message->{_CB_COUNTDOWN}\n---\n" if DEBUG;
 110
 111         $message->{_CB_DONE}{$callback->{NAME}} = 1;
 112
 113         if(!$message->{SYNC} and defined($message->{_CB_QUEUE}) and @{$message->{_CB_QUEUE}}) {
 114                 trigger_callbacks($message);
 115         }
 116
 117         if($message->{_CB_COUNTDOWN} == 0) {
 118                 message_finished($message);
 119         }
 120 }
 121
 122 sub message_finished($) {
 123         my ($message) = @_;
 124
 125         print "Message finished: $message->{CLASS}\n" if DEBUG;
 126
 127         for(qw(_CB_QUEUE _CB_COUNTDOWN _CB_DONE _CB_TODO)) {
 128                 undef $message->{$_};
 129         }
 130
 131         safe_call($message->{ON_FINISH}, [$message]) if $message->{ON_FINISH};
 132 }
 133
 134 ### Private functions ###
 135
 136 sub trigger_callbacks($) {
 137         my ($message) = @_;
 138
 139         my $callbacks;
 140
 141         if(defined($message->{_CB_QUEUE})) {
 142                 $callbacks = $message->{_CB_QUEUE};
 143         } else {
 144                 $callbacks = get_matching_callbacks($message);
 145         }
 146
 147         if(@$callbacks) {
 148                 $message->{_CB_COUNTDOWN} = @$callbacks unless defined($message->{_CB_COUNTDOWN});
 149
 150                 my $do_next = [];
 151
 152                 foreach my $callback (@$callbacks) {
 153                         my $after = $callback->{AFTER};
 154                         if($after and $message->{_CB_TODO}{$after} and not $message->{_CB_DONE}{$after}) {
 155                                 push @$do_next, $callback;
 156                         } else {
 157                                 do_unit($callback, $message);
 158                         }
 159                 }
 160
 161                 $message->{_CB_QUEUE} = $do_next;
 162
 163                 goto &trigger_callbacks if($message->{SYNC} and @$do_next > 0);
 164         }
 165
 166         else {
 167                 if(DEBUG) {
 168                         print "Message with no callbacks: ".Dumper($message);
 169                 }
 170
 171                 message_finished($message);
 172         }
 173 }
 174
 175 sub do_unit($$) {
 176         my ($callback, $message) = @_;
 177
 178         if(!multi or $callback->{PARENTONLY} or $message->{SYNC}) {
 179                 call_callback($callback, $message);
 180                 unit_finished($callback, $message);
 181         } else {
 182                 do_callback_in_child($callback, $message);
 183         }
 184 }
 185
 186 sub get_matching_callbacks($) {
 187         my ($message) = @_;
 188         my $ret = [];
 189
 190         my $class = $message->{CLASS};
 191
 192         foreach my $callback (@{$callbacks_by_trigger_class{$class}}) {
 193                 if(callback_matches($message, $callback)) {
 194                         push @$ret, $callback;
 195                         $message->{_CB_TODO}{$callback->{NAME}} = 1;
 196                 }
 197         }
 198
 199         return $ret;
 200 }
 201
 202 sub callback_matches($$) {
 203         my ($message, $callback) = @_;
 204
 205         foreach my $cond (keys(%{$callback->{TRIGGER_COND}})) {
 206                 if(ref($callback->{TRIGGER_COND}{$cond}) eq 'Regexp') {
 207                         return 0 if defined($message->{$cond}) && !($message->{$cond} =~ $callback->{TRIGGER_COND}{$cond});
 208                 } else {
 209                         return 0 if defined($message->{$cond}) && !(lc $message->{$cond} eq lc $callback->{TRIGGER_COND}{$cond});
 210                 }
 211         }
 212
 213         return 1;
 214 }
 215
 216 sub current_message() { return $current_message }
 217
 218 1;