]> scm.dxcluster.org Git - spider.git/blob - perl/DXMsg.pm
standardise build number calc
[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 use DXUtil;
19 use DXChannel;
20 use DXUser;
21 use DXM;
22 use DXCluster;
23 use DXProtVars;
24 use DXProtout;
25 use DXDebug;
26 use DXLog;
27 use IO::File;
28 use Fcntl;
29
30 use strict;
31 use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean
32                         @badmsg @swop $swopfn $badmsgfn $forwardfn @forward $timeout $waittime
33                     $queueinterval $lastq $importfn $minchunk $maxchunk);
34
35 %work = ();                                             # outstanding jobs
36 @msg = ();                                              # messages we have
37 %busy = ();                                             # station interlocks
38 $msgdir = "$main::root/msg";    # directory contain the msgs
39 $maxage = 30 * 86400;                   # the maximum age that a message shall live for if not marked 
40 $last_clean = 0;                                # last time we did a clean
41 @forward = ();                  # msg forward table
42 @badmsg = ();                                   # bad message table
43 @swop = ();                                             # swop table
44 $timeout = 30*60;               # forwarding timeout
45 $waittime = 30*60;              # time an aborted outgoing message waits before trying again
46 $queueinterval = 1*60;          # run the queue every 1 minute
47 $lastq = 0;
48
49 $minchunk = 4800;               # minimum chunk size for a split message
50 $maxchunk = 6000;               # maximum chunk size
51
52 $badmsgfn = "$msgdir/badmsg.pl";    # list of TO address we wont store
53 $forwardfn = "$msgdir/forward.pl";  # the forwarding table
54 $swopfn = "$msgdir/swop.pl";        # the swopping table
55 $importfn = "$msgdir/import";       # import directory
56
57
58 %valid = (
59                   fromnode => '5,From Node',
60                   tonode => '5,To Node',
61                   to => '0,To',
62                   from => '0,From',
63                   t => '0,Msg Time,cldatetime',
64                   private => '5,Private',
65                   subject => '0,Subject',
66                   linesreq => '0,Lines per Gob',
67                   rrreq => '5,Read Confirm',
68                   origin => '0,Origin',
69                   lines => '5,Data',
70                   stream => '9,Stream No',
71                   count => '5,Gob Linecnt',
72                   file => '5,File?,yesno',
73                   gotit => '5,Got it Nodes,parray',
74                   lines => '5,Lines,parray',
75                   'read' => '5,Times read',
76                   size => '0,Size',
77                   msgno => '0,Msgno',
78                   keep => '0,Keep this?,yesno',
79                   lastt => '5,Last processed,cldatetime',
80                   waitt => '5,Wait until,cldatetime',
81                  );
82
83 # allocate a new object
84 # called fromnode, tonode, from, to, datetime, private?, subject, nolinesper  
85 sub alloc                  
86 {
87         my $pkg = shift;
88         my $self = bless {}, $pkg;
89         $self->{msgno} = shift;
90         my $to = shift;
91         #  $to =~ s/-\d+$//o;
92         $self->{to} = ($to eq $main::mycall) ? $main::myalias : $to;
93         my $from = shift;
94         $from =~ s/-\d+$//o;
95         $self->{from} = uc $from;
96         $self->{t} = shift;
97         $self->{private} = shift;
98         $self->{subject} = shift;
99         $self->{origin} = shift;
100         $self->{'read'} = shift;
101         $self->{rrreq} = shift;
102         $self->{gotit} = [];
103         $self->{lastt} = $main::systime;
104         $self->{lines} = [];
105     
106         return $self;
107 }
108
109 sub workclean
110 {
111         my $ref = shift;
112         delete $ref->{lines};
113         delete $ref->{linesreq};
114         delete $ref->{tonode};
115         delete $ref->{fromnode};
116         delete $ref->{stream};
117         delete $ref->{file};
118         delete $ref->{count};
119         delete $ref->{lastt} if exists $ref->{lastt};
120         delete $ref->{waitt} if exists $ref->{waitt};
121 }
122
123 sub process
124 {
125         my ($self, $line) = @_;
126
127         # this is periodic processing
128         if (!$self || !$line) {
129
130                 if ($main::systime >= $lastq + $queueinterval) {
131
132                         # wander down the work queue stopping any messages that have timed out
133                         for (keys %busy) {
134                                 my $node = $_;
135                                 my $ref = $busy{$_};
136                                 if (exists $ref->{lastt} && $main::systime >= $ref->{lastt} + $timeout) {
137                                         dbg('msg', "Timeout, stopping msgno: $ref->{msgno} -> $node");
138                                         Log('msg', "Timeout, stopping msgno: $ref->{msgno} -> $node");
139                                         $ref->stop_msg($node);
140                                         
141                                         # delay any outgoing messages that fail
142                                         $ref->{waitt} = $main::systime + $waittime + rand(120) if $node ne $main::mycall;
143                                 }
144                         }
145
146                         # queue some message if the interval timer has gone off
147                         queue_msg(0);
148
149                         # import any messages in the import directory
150                         import_msgs();
151                         
152                         $lastq = $main::systime;
153                 }
154
155                 # clean the message queue
156                 clean_old() if $main::systime - $last_clean > 3600 ;
157                 return;
158         }
159
160         my @f = split /\^/, $line;
161         my ($pcno) = $f[0] =~ /^PC(\d\d)/; # just get the number
162
163  SWITCH: {
164                 if ($pcno == 28) {              # incoming message
165
166                         # first look for any messages in the busy queue 
167                         # and cancel them this should both resolve timed out incoming messages
168                         # and crossing of message between nodes, incoming messages have priority
169                         if (exists $busy{$f[2]}) {
170                                 my $ref = $busy{$f[2]};
171                                 my $tonode = $ref->{tonode};
172                                 dbg('msg', "Busy, stopping msgno: $ref->{msgno} -> $f[2]");
173                                 $ref->stop_msg($self->call);
174                         }
175
176                         my $t = cltounix($f[5], $f[6]);
177                         my $stream = next_transno($f[2]);
178                         $f[13] = $self->call unless $f[13] && $f[13] gt ' ';
179                         my $ref = DXMsg->alloc($stream, uc $f[3], $f[4], $t, $f[7], $f[8], $f[13], '0', $f[11]);
180                         
181                         # fill in various forwarding state variables
182                         $ref->{fromnode} = $f[2];
183                         $ref->{tonode} = $f[1];
184                         $ref->{rrreq} = $f[11];
185                         $ref->{linesreq} = $f[10];
186                         $ref->{stream} = $stream;
187                         $ref->{count} = 0;      # no of lines between PC31s
188                         dbg('msg', "new message from $f[4] to $f[3] '$f[8]' stream $stream\n");
189                         Log('msg', "Incoming message $f[4] to $f[3] '$f[8]'" );
190                         $work{"$f[2]$stream"} = $ref; # store in work
191                         $busy{$f[2]} = $ref; # set interlock
192                         $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack
193                         $ref->{lastt} = $main::systime;
194
195                         # look to see whether this is a non private message sent to a known callsign
196                         my $uref = DXUser->get_current($ref->{to});
197                         if (is_callsign($ref->{to}) && !$ref->{private} && $uref && $uref->homenode) {
198                                 $ref->{private} = 1;
199                                 dbg('msg', "set bull to $ref->{to} to private");
200                         }
201                         last SWITCH;
202                 }
203                 
204                 if ($pcno == 29) {              # incoming text
205                         my $ref = $work{"$f[2]$f[3]"};
206                         if ($ref) {
207                                 $f[4] =~ s/\%5E/^/g;
208                                 push @{$ref->{lines}}, $f[4];
209                                 $ref->{count}++;
210                                 if ($ref->{count} >= $ref->{linesreq}) {
211                                         $self->send(DXProt::pc31($f[2], $f[1], $f[3]));
212                                         dbg('msg', "stream $f[3]: $ref->{count} lines received\n");
213                                         $ref->{count} = 0;
214                                 }
215                                 $ref->{lastt} = $main::systime;
216                         } else {
217                                 dbg('msg', "PC29 from unknown stream $f[3] from $f[2]" );
218                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
219                         }
220                         last SWITCH;
221                 }
222                 
223                 if ($pcno == 30) {              # this is a incoming subject ack
224                         my $ref = $work{$f[2]}; # note no stream at this stage
225                         if ($ref) {
226                                 delete $work{$f[2]};
227                                 $ref->{stream} = $f[3];
228                                 $ref->{count} = 0;
229                                 $ref->{linesreq} = 5;
230                                 $work{"$f[2]$f[3]"} = $ref;     # new ref
231                                 dbg('msg', "incoming subject ack stream $f[3]\n");
232                                 $busy{$f[2]} = $ref; # interlock
233                                 push @{$ref->{lines}}, ($ref->read_msg_body);
234                                 $ref->send_tranche($self);
235                                 $ref->{lastt} = $main::systime;
236                         } else {
237                                 dbg('msg', "PC30 from unknown stream $f[3] from $f[2]" );
238                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
239                         } 
240                         last SWITCH;
241                 }
242                 
243                 if ($pcno == 31) {              # acknowledge a tranche of lines
244                         my $ref = $work{"$f[2]$f[3]"};
245                         if ($ref) {
246                                 dbg('msg', "tranche ack stream $f[3]\n");
247                                 $ref->send_tranche($self);
248                                 $ref->{lastt} = $main::systime;
249                         } else {
250                                 dbg('msg', "PC31 from unknown stream $f[3] from $f[2]" );
251                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
252                         } 
253                         last SWITCH;
254                 }
255                 
256                 if ($pcno == 32) {              # incoming EOM
257                         dbg('msg', "stream $f[3]: EOM received\n");
258                         my $ref = $work{"$f[2]$f[3]"};
259                         if ($ref) {
260                                 $self->send(DXProt::pc33($f[2], $f[1], $f[3])); # acknowledge it
261                                 
262                                 # get the next msg no - note that this has NOTHING to do with the stream number in PC protocol
263                                 # store the file or message
264                                 # remove extraneous rubbish from the hash
265                                 # remove it from the work in progress vector
266                                 # stuff it on the msg queue
267                                 if ($ref->{lines}) {
268                                         if ($ref->{file}) {
269                                                 $ref->store($ref->{lines});
270                                         } else {
271
272                                                 # does an identical message already exist?
273                                                 my $m;
274                                                 for $m (@msg) {
275                                                         if ($ref->{subject} eq $m->{subject} && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from} && $ref->{to} eq $m->{to}) {
276                                                                 $ref->stop_msg($self->call);
277                                                                 my $msgno = $m->{msgno};
278                                                                 dbg('msg', "duplicate message from $ref->{from} -> $ref->{to} to $msgno");
279                                                                 Log('msg', "duplicate message from $ref->{from} -> $ref->{to} to $msgno");
280                                                                 return;
281                                                         }
282                                                 }
283
284                                                 # swop addresses
285                                                 $ref->swop_it($self->call);
286                                                 
287                                                 # look for 'bad' to addresses 
288                                                 if ($ref->dump_it) {
289                                                         $ref->stop_msg($self->call);
290                                                         dbg('msg', "'Bad' message $ref->{to}");
291                                                         Log('msg', "'Bad' message $ref->{to}");
292                                                         return;
293                                                 }
294
295                                                 $ref->{msgno} = next_transno("Msgno");
296                                                 push @{$ref->{gotit}}, $f[2]; # mark this up as being received
297                                                 $ref->store($ref->{lines});
298                                                 add_dir($ref);
299                                                 my $dxchan = DXChannel->get($ref->{to});
300                                                 $dxchan->send($dxchan->msg('m9')) if $dxchan && $dxchan->is_user;
301                                                 Log('msg', "Message $ref->{msgno} from $ref->{from} received from $f[2] for $ref->{to}");
302                                         }
303                                 }
304                                 $ref->stop_msg($self->call);
305                         } else {
306                                 dbg('msg', "PC32 from unknown stream $f[3] from $f[2]" );
307                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
308                         }
309                         # queue_msg(0);
310                         last SWITCH;
311                 }
312                 
313                 if ($pcno == 33) {              # acknowledge the end of message
314                         my $ref = $work{"$f[2]$f[3]"};
315                         if ($ref) {
316                                 if ($ref->{private}) { # remove it if it private and gone off site#
317                                         Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2] and deleted");
318                                         $ref->del_msg;
319                                 } else {
320                                         Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2]");
321                                         push @{$ref->{gotit}}, $f[2]; # mark this up as being received
322                                         $ref->store($ref->{lines});     # re- store the file
323                                 }
324                                 $ref->stop_msg($self->call);
325                         } else {
326                                 dbg('msg', "PC33 from unknown stream $f[3] from $f[2]" );
327                                 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
328                         } 
329
330                         # send next one if present
331                         queue_msg(0);
332                         last SWITCH;
333                 }
334                 
335                 if ($pcno == 40) {              # this is a file request
336                         $f[3] =~ s/\\/\//og; # change the slashes
337                         $f[3] =~ s/\.//og;      # remove dots
338                         $f[3] =~ s/^\///o;   # remove the leading /
339                         $f[3] = lc $f[3];       # to lower case;
340                         dbg('msg', "incoming file $f[3]\n");
341                         $f[3] = 'packclus/' . $f[3] unless $f[3] =~ /^packclus\//o;
342                         
343                         # create any directories
344                         my @part = split /\//, $f[3];
345                         my $part;
346                         my $fn = "$main::root";
347                         pop @part;                      # remove last part
348                         foreach $part (@part) {
349                                 $fn .= "/$part";
350                                 next if -e $fn;
351                                 last SWITCH if !mkdir $fn, 0777;
352                                 dbg('msg', "created directory $fn\n");
353                         }
354                         my $stream = next_transno($f[2]);
355                         my $ref = DXMsg->alloc($stream, "$main::root/$f[3]", $self->call, time, !$f[4], $f[3], ' ', '0', '0');
356                         
357                         # forwarding variables
358                         $ref->{fromnode} = $f[1];
359                         $ref->{tonode} = $f[2];
360                         $ref->{linesreq} = $f[5];
361                         $ref->{stream} = $stream;
362                         $ref->{count} = 0;      # no of lines between PC31s
363                         $ref->{file} = 1;
364                         $ref->{lastt} = $main::systime;
365                         $work{"$f[2]$stream"} = $ref; # store in work
366                         $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack 
367                         
368                         last SWITCH;
369                 }
370                 
371                 if ($pcno == 42) {              # abort transfer
372                         dbg('msg', "stream $f[3]: abort received\n");
373                         my $ref = $work{"$f[2]$f[3]"};
374                         if ($ref) {
375                                 $ref->stop_msg($self->call);
376                                 $ref = undef;
377                         }
378                         last SWITCH;
379                 }
380
381                 if ($pcno == 49) {      # global delete on subject
382                         for (@msg) {
383                                 if ($_->{from} eq $f[1] && $_->{subject} eq $f[2]) {
384                                         $_->del_msg();
385                                         Log('msg', "Message $_->{msgno} from $_->{from} ($_->{subject}) fully deleted");
386                                         DXProt::broadcast_ak1a($line, $self);
387                                 }
388                         }
389                 }
390         }
391 }
392
393
394 # store a message away on disc or whatever
395 #
396 # NOTE the second arg is a REFERENCE not a list
397 sub store
398 {
399         my $ref = shift;
400         my $lines = shift;
401
402         if ($ref->{file}) {                     # a file
403                 dbg('msg', "To be stored in $ref->{to}\n");
404                 
405                 my $fh = new IO::File "$ref->{to}", "w";
406                 if (defined $fh) {
407                         my $line;
408                         foreach $line (@{$lines}) {
409                                 print $fh "$line\n";
410                         }
411                         $fh->close;
412                         dbg('msg', "file $ref->{to} stored\n");
413                         Log('msg', "file $ref->{to} from $ref->{from} stored" );
414                 } else {
415                         confess "can't open file $ref->{to} $!";  
416                 }
417         } else {                                        # a normal message
418
419                 # attempt to open the message file
420                 my $fn = filename($ref->{msgno});
421                 
422                 dbg('msg', "To be stored in $fn\n");
423                 
424                 # now save the file, overwriting what's there, YES I KNOW OK! (I will change it if it's a problem)
425                 my $fh = new IO::File "$fn", "w";
426                 if (defined $fh) {
427                         my $rr = $ref->{rrreq} ? '1' : '0';
428                         my $priv = $ref->{private} ? '1': '0';
429                         print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{'read'}^$rr\n";
430                         print $fh "=== ", join('^', @{$ref->{gotit}}), "\n";
431                         my $line;
432                         $ref->{size} = 0;
433                         foreach $line (@{$lines}) {
434                                 $ref->{size} += (length $line) + 1;
435                                 print $fh "$line\n";
436                         }
437                         $fh->close;
438                         dbg('msg', "msg $ref->{msgno} stored\n");
439                         Log('msg', "msg $ref->{msgno} from $ref->{from} to $ref->{to} stored" );
440                 } else {
441                         confess "can't open msg file $fn $!";  
442                 }
443         }
444 }
445
446 # delete a message
447 sub del_msg
448 {
449         my $self = shift;
450         
451         # remove it from the active message list
452         dbg('msg', "\@msg = " . scalar @msg . " before delete");
453         @msg = grep { $_ != $self } @msg;
454         
455         # remove the file
456         unlink filename($self->{msgno});
457         dbg('msg', "deleting $self->{msgno}\n");
458         dbg('msg', "\@msg = " . scalar @msg . " after delete");
459 }
460
461 # clean out old messages from the message queue
462 sub clean_old
463 {
464         my $ref;
465         
466         # mark old messages for deletion
467         dbg('msg', "\@msg = " . scalar @msg . " before delete");
468         foreach $ref (@msg) {
469                 if (ref($ref) && !$ref->{keep} && $ref->{t} < $main::systime - $maxage) {
470                         $ref->{deleteme} = 1;
471                         unlink filename($ref->{msgno});
472                         dbg('msg', "deleting old $ref->{msgno}\n");
473                 }
474         }
475         
476         # remove them all from the active message list
477         @msg = grep { !$_->{deleteme} } @msg;
478         dbg('msg', "\@msg = " . scalar @msg . " after delete");
479         $last_clean = $main::systime;
480 }
481
482 # read in a message header
483 sub read_msg_header
484
485         my $fn = shift;
486         my $file;
487         my $line;
488         my $ref;
489         my @f;
490         my $size;
491         
492         $file = new IO::File "$fn";
493         if (!$file) {
494             dbg('err', "Error reading $fn $!");
495             Log('err', "Error reading $fn $!");
496                 return undef;
497         }
498         $size = -s $fn;
499         $line = <$file>;                        # first line
500         if ($size == 0 || !$line) {
501             dbg('err', "Empty $fn $!");
502             Log('err', "Empty $fn $!");
503                 return undef;
504         }
505         chomp $line;
506         $size -= length $line;
507         if (! $line =~ /^===/o) {
508                 dbg('err', "corrupt first line in $fn ($line)");
509                 Log('err', "corrupt first line in $fn ($line)");
510                 return undef;
511         }
512         $line =~ s/^=== //o;
513         @f = split /\^/, $line;
514         $ref = DXMsg->alloc(@f);
515         
516         $line = <$file>;                        # second line
517         chomp $line;
518         $size -= length $line;
519         if (! $line =~ /^===/o) {
520             dbg('err', "corrupt second line in $fn ($line)");
521             Log('err', "corrupt second line in $fn ($line)");
522                 return undef;
523         }
524         $line =~ s/^=== //o;
525         $ref->{gotit} = [];
526         @f = split /\^/, $line;
527         push @{$ref->{gotit}}, @f;
528         $ref->{size} = $size;
529         
530         close($file);
531         
532         return $ref;
533 }
534
535 # read in a message header
536 sub read_msg_body
537 {
538         my $self = shift;
539         my $msgno = $self->{msgno};
540         my $file;
541         my $line;
542         my $fn = filename($msgno);
543         my @out;
544         
545         $file = new IO::File;
546         if (!open($file, $fn)) {
547                 dbg('err' ,"Error reading $fn $!");
548                 Log('err' ,"Error reading $fn $!");
549                 return undef;
550         }
551         @out = map {chomp; $_} <$file>;
552         close($file);
553         
554         shift @out if $out[0] =~ /^=== /;
555         shift @out if $out[0] =~ /^=== /;
556         return @out;
557 }
558
559 # send a tranche of lines to the other end
560 sub send_tranche
561 {
562         my ($self, $dxchan) = @_;
563         my @out;
564         my $to = $self->{tonode};
565         my $from = $self->{fromnode};
566         my $stream = $self->{stream};
567         my $lines = $self->{lines};
568         my ($c, $i);
569         
570         for ($i = 0, $c = $self->{count}; $i < $self->{linesreq} && $c < @$lines; $i++, $c++) {
571                 push @out, DXProt::pc29($to, $from, $stream, $lines->[$c]);
572     }
573     $self->{count} = $c;
574
575     push @out, DXProt::pc32($to, $from, $stream) if $i < $self->{linesreq};
576         $dxchan->send(@out);
577 }
578
579         
580 # find a message to send out and start the ball rolling
581 sub queue_msg
582 {
583         my $sort = shift;
584         my $call = shift;
585         my $ref;
586         my $clref;
587         
588         # bat down the message list looking for one that needs to go off site and whose
589         # nearest node is not busy.
590
591         dbg('msg', "queue msg ($sort)\n");
592         my @nodelist = DXChannel::get_all_nodes;
593         foreach $ref (@msg) {
594                 # firstly, is it private and unread? if so can I find the recipient
595                 # in my cluster node list offsite?
596
597                 # ignore 'delayed' messages until their waiting time has expired
598                 if (exists $ref->{waitt}) {
599                         next if $ref->{waitt} > $main::systime;
600                         delete $ref->{waitt};
601                 } 
602
603                 # deal with routed private messages
604                 my $dxchan;
605                 if ($ref->{private}) {
606                         next if $ref->{'read'};           # if it is read, it is stuck here
607                         $clref = DXCluster->get_exact($ref->{to});
608                         unless ($clref) {             # otherwise look for a homenode
609                                 my $uref = DXUser->get_current($ref->{to});
610                                 my $hnode =  $uref->homenode if $uref;
611                                 $clref = DXCluster->get_exact($hnode) if $hnode;
612                         }
613                         if ($clref && !grep { $clref->dxchan == $_ } DXCommandmode::get_all()) {
614                                 next if $clref->call eq $main::mycall;  # i.e. it lives here
615                                 $dxchan = $clref->dxchan;
616                                 $ref->start_msg($dxchan) if $dxchan && !get_busy($dxchan->call)  && $dxchan->state eq 'normal';
617                         }
618                 }
619                 
620                 # otherwise we are dealing with a bulletin or forwarded private message
621                 # compare the gotit list with
622                 # the nodelist up above, if there are sites that haven't got it yet
623                 # then start sending it - what happens when we get loops is anyone's
624                 # guess, use (to, from, time, subject) tuple?
625                 foreach $dxchan (@nodelist) {
626                         my $call = $dxchan->call;
627                         next unless $call;
628                         next if $call eq $main::mycall;
629                         next if ref $ref->{gotit} && grep $_ eq $call, @{$ref->{gotit}};
630                         next unless $ref->forward_it($call);           # check the forwarding file
631
632                         # if we are here we have a node that doesn't have this message
633                         $ref->start_msg($dxchan) if !get_busy($call)  && $dxchan->state eq 'normal';
634                         last;
635                 }
636
637                 # if all the available nodes are busy then stop
638                 last if @nodelist == scalar grep { get_busy($_->call) } @nodelist;
639         }
640 }
641
642 # is there a message for me?
643 sub for_me
644 {
645         my $call = uc shift;
646         my $ref;
647         
648         foreach $ref (@msg) {
649                 # is it for me, private and unread? 
650                 if ($ref->{to} eq $call && $ref->{private}) {
651                         return 1 if !$ref->{'read'};
652                 }
653         }
654         return 0;
655 }
656
657 # start the message off on its travels with a PC28
658 sub start_msg
659 {
660         my ($self, $dxchan) = @_;
661         
662         dbg('msg', "start msg $self->{msgno}\n");
663         $self->{linesreq} = 10;
664         $self->{count} = 0;
665         $self->{tonode} = $dxchan->call;
666         $self->{fromnode} = $main::mycall;
667         $busy{$self->{tonode}} = $self;
668         $work{$self->{tonode}} = $self;
669         $self->{lastt} = $main::systime;
670         $dxchan->send(DXProt::pc28($self->{tonode}, $self->{fromnode}, $self->{to}, $self->{from}, $self->{t}, $self->{private}, $self->{subject}, $self->{origin}, $self->{rrreq}));
671 }
672
673 # get the ref of a busy node
674 sub get_busy
675 {
676         my $call = shift;
677         return $busy{$call};
678 }
679
680 # get the busy queue
681 sub get_all_busy
682 {
683         return values %busy;
684 }
685
686 # get the forwarding queue
687 sub get_fwq
688 {
689         return values %work;
690 }
691
692 # stop a message from continuing, clean it out, unlock interlocks etc
693 sub stop_msg
694 {
695         my $self = shift;
696         my $node = shift;
697         my $stream = $self->{stream} if exists $self->{stream};
698         
699         
700         dbg('msg', "stop msg $self->{msgno} -> node $node\n");
701         delete $work{$node};
702         delete $work{"$node$stream"} if $stream;
703         $self->workclean;
704         delete $busy{$node};
705 }
706
707 # get a new transaction number from the file specified
708 sub next_transno
709 {
710         my $name = shift;
711         $name =~ s/\W//og;                      # remove non-word characters
712         my $fn = "$msgdir/$name";
713         my $msgno;
714         
715         my $fh = new IO::File;
716         if (sysopen($fh, $fn, O_RDWR|O_CREAT, 0666)) {
717                 $fh->autoflush(1);
718                 $msgno = $fh->getline;
719                 chomp $msgno;
720                 $msgno++;
721                 seek $fh, 0, 0;
722                 $fh->print("$msgno\n");
723                 dbg('msg', "msgno $msgno allocated for $name\n");
724                 $fh->close;
725         } else {
726                 confess "can't open $fn $!";
727         }
728         return $msgno;
729 }
730
731 # initialise the message 'system', read in all the message headers
732 sub init
733 {
734         my $dir = new IO::File;
735         my @dir;
736         my $ref;
737                 
738         # load various control files
739         dbg('err', "load badmsg: " . (load_badmsg() or "Ok"));
740         dbg('err', "load forward: " . (load_forward() or "Ok"));
741         dbg('err', "load swop: " . (load_swop() or "Ok"));
742
743         # read in the directory
744         opendir($dir, $msgdir) or confess "can't open $msgdir $!";
745         @dir = readdir($dir);
746         closedir($dir);
747
748         @msg = ();
749         for (sort @dir) {
750                 next unless /^m\d\d\d\d\d\d$/;
751                 
752                 $ref = read_msg_header("$msgdir/$_");
753                 unless ($ref) {
754                         dbg('err', "Deleting $_");
755                         Log('err', "Deleting $_");
756                         unlink "$msgdir/$_";
757                         next;
758                 }
759                 
760                 # delete any messages to 'badmsg.pl' places
761                 if ($ref->dump_it) {
762                         dbg('msg', "'Bad' TO address $ref->{to}");
763                         Log('msg', "'Bad' TO address $ref->{to}");
764                         $ref->del_msg;
765                         next;
766                 }
767
768                 # add the message to the available queue
769                 add_dir($ref); 
770         }
771 }
772
773 # add the message to the directory listing
774 sub add_dir
775 {
776         my $ref = shift;
777         confess "tried to add a non-ref to the msg directory" if !ref $ref;
778         push @msg, $ref;
779 }
780
781 # return all the current messages
782 sub get_all
783 {
784         return @msg;
785 }
786
787 # get a particular message
788 sub get
789 {
790         my $msgno = shift;
791         for (@msg) {
792                 return $_ if $_->{msgno} == $msgno;
793                 last if $_->{msgno} > $msgno;
794         }
795         return undef;
796 }
797
798 # return the official filename for a message no
799 sub filename
800 {
801         return sprintf "$msgdir/m%06d", shift;
802 }
803
804 #
805 # return a list of valid elements 
806
807
808 sub fields
809 {
810         return keys(%valid);
811 }
812
813 #
814 # return a prompt for a field
815 #
816
817 sub field_prompt
818
819         my ($self, $ele) = @_;
820         return $valid{$ele};
821 }
822
823 #
824 # send a message state machine
825 sub do_send_stuff
826 {
827         my $self = shift;
828         my $line = shift;
829         my @out;
830         
831         if ($self->state eq 'send1') {
832                 #  $DB::single = 1;
833                 confess "local var gone missing" if !ref $self->{loc};
834                 my $loc = $self->{loc};
835                 $loc->{subject} = $line;
836                 $loc->{lines} = [];
837                 $self->state('sendbody');
838                 #push @out, $self->msg('sendbody');
839                 push @out, $self->msg('m8');
840         } elsif ($self->state eq 'sendbody') {
841                 confess "local var gone missing" if !ref $self->{loc};
842                 my $loc = $self->{loc};
843                 if ($line eq "\032" || $line eq '%1A' || uc $line eq "/EX") {
844                         my $to;
845                         
846                         foreach $to (@{$loc->{to}}) {
847                                 my $ref;
848                                 my $systime = $main::systime;
849                                 my $mycall = $main::mycall;
850                                 $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'),
851                                                                         uc $to,
852                                                                         exists $loc->{from} ? $loc->{from} : $self->call, 
853                                                                         $systime,
854                                                                         $loc->{private}, 
855                                                                         $loc->{subject}, 
856                                                                         exists $loc->{origin} ? $loc->{origin} : $mycall,
857                                                                         '0',
858                                                                         $loc->{rrreq});
859                                 $ref->swop_it($self->call);
860                                 $ref->store($loc->{lines});
861                                 $ref->add_dir();
862                                 push @out, $self->msg('m11', $ref->{msgno}, $to);
863                                 #push @out, "msgno $ref->{msgno} sent to $to";
864                                 my $dxchan = DXChannel->get(uc $to);
865                                 if ($dxchan) {
866                                         if ($dxchan->is_user()) {
867                                                 $dxchan->send($dxchan->msg('m9'));
868                                         }
869                                 }
870                         }
871
872                         delete $loc->{lines};
873                         delete $loc->{to};
874                         delete $self->{loc};
875                         $self->func(undef);
876                         
877                         $self->state('prompt');
878                 } elsif ($line eq "\031" || uc $line eq "/ABORT" || uc $line eq "/QUIT") {
879                         #push @out, $self->msg('sendabort');
880                         push @out, $self->msg('m10');
881                         delete $loc->{lines};
882                         delete $loc->{to};
883                         delete $self->{loc};
884                         $self->func(undef);
885                         $self->state('prompt');
886                 } else {
887                         
888                         # i.e. it ain't and end or abort, therefore store the line
889                         push @{$loc->{lines}}, length($line) > 0 ? $line : " ";
890                 }
891         }
892         return (1, @out);
893 }
894
895 # return the standard directory line for this ref 
896 sub dir
897 {
898         my $ref = shift;
899         return sprintf "%6d%s%s%5d %8.8s %8.8s %-6.6s %5.5s %-30.30s", 
900                 $ref->msgno, $ref->read ? '-' : ' ', $ref->private ? 'p' : ' ', $ref->size,
901                         $ref->to, $ref->from, cldate($ref->t), ztime($ref->t), $ref->subject;
902 }
903
904 # load the forward table
905 sub load_forward
906 {
907         my @out;
908         my $s = readfilestr($forwardfn);
909         if ($s) {
910                 eval $s;
911                 push @out, $@ if $@;
912         }
913         return @out;
914 }
915
916 # load the bad message table
917 sub load_badmsg
918 {
919         my @out;
920         my $s = readfilestr($badmsgfn);
921         if ($s) {
922                 eval $s;
923                 push @out, $@ if $@;
924         }
925         return @out;
926 }
927
928 # load the swop message table
929 sub load_swop
930 {
931         my @out;
932         my $s = readfilestr($swopfn);
933         if ($s) {
934                 eval $s;
935                 push @out, $@ if $@;
936         }
937         return @out;
938 }
939
940 #
941 # forward that message or not according to the forwarding table
942 # returns 1 for forward, 0 - to ignore
943 #
944
945 sub forward_it
946 {
947         my $ref = shift;
948         my $call = shift;
949         my $i;
950         
951         for ($i = 0; $i < @forward; $i += 5) {
952                 my ($sort, $field, $pattern, $action, $bbs) = @forward[$i..($i+4)]; 
953                 my $tested;
954                 
955                 # are we interested?
956                 next if $ref->{private} && $sort ne 'P';
957                 next if !$ref->{private} && $sort ne 'B';
958                 
959                 # select field
960                 $tested = $ref->{to} if $field eq 'T';
961                 $tested = $ref->{from} if $field eq 'F';
962                 $tested = $ref->{origin} if $field eq 'O';
963                 $tested = $ref->{subject} if $field eq 'S';
964
965                 if (!$pattern || $tested =~ m{$pattern}i) {
966                         return 0 if $action eq 'I';
967                         return 1 if !$bbs || grep $_ eq $call, @{$bbs};
968                 }
969         }
970         return 0;
971 }
972
973 sub dump_it
974 {
975         my $ref = shift;
976         my $i;
977         
978         for ($i = 0; $i < @badmsg; $i += 3) {
979                 my ($sort, $field, $pattern) = @badmsg[$i..($i+2)]; 
980                 my $tested;
981                 
982                 # are we interested?
983                 next if $ref->{private} && $sort ne 'P';
984                 next if !$ref->{private} && $sort ne 'B';
985                 
986                 # select field
987                 $tested = $ref->{to} if $field eq 'T';
988                 $tested = $ref->{from} if $field eq 'F';
989                 $tested = $ref->{origin} if $field eq 'O';
990                 $tested = $ref->{subject} if $field eq 'S';
991
992                 if (!$pattern || $tested =~ m{$pattern}i) {
993                         return 1;
994                 }
995         }
996         return 0;
997 }
998
999 sub swop_it
1000 {
1001         my $ref = shift;
1002         my $call = shift;
1003         my $i;
1004         my $count = 0;
1005         
1006         for ($i = 0; $i < @swop; $i += 5) {
1007                 my ($sort, $field, $pattern, $tfield, $topattern) = @swop[$i..($i+4)]; 
1008                 my $tested;
1009                 my $swop;
1010                 my $old;
1011                 
1012                 # are we interested?
1013                 next if $ref->{private} && $sort ne 'P';
1014                 next if !$ref->{private} && $sort ne 'B';
1015                 
1016                 # select field
1017                 $tested = $ref->{to} if $field eq 'T';
1018                 $tested = $ref->{from} if $field eq 'F';
1019                 $tested = $ref->{origin} if $field eq 'O';
1020                 $tested = $ref->{subject} if $field eq 'S';
1021
1022                 # select swop field
1023                 $old = $swop = $ref->{to} if $tfield eq 'T';
1024                 $old = $swop = $ref->{from} if $tfield eq 'F';
1025                 $old = $swop = $ref->{origin} if $tfield eq 'O';
1026                 $old = $swop = $ref->{subject} if $tfield eq 'S';
1027
1028                 if ($tested =~ m{$pattern}i) {
1029                         if ($tested eq $swop) {
1030                                 $swop =~ s{$pattern}{$topattern}i;
1031                         } else {
1032                                 $swop = $topattern;
1033                         }
1034                         Log('msg', "Msg $ref->{msgno}: $tfield $old -> $swop");
1035                         Log('dbg', "Msg $ref->{msgno}: $tfield $old -> $swop");
1036                         $ref->{to} = $swop if $tfield eq 'T';
1037                         $ref->{from} = $swop if $tfield eq 'F';
1038                         $ref->{origin} = $swop if $tfield eq 'O';
1039                         $ref->{subject} = $swop if $tfield eq 'S';
1040                         ++$count;
1041                 }
1042         }
1043         return $count;
1044 }
1045
1046 # import any msgs in the import directory
1047 # the messages are in BBS format (but may have cluster extentions
1048 # so SB UK < GB7TLH is legal
1049 sub import_msgs
1050 {
1051         # are there any to do in this directory?
1052         return unless -d $importfn;
1053         unless (opendir(DIR, $importfn)) {
1054                 dbg('msg', "can\'t open $importfn $!");
1055                 Log('msg', "can\'t open $importfn $!");
1056                 return;
1057         } 
1058
1059         my @names = readdir(DIR);
1060         closedir(DIR);
1061         my $name;
1062         foreach $name (@names) {
1063                 next if $name =~ /^\./;
1064                 my $splitit = $name =~ /^split/;
1065                 my $fn = "$importfn/$name";
1066                 next unless -f $fn;
1067                 unless (open(MSG, $fn)) {
1068                         dbg('msg', "can\'t open import file $fn $!");
1069                         Log('msg', "can\'t open import file $fn $!");
1070                         unlink($fn);
1071                         next;
1072                 }
1073                 my @msg = map { chomp; $_ } <MSG>;
1074                 close(MSG);
1075                 unlink($fn);
1076                 my @out = import_one($DXProt::me, \@msg, $splitit);
1077                 Log('msg', @out);
1078         }
1079 }
1080
1081 # import one message as a list in bbs (as extended) mode
1082 # takes a reference to an array containing the whole message
1083 sub import_one
1084 {
1085         my $dxchan = shift;
1086         my $ref = shift;
1087         my $splitit = shift;
1088         my $private = '1';
1089         my $rr = '0';
1090         my $notincalls = 1;
1091         my $from = $dxchan->call;
1092         my $origin = $main::mycall;
1093         my @to;
1094         my @out;
1095                                 
1096         # first line;
1097         my $line = shift @$ref;
1098         my @f = split /\s+/, $line;
1099         unless (@f && $f[0] =~ /^(:?S|SP|SB|SEND)$/ ) {
1100                 my $m = "invalid first line in import '$line'";
1101                 dbg('MSG', $m );
1102                 return (1, $m);
1103         }
1104         while (@f) {
1105                 my $f = uc shift @f;
1106                 next if $f eq 'SEND';
1107
1108                 # private / noprivate / rr
1109                 if ($notincalls && ($f eq 'B' || $f eq 'SB' || $f =~ /^NOP/oi)) {
1110                         $private = '0';
1111                 } elsif ($notincalls && ($f eq 'P' || $f eq 'SP' || $f =~ /^PRI/oi)) {
1112                         ;
1113                 } elsif ($notincalls && ($f eq 'RR')) {
1114                         $rr = '1';
1115                 } elsif ($f eq '@' && @f) {       # this is bbs syntax, for origin
1116                         $origin = uc shift @f;
1117                 } elsif ($f eq '<' && @f) {     # this is bbs syntax  for from call
1118                         $from = uc shift @f;
1119                 } elsif ($f =~ /^\$/) {     # this is bbs syntax  for a bid
1120                         next;
1121                 } elsif ($f =~ /^<\S+/) {     # this is bbs syntax  for from call
1122                         ($from) = $f =~ /^<(\S+)$/;
1123                 } elsif ($f =~ /^\@\S+/) {     # this is bbs syntax for origin
1124                         ($origin) = $f =~ /^\@(\S+)$/;
1125                 } else {
1126
1127                         # callsign ?
1128                         $notincalls = 0;
1129
1130                         # is this callsign a distro?
1131                         my $fn = "$msgdir/distro/$f.pl";
1132                         if (-e $fn) {
1133                                 my $fh = new IO::File $fn;
1134                                 if ($fh) {
1135                                         local $/ = undef;
1136                                         my $s = <$fh>;
1137                                         $fh->close;
1138                                         my @call;
1139                                         @call = eval $s;
1140                                         return (1, "Error in Distro $f.pl:", $@) if $@;
1141                                         if (@call > 0) {
1142                                                 push @f, @call;
1143                                                 next;
1144                                         }
1145                                 }
1146                         }
1147                         
1148                         if (grep $_ eq $f, @DXMsg::badmsg) {
1149                                 push @out, $dxchan->msg('m3', $f);
1150                         } else {
1151                                 push @to, $f;
1152                         }
1153                 }
1154         }
1155         
1156         # subject is the next line
1157         my $subject = shift @$ref;
1158         
1159         # strip off trailing lines 
1160         pop @$ref while (@$ref && $$ref[-1] =~ /^\s*$/);
1161         
1162         # strip off /EX or /ABORT
1163         return ("aborted") if @$ref && $$ref[-1] =~ m{^/ABORT$}i; 
1164         pop @$ref if (@$ref && $$ref[-1] =~ m{^/EX$}i);                                                                  
1165
1166         # sort out any splitting that needs to be done
1167         my @chunk;
1168         if ($splitit) {
1169                 my $lth = 0;
1170                 my $lines = [];
1171                 for (@$ref) {
1172                         if ($lth >= $maxchunk || ($lth > $minchunk && /^\s*$/)) {
1173                                 push @chunk, $lines;
1174                                 $lines = [];
1175                                 $lth = 0;
1176                         } 
1177                         push @$lines, $_;
1178                         $lth += length; 
1179                 }
1180                 push @chunk, $lines if @$lines;
1181         } else {
1182                 push @chunk, $ref;
1183         }
1184                                   
1185     # write all the messages away
1186         my $i;
1187         for ( $i = 0;  $i < @chunk; $i++) {
1188                 my $chunk = $chunk[$i];
1189                 my $ch_subject;
1190                 if (@chunk > 1) {
1191                         my $num = " [" . ($i+1) . "/" . scalar @chunk . "]";
1192                         $ch_subject = substr($subject, 0, 27 - length $num) .  $num;
1193                 } else {
1194                         $ch_subject = $subject;
1195                 }
1196                 my $to;
1197                 foreach $to (@to) {
1198                         my $systime = $main::systime;
1199                         my $mycall = $main::mycall;
1200                         my $mref = DXMsg->alloc(DXMsg::next_transno('Msgno'),
1201                                                                         $to,
1202                                                                         $from, 
1203                                                                         $systime,
1204                                                                         $private, 
1205                                                                         $ch_subject, 
1206                                                                         $origin,
1207                                                                         '0',
1208                                                                         $rr);
1209                         $mref->swop_it($main::mycall);
1210                         $mref->store($chunk);
1211                         $mref->add_dir();
1212                         push @out, $dxchan->msg('m11', $mref->{msgno}, $to);
1213                         #push @out, "msgno $ref->{msgno} sent to $to";
1214                         my $todxchan = DXChannel->get(uc $to);
1215                         if ($todxchan) {
1216                                 if ($todxchan->is_user()) {
1217                                         $todxchan->send($todxchan->msg('m9'));
1218                                 }
1219                         }
1220                 }
1221         }
1222         return @out;
1223 }
1224
1225 no strict;
1226 sub AUTOLOAD
1227 {
1228         my $self = shift;
1229         my $name = $AUTOLOAD;
1230         return if $name =~ /::DESTROY$/;
1231         $name =~ s/.*:://o;
1232         
1233         confess "Non-existant field '$AUTOLOAD'" if !$valid{$name};
1234         # this clever line of code creates a subroutine which takes over from autoload
1235         # from OO Perl - Conway
1236         *{$AUTOLOAD} = sub {@_ > 1 ? $_[0]->{$name} = $_[1] : $_[0]->{$name}} ;
1237         @_ ? $self->{$name} = shift : $self->{$name} ;
1238 }
1239
1240 1;
1241
1242 __END__