]> scm.dxcluster.org Git - spider.git/blob - perl/DXMsg.pm
allowed a queue_msg after a PC33
[spider.git] / perl / DXMsg.pm
1 #!/usr/bin/perl
2 #
3 # This module impliments the message handling for a dx cluster
4 #
5 # Copyright (c) 1998 Dirk Koopman G1TLH
6 #
7 # $Id$
8 #
9 #
10 # Notes for implementors:-
11 #
12 # PC28 field 11 is the RR required flag
13 # PC28 field 12 is a VIA routing (ie it is a node call) 
14 #
15
16 package DXMsg;
17
18 @ISA = qw(DXProt DXChannel);
19
20 use DXUtil;
21 use DXChannel;
22 use DXUser;
23 use DXM;
24 use DXCluster;
25 use DXProtVars;
26 use DXProtout;
27 use DXDebug;
28 use DXLog;
29 use IO::File;
30 use Fcntl;
31 use Carp;
32
33 use strict;
34 use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean
35                         @badmsg $badmsgfn $forwardfn @forward $timeout $waittime
36                     $queueinterval $lastq);
37
38 %work = ();                                             # outstanding jobs
39 @msg = ();                                              # messages we have
40 %busy = ();                                             # station interlocks
41 $msgdir = "$main::root/msg";    # directory contain the msgs
42 $maxage = 30 * 86400;                   # the maximum age that a message shall live for if not marked 
43 $last_clean = 0;                                # last time we did a clean
44 @forward = ();                  # msg forward table
45 $timeout = 30*60;               # forwarding timeout
46 $waittime = 60*60;              # time an aborted outgoing message waits before trying again
47 $queueinterval = 5*60;          # run the queue every 5 minutes
48 $lastq = 0;
49
50
51 $badmsgfn = "$msgdir/badmsg.pl";  # list of TO address we wont store
52 $forwardfn = "$msgdir/forward.pl";  # the forwarding table
53
54 %valid = (
55                   fromnode => '9,From Node',
56                   tonode => '9,To Node',
57                   to => '0,To',
58                   from => '0,From',
59                   t => '0,Msg Time,cldatetime',
60                   private => '9,Private',
61                   subject => '0,Subject',
62                   linesreq => '0,Lines per Gob',
63                   rrreq => '9,Read Confirm',
64                   origin => '0,Origin',
65                   lines => '5,Data',
66                   stream => '9,Stream No',
67                   count => '9,Gob Linecnt',
68                   file => '9,File?,yesno',
69                   gotit => '9,Got it Nodes,parray',
70                   lines => '9,Lines,parray',
71                   'read' => '9,Times read',
72                   size => '0,Size',
73                   msgno => '0,Msgno',
74                   keep => '0,Keep this?,yesno',
75                   lastt => '9,Last processed,cldatetime',
76                   waitt => '9,Wait until,cldatetime',
77                  );
78
79 sub DESTROY
80 {
81         my $self = shift;
82         undef $self->{lines};
83         undef $self->{gotit};
84 }
85
86 # allocate a new object
87 # called fromnode, tonode, from, to, datetime, private?, subject, nolinesper  
88 sub alloc                  
89 {
90         my $pkg = shift;
91         my $self = bless {}, $pkg;
92         $self->{msgno} = shift;
93         my $to = shift;
94         #  $to =~ s/-\d+$//o;
95         $self->{to} = ($to eq $main::mycall) ? $main::myalias : $to;
96         my $from = shift;
97         $from =~ s/-\d+$//o;
98         $self->{from} = uc $from;
99         $self->{t} = shift;
100         $self->{private} = shift;
101         $self->{subject} = shift;
102         $self->{origin} = shift;
103         $self->{'read'} = shift;
104         $self->{rrreq} = shift;
105         $self->{gotit} = [];
106         $self->{lastt} = $main::systime;
107     
108         return $self;
109 }
110
111 sub workclean
112 {
113         my $ref = shift;
114         delete $ref->{lines};
115         delete $ref->{linesreq};
116         delete $ref->{tonode};
117         delete $ref->{fromnode};
118         delete $ref->{stream};
119         delete $ref->{lines};
120         delete $ref->{file};
121         delete $ref->{count};
122         delete $ref->{lastt} if exists $ref->{lastt};
123         delete $ref->{waitt} if exists $ref->{waitt};
124 }
125
126 sub process
127 {
128         my ($self, $line) = @_;
129
130         # this is periodic processing
131         if (!$self || !$line) {
132
133                 # wander down the work queue stopping any messages that have timed out
134                 for (keys %busy) {
135                         my $node = $_;
136                         my $ref = $busy{$_};
137                         if (exists $ref->{lastt} && $main::systime > $ref->{lastt} + $timeout) {
138                                 $ref->stop_msg($node);
139
140                                 # delay any outgoing messages that fail
141                                 $ref->{waitt} = $main::systime + $waittime + rand(120) if $node ne $main::mycall;
142                         }
143                 }
144
145                 # queue some message if the interval timer has gone off
146                 if ($main::systime > $lastq + $queueinterval) {
147                         queue_msg(0);
148                         $lastq = $main::systime;
149                 }
150
151                 # clean the message queue
152                 clean_old() if $main::systime - $last_clean > 3600 ;
153                 return;
154         }
155
156         my @f = split /\^/, $line;
157         my ($pcno) = $f[0] =~ /^PC(\d\d)/; # just get the number
158
159  SWITCH: {
160                 if ($pcno == 28) {              # incoming message
161
162                         # first look for any messages in the busy queue 
163                         # and cancel them this should both resolve timed out incoming messages
164                         # and crossing of message between nodes, incoming messages have priority
165                         if (exists $busy{$f[2]}) {
166                                 my $ref = $busy{$f[2]};
167                                 my $tonode = $ref->{tonode};
168                                 $ref->stop_msg($self->call);
169                         }
170
171                         my $t = cltounix($f[5], $f[6]);
172                         my $stream = next_transno($f[2]);
173                         my $ref = DXMsg->alloc($stream, uc $f[3], $f[4], $t, $f[7], $f[8], $f[13], '0', $f[11]);
174                         
175                         # fill in various forwarding state variables
176                         $ref->{fromnode} = $f[2];
177                         $ref->{tonode} = $f[1];
178                         $ref->{rrreq} = $f[11];
179                         $ref->{linesreq} = $f[10];
180                         $ref->{stream} = $stream;
181                         $ref->{count} = 0;      # no of lines between PC31s
182                         dbg('msg', "new message from $f[4] to $f[3] '$f[8]' stream $stream\n");
183                         $work{"$f[2]$stream"} = $ref; # store in work
184                         $busy{$f[2]} = $ref; # set interlock
185                         $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack
186                         $ref->{lastt} = $main::systime;
187                         last SWITCH;
188                 }
189                 
190                 if ($pcno == 29) {              # incoming text
191                         my $ref = $work{"$f[2]$f[3]"};
192                         if ($ref) {
193                                 push @{$ref->{lines}}, $f[4];
194                                 $ref->{count}++;
195                                 if ($ref->{count} >= $ref->{linesreq}) {
196                                         $self->send(DXProt::pc31($f[2], $f[1], $f[3]));
197                                         dbg('msg', "stream $f[3]: $ref->{count} lines received\n");
198                                         $ref->{count} = 0;
199                                 }
200                                 $ref->{lastt} = $main::systime;
201                         }
202                         last SWITCH;
203                 }
204                 
205                 if ($pcno == 30) {              # this is a incoming subject ack
206                         my $ref = $work{$f[2]}; # note no stream at this stage
207                         if ($ref) {
208                                 delete $work{$f[2]};
209                                 $ref->{stream} = $f[3];
210                                 $ref->{count} = 0;
211                                 $ref->{linesreq} = 5;
212                                 $work{"$f[2]$f[3]"} = $ref;     # new ref
213                                 dbg('msg', "incoming subject ack stream $f[3]\n");
214                                 $busy{$f[2]} = $ref; # interlock
215                                 $ref->{lines} = [];
216                                 push @{$ref->{lines}}, ($ref->read_msg_body);
217                                 $ref->send_tranche($self);
218                                 $ref->{lastt} = $main::systime;
219                         } else {
220                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
221                         } 
222                         last SWITCH;
223                 }
224                 
225                 if ($pcno == 31) {              # acknowledge a tranche of lines
226                         my $ref = $work{"$f[2]$f[3]"};
227                         if ($ref) {
228                                 dbg('msg', "tranche ack stream $f[3]\n");
229                                 $ref->send_tranche($self);
230                                 $ref->{lastt} = $main::systime;
231                         } else {
232                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
233                         } 
234                         last SWITCH;
235                 }
236                 
237                 if ($pcno == 32) {              # incoming EOM
238                         dbg('msg', "stream $f[3]: EOM received\n");
239                         my $ref = $work{"$f[2]$f[3]"};
240                         if ($ref) {
241                                 $self->send(DXProt::pc33($f[2], $f[1], $f[3])); # acknowledge it
242                                 
243                                 # get the next msg no - note that this has NOTHING to do with the stream number in PC protocol
244                                 # store the file or message
245                                 # remove extraneous rubbish from the hash
246                                 # remove it from the work in progress vector
247                                 # stuff it on the msg queue
248                                 if ($ref->{lines} && @{$ref->{lines}} > 0) { # ignore messages with 0 lines
249                                         if ($ref->{file}) {
250                                                 $ref->store($ref->{lines});
251                                         } else {
252
253                                                 # does an identical message already exist?
254                                                 my $m;
255                                                 for $m (@msg) {
256                                                         if ($ref->{subject} eq $m->{subject} && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from}) {
257                                                                 $ref->stop_msg($self->call);
258                                                                 my $msgno = $m->{msgno};
259                                                                 dbg('msg', "duplicate message to $msgno\n");
260                                                                 Log('msg', "duplicate message to $msgno");
261                                                                 return;
262                                                         }
263                                                 }
264                                                         
265                                                 # look for 'bad' to addresses 
266                                                 if (grep $ref->{to} eq $_, @badmsg) {
267                                                         $ref->stop_msg($self->call);
268                                                         dbg('msg', "'Bad' TO address $ref->{to}");
269                                                         Log('msg', "'Bad' TO address $ref->{to}");
270                                                         return;
271                                                 }
272
273                                                 $ref->{msgno} = next_transno("Msgno");
274                                                 push @{$ref->{gotit}}, $f[2]; # mark this up as being received
275                                                 $ref->store($ref->{lines});
276                                                 add_dir($ref);
277                                                 my $dxchan = DXChannel->get($ref->{to});
278                                                 $dxchan->send($dxchan->msg('m9')) if $dxchan;
279                                                 Log('msg', "Message $ref->{msgno} from $ref->{from} received from $f[2] for $ref->{to}");
280                                         }
281                                 }
282                                 $ref->stop_msg($self->call);
283                         } else {
284                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
285                         }
286                         # queue_msg(0);
287                         last SWITCH;
288                 }
289                 
290                 if ($pcno == 33) {              # acknowledge the end of message
291                         my $ref = $work{"$f[2]$f[3]"};
292                         if ($ref) {
293                                 if ($ref->{private}) { # remove it if it private and gone off site#
294                                         Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2] and deleted");
295                                         $ref->del_msg;
296                                 } else {
297                                         Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2]");
298                                         push @{$ref->{gotit}}, $f[2]; # mark this up as being received
299                                         $ref->store($ref->{lines});     # re- store the file
300                                 }
301                                 $ref->stop_msg($self->call);
302                         } else {
303                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
304                         } 
305
306                         # send next one if present
307                         queue_msg(0);
308                         last SWITCH;
309                 }
310                 
311                 if ($pcno == 40) {              # this is a file request
312                         $f[3] =~ s/\\/\//og; # change the slashes
313                         $f[3] =~ s/\.//og;      # remove dots
314                         $f[3] =~ s/^\///o;   # remove the leading /
315                         $f[3] = lc $f[3];       # to lower case;
316                         dbg('msg', "incoming file $f[3]\n");
317                         $f[3] = 'packclus/' . $f[3] unless $f[3] =~ /^packclus\//o;
318                         
319                         # create any directories
320                         my @part = split /\//, $f[3];
321                         my $part;
322                         my $fn = "$main::root";
323                         pop @part;                      # remove last part
324                         foreach $part (@part) {
325                                 $fn .= "/$part";
326                                 next if -e $fn;
327                                 last SWITCH if !mkdir $fn, 0777;
328                                 dbg('msg', "created directory $fn\n");
329                         }
330                         my $stream = next_transno($f[2]);
331                         my $ref = DXMsg->alloc($stream, "$main::root/$f[3]", $self->call, time, !$f[4], $f[3], ' ', '0', '0');
332                         
333                         # forwarding variables
334                         $ref->{fromnode} = $f[1];
335                         $ref->{tonode} = $f[2];
336                         $ref->{linesreq} = $f[5];
337                         $ref->{stream} = $stream;
338                         $ref->{count} = 0;      # no of lines between PC31s
339                         $ref->{file} = 1;
340                         $work{"$f[2]$stream"} = $ref; # store in work
341                         $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack 
342                         
343                         last SWITCH;
344                 }
345                 
346                 if ($pcno == 42) {              # abort transfer
347                         dbg('msg', "stream $f[3]: abort received\n");
348                         my $ref = $work{"$f[2]$f[3]"};
349                         if ($ref) {
350                                 $ref->stop_msg($self->call);
351                                 $ref = undef;
352                         }
353                         
354                         last SWITCH;
355                 }
356
357                 if ($pcno == 49) {      # global delete on subject
358                         for (@msg) {
359                                 if ($_->{from} eq $f[1] && $_->{subject} eq $f[2]) {
360                                         $_->del_msg();
361                                         Log('msg', "Message $_->{msgno} from $_->{from} ($_->{subject}) fully deleted");
362                                         DXProt::broadcast_ak1a($line, $self);
363                                 }
364                         }
365                 }
366         }
367 }
368
369
370 # store a message away on disc or whatever
371 #
372 # NOTE the second arg is a REFERENCE not a list
373 sub store
374 {
375         my $ref = shift;
376         my $lines = shift;
377         
378         # we only proceed if there are actually any lines in the file
379         if (!$lines || @{$lines} == 0) {
380                 return;
381         }
382         
383         if ($ref->{file}) {                     # a file
384                 dbg('msg', "To be stored in $ref->{to}\n");
385                 
386                 my $fh = new IO::File "$ref->{to}", "w";
387                 if (defined $fh) {
388                         my $line;
389                         foreach $line (@{$lines}) {
390                                 print $fh "$line\n";
391                         }
392                         $fh->close;
393                         dbg('msg', "file $ref->{to} stored\n");
394                         Log('msg', "file $ref->{to} from $ref->{from} stored" );
395                 } else {
396                         confess "can't open file $ref->{to} $!";  
397                 }
398         } else {                                        # a normal message
399
400                 # attempt to open the message file
401                 my $fn = filename($ref->{msgno});
402                 
403                 dbg('msg', "To be stored in $fn\n");
404                 
405                 # now save the file, overwriting what's there, YES I KNOW OK! (I will change it if it's a problem)
406                 my $fh = new IO::File "$fn", "w";
407                 if (defined $fh) {
408                         my $rr = $ref->{rrreq} ? '1' : '0';
409                         my $priv = $ref->{private} ? '1': '0';
410                         print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{'read'}^$rr\n";
411                         print $fh "=== ", join('^', @{$ref->{gotit}}), "\n";
412                         my $line;
413                         $ref->{size} = 0;
414                         foreach $line (@{$lines}) {
415                                 $ref->{size} += (length $line) + 1;
416                                 print $fh "$line\n";
417                         }
418                         $fh->close;
419                         dbg('msg', "msg $ref->{msgno} stored\n");
420                         Log('msg', "msg $ref->{msgno} from $ref->{from} to $ref->{to} stored" );
421                 } else {
422                         confess "can't open msg file $fn $!";  
423                 }
424         }
425 }
426
427 # delete a message
428 sub del_msg
429 {
430         my $self = shift;
431         
432         # remove it from the active message list
433         @msg = map { $_ != $self ? $_ : () } @msg;
434         
435         # belt and braces (one day I will ask someone if this is REALLY necessary)
436         delete $self->{gotit};
437         delete $self->{list};
438         
439         # remove the file
440         unlink filename($self->{msgno});
441         dbg('msg', "deleting $self->{msgno}\n");
442 }
443
444 # clean out old messages from the message queue
445 sub clean_old
446 {
447         my $ref;
448         
449         # mark old messages for deletion
450         foreach $ref (@msg) {
451                 if (!$ref->{keep} && $ref->{t} < $main::systime - $maxage) {
452                         $ref->{deleteme} = 1;
453                         delete $ref->{gotit};
454                         delete $ref->{list};
455                         unlink filename($ref->{msgno});
456                         dbg('msg', "deleting old $ref->{msgno}\n");
457                 }
458         }
459         
460         # remove them all from the active message list
461         @msg = map { $_->{deleteme} ? () : $_ } @msg;
462         $last_clean = $main::systime;
463 }
464
465 # read in a message header
466 sub read_msg_header
467
468         my $fn = shift;
469         my $file;
470         my $line;
471         my $ref;
472         my @f;
473         my $size;
474         
475         $file = new IO::File;
476         if (!open($file, $fn)) {
477                 print "Error reading $fn $!\n";
478                 return undef;
479         }
480         $size = -s $fn;
481         $line = <$file>;                        # first line
482         chomp $line;
483         $size -= length $line;
484         if (! $line =~ /^===/o) {
485                 print "corrupt first line in $fn ($line)\n";
486                 return undef;
487         }
488         $line =~ s/^=== //o;
489         @f = split /\^/, $line;
490         $ref = DXMsg->alloc(@f);
491         
492         $line = <$file>;                        # second line
493         chomp $line;
494         $size -= length $line;
495         if (! $line =~ /^===/o) {
496                 print "corrupt second line in $fn ($line)\n";
497                 return undef;
498         }
499         $line =~ s/^=== //o;
500         $ref->{gotit} = [];
501         @f = split /\^/, $line;
502         push @{$ref->{gotit}}, @f;
503         $ref->{size} = $size;
504         
505         close($file);
506         
507         return $ref;
508 }
509
510 # read in a message header
511 sub read_msg_body
512 {
513         my $self = shift;
514         my $msgno = $self->{msgno};
515         my $file;
516         my $line;
517         my $fn = filename($msgno);
518         my @out;
519         
520         $file = new IO::File;
521         if (!open($file, $fn)) {
522                 print "Error reading $fn $!\n";
523                 return undef;
524         }
525         chomp (@out = <$file>);
526         close($file);
527         
528         shift @out if $out[0] =~ /^=== /;
529         shift @out if $out[0] =~ /^=== /;
530         return @out;
531 }
532
533 # send a tranche of lines to the other end
534 sub send_tranche
535 {
536         my ($self, $dxchan) = @_;
537         my @out;
538         my $to = $self->{tonode};
539         my $from = $self->{fromnode};
540         my $stream = $self->{stream};
541         my $lines = $self->{lines};
542         my ($c, $i);
543         
544         for ($i = 0, $c = $self->{count}; $i < $self->{linesreq} && $c < @$lines; $i++, $c++) {
545                 push @out, DXProt::pc29($to, $from, $stream, $lines->[$c]);
546     }
547     $self->{count} = $c;
548
549     push @out, DXProt::pc32($to, $from, $stream) if $i < $self->{linesreq};
550         $dxchan->send(@out);
551 }
552
553         
554 # find a message to send out and start the ball rolling
555 sub queue_msg
556 {
557         my $sort = shift;
558         my $call = shift;
559         my $ref;
560         my $clref;
561         my $dxchan;
562         my @nodelist = DXProt::get_all_ak1a();
563         
564         # bat down the message list looking for one that needs to go off site and whose
565         # nearest node is not busy.
566
567         dbg('msg', "queue msg ($sort)\n");
568         foreach $ref (@msg) {
569                 # firstly, is it private and unread? if so can I find the recipient
570                 # in my cluster node list offsite?
571
572                 # ignore 'delayed' messages until their waiting time has expired
573                 if (exists $ref->{waitt}) {
574                         next if $ref->{waitt} < $main::systime;
575                         delete $ref->{waitt};
576                 } 
577                 
578                 if ($ref->{private}) {
579                         if ($ref->{'read'} == 0) {
580                                 $clref = DXCluster->get_exact($ref->{to});
581                                 unless ($clref) {             # otherwise look for a homenode
582                                         my $uref = DXUser->get($ref->{to});
583                                         my $hnode =  $uref->homenode if $uref;
584                                         $clref = DXCluster->get_exact($hnode) if $hnode;
585                                 }
586                                 if ($clref && !grep { $clref->{dxchan} == $_ } DXCommandmode::get_all) {
587                                         $dxchan = $clref->{dxchan};
588                                         $ref->start_msg($dxchan) if $dxchan && $clref && !get_busy($dxchan->call) && $dxchan->state eq 'normal';
589                                 }
590                         }
591                 } elsif (!$sort) {
592                         # otherwise we are dealing with a bulletin, compare the gotit list with
593                         # the nodelist up above, if there are sites that haven't got it yet
594                         # then start sending it - what happens when we get loops is anyone's
595                         # guess, use (to, from, time, subject) tuple?
596                         my $noderef;
597                         foreach $noderef (@nodelist) {
598                                 next if $noderef->call eq $main::mycall;
599                                 next if grep { $_ eq $noderef->call } @{$ref->{gotit}};
600                                 next unless $ref->forward_it($noderef->call);           # check the forwarding file
601                                 # next if $noderef->isolate;               # maybe add code for stuff originated here?
602                                 # next if DXUser->get( ${$ref->{gotit}}[0] )->isolate;  # is the origin isolated?
603                                 
604                                 # if we are here we have a node that doesn't have this message
605                                 $ref->start_msg($noderef) if !get_busy($noderef->call)  && $noderef->state eq 'normal';
606                                 last;
607                         }
608                 }
609                 
610                 # if all the available nodes are busy then stop
611                 last if @nodelist == scalar grep { get_busy($_->call) } @nodelist;
612         }
613 }
614
615 # is there a message for me?
616 sub for_me
617 {
618         my $call = uc shift;
619         my $ref;
620         
621         foreach $ref (@msg) {
622                 # is it for me, private and unread? 
623                 if ($ref->{to} eq $call && $ref->{private}) {
624                         return 1 if !$ref->{'read'};
625                 }
626         }
627         return 0;
628 }
629
630 # start the message off on its travels with a PC28
631 sub start_msg
632 {
633         my ($self, $dxchan) = @_;
634         
635         dbg('msg', "start msg $self->{msgno}\n");
636         $self->{linesreq} = 5;
637         $self->{count} = 0;
638         $self->{tonode} = $dxchan->call;
639         $self->{fromnode} = $main::mycall;
640         $busy{$self->{tonode}} = $self;
641         $work{$self->{tonode}} = $self;
642         $dxchan->send(DXProt::pc28($self->{tonode}, $self->{fromnode}, $self->{to}, $self->{from}, $self->{t}, $self->{private}, $self->{subject}, $self->{origin}, $self->{rrreq}));
643 }
644
645 # get the ref of a busy node
646 sub get_busy
647 {
648         my $call = shift;
649         return $busy{$call};
650 }
651
652 # get the busy queue
653 sub get_all_busy
654 {
655         return values %busy;
656 }
657
658 # get the forwarding queue
659 sub get_fwq
660 {
661         return values %work;
662 }
663
664 # stop a message from continuing, clean it out, unlock interlocks etc
665 sub stop_msg
666 {
667         my $self = shift;
668         my $node = shift;
669         my $stream = $self->{stream} if exists $self->{stream};
670         
671         
672         dbg('msg', "stop msg $self->{msgno} -> node $node\n");
673         delete $work{$node};
674         delete $work{"$node$stream"} if $stream;
675         $self->workclean;
676         delete $busy{$node};
677 }
678
679 # get a new transaction number from the file specified
680 sub next_transno
681 {
682         my $name = shift;
683         $name =~ s/\W//og;                      # remove non-word characters
684         my $fn = "$msgdir/$name";
685         my $msgno;
686         
687         my $fh = new IO::File;
688         if (sysopen($fh, $fn, O_RDWR|O_CREAT, 0666)) {
689                 $fh->autoflush(1);
690                 $msgno = $fh->getline;
691                 chomp $msgno;
692                 $msgno++;
693                 seek $fh, 0, 0;
694                 $fh->print("$msgno\n");
695                 dbg('msg', "msgno $msgno allocated for $name\n");
696                 $fh->close;
697         } else {
698                 confess "can't open $fn $!";
699         }
700         return $msgno;
701 }
702
703 # initialise the message 'system', read in all the message headers
704 sub init
705 {
706         my $dir = new IO::File;
707         my @dir;
708         my $ref;
709
710         # load various control files
711         my @in = load_badmsg();
712         print "@in\n" if @in;
713         @in = load_forward();
714         print "@in\n" if @in;
715
716         # read in the directory
717         opendir($dir, $msgdir) or confess "can't open $msgdir $!";
718         @dir = readdir($dir);
719         closedir($dir);
720
721         @msg = ();
722         for (sort @dir) {
723                 next unless /^m\d+$/o;
724                 
725                 $ref = read_msg_header("$msgdir/$_");
726                 next unless $ref;
727                 
728                 # delete any messages to 'badmsg.pl' places
729                 if (grep $ref->{to} eq $_, @badmsg) {
730                         dbg('msg', "'Bad' TO address $ref->{to}");
731                         Log('msg', "'Bad' TO address $ref->{to}");
732                         $ref->del_msg;
733                         next;
734                 }
735
736                 # add the message to the available queue
737                 add_dir($ref); 
738         }
739 }
740
741 # add the message to the directory listing
742 sub add_dir
743 {
744         my $ref = shift;
745         confess "tried to add a non-ref to the msg directory" if !ref $ref;
746         push @msg, $ref;
747 }
748
749 # return all the current messages
750 sub get_all
751 {
752         return @msg;
753 }
754
755 # get a particular message
756 sub get
757 {
758         my $msgno = shift;
759         for (@msg) {
760                 return $_ if $_->{msgno} == $msgno;
761                 last if $_->{msgno} > $msgno;
762         }
763         return undef;
764 }
765
766 # return the official filename for a message no
767 sub filename
768 {
769         return sprintf "$msgdir/m%06d", shift;
770 }
771
772 #
773 # return a list of valid elements 
774
775
776 sub fields
777 {
778         return keys(%valid);
779 }
780
781 #
782 # return a prompt for a field
783 #
784
785 sub field_prompt
786
787         my ($self, $ele) = @_;
788         return $valid{$ele};
789 }
790
791 #
792 # send a message state machine
793 sub do_send_stuff
794 {
795         my $self = shift;
796         my $line = shift;
797         my @out;
798         
799         if ($self->state eq 'send1') {
800                 #  $DB::single = 1;
801                 confess "local var gone missing" if !ref $self->{loc};
802                 my $loc = $self->{loc};
803                 $loc->{subject} = $line;
804                 $loc->{lines} = [];
805                 $self->state('sendbody');
806                 #push @out, $self->msg('sendbody');
807                 push @out, $self->msg('m8');
808         } elsif ($self->state eq 'sendbody') {
809                 confess "local var gone missing" if !ref $self->{loc};
810                 my $loc = $self->{loc};
811                 if ($line eq "\032" || uc $line eq "/EX") {
812                         my $to;
813                         
814                         if (@{$loc->{lines}} > 0) {
815                                 foreach $to (@{$loc->{to}}) {
816                                         my $ref;
817                                         my $systime = $main::systime;
818                                         my $mycall = $main::mycall;
819                                         $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'),
820                                                                                 uc $to,
821                                                                                 $self->call, 
822                                                                                 $systime,
823                                                                                 $loc->{private}, 
824                                                                                 $loc->{subject}, 
825                                                                                 $mycall,
826                                                                                 '0',
827                                                                                 $loc->{rrreq});
828                                         $ref->store($loc->{lines});
829                                         $ref->add_dir();
830                                         push @out, $self->msg('m11', $ref->{msgno}, $to);
831                                         #push @out, "msgno $ref->{msgno} sent to $to";
832                                         my $dxchan = DXChannel->get(uc $to);
833                                         if ($dxchan) {
834                                                 if ($dxchan->is_user()) {
835                                                         $dxchan->send($dxchan->msg('m9'));
836                                                 }
837                                         }
838                                 }
839                         }
840                         delete $loc->{lines};
841                         delete $loc->{to};
842                         delete $self->{loc};
843                         $self->func(undef);
844                         
845                         $self->state('prompt');
846                 } elsif ($line eq "\031" || uc $line eq "/ABORT" || uc $line eq "/QUIT") {
847                         #push @out, $self->msg('sendabort');
848                         push @out, $self->msg('m10');
849                         delete $loc->{lines};
850                         delete $loc->{to};
851                         delete $self->{loc};
852                         $self->func(undef);
853                         $self->state('prompt');
854                 } else {
855                         
856                         # i.e. it ain't and end or abort, therefore store the line
857                         push @{$loc->{lines}}, length($line) > 0 ? $line : " ";
858                 }
859         }
860         return (1, @out);
861 }
862
863 # return the standard directory line for this ref 
864 sub dir
865 {
866         my $ref = shift;
867         return sprintf "%6d%s%s%5d %8.8s %8.8s %-6.6s %5.5s %-30.30s", 
868                 $ref->msgno, $ref->read ? '-' : ' ', $ref->private ? 'p' : ' ', $ref->size,
869                         $ref->to, $ref->from, cldate($ref->t), ztime($ref->t), $ref->subject;
870 }
871
872 # load the forward table
873 sub load_forward
874 {
875         my @out;
876         do "$forwardfn" if -e "$forwardfn";
877         push @out, $@ if $@;
878         return @out;
879 }
880
881 # load the bad message table
882 sub load_badmsg
883 {
884         my @out;
885         do "$badmsgfn" if -e "$badmsgfn";
886         push @out, $@ if $@;
887         return @out;
888 }
889
890 #
891 # forward that message or not according to the forwarding table
892 # returns 1 for forward, 0 - to ignore
893 #
894
895 sub forward_it
896 {
897         my $ref = shift;
898         my $call = shift;
899         my $i;
900         
901         for ($i = 0; $i < @forward; $i += 5) {
902                 my ($sort, $field, $pattern, $action, $bbs) = @forward[$i..($i+4)]; 
903                 my $tested;
904                 
905                 # are we interested?
906                 last if $ref->{private} && $sort ne 'P';
907                 last if !$ref->{private} && $sort ne 'B';
908                 
909                 # select field
910                 $tested = $ref->{to} if $field eq 'T';
911                 $tested = $ref->{from} if $field eq 'F';
912                 $tested = $ref->{origin} if $field eq 'O';
913                 $tested = $ref->{subject} if $field eq 'S';
914
915                 if (!$pattern || $tested =~ m{$pattern}i) {
916                         return 0 if $action eq 'I';
917                         return 1 if !$bbs || grep $_ eq $call, @{$bbs};
918                 }
919         }
920         return 0;
921 }
922
923 no strict;
924 sub AUTOLOAD
925 {
926         my $self = shift;
927         my $name = $AUTOLOAD;
928         return if $name =~ /::DESTROY$/;
929         $name =~ s/.*:://o;
930         
931         confess "Non-existant field '$AUTOLOAD'" if !$valid{$name};
932         @_ ? $self->{$name} = shift : $self->{$name} ;
933 }
934
935 1;
936
937 __END__