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