3 # This module impliments the message handling for a dx cluster
5 # Copyright (c) 1998 Dirk Koopman G1TLH
10 # Notes for implementors:-
12 # PC28 field 11 is the RR required flag
13 # PC28 field 12 is a VIA routing (ie it is a node call)
18 @ISA = qw(DXProt DXChannel);
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);
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
51 $minchunk = 4800; # minimum chunk size for a split message
52 $maxchunk = 6000; # maximum chunk size
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
61 fromnode => '5,From Node',
62 tonode => '5,To Node',
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',
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',
80 keep => '0,Keep this?,yesno',
81 lastt => '5,Last processed,cldatetime',
82 waitt => '5,Wait until,cldatetime',
92 # allocate a new object
93 # called fromnode, tonode, from, to, datetime, private?, subject, nolinesper
97 my $self = bless {}, $pkg;
98 $self->{msgno} = shift;
101 $self->{to} = ($to eq $main::mycall) ? $main::myalias : $to;
104 $self->{from} = uc $from;
106 $self->{private} = shift;
107 $self->{subject} = shift;
108 $self->{origin} = shift;
109 $self->{'read'} = shift;
110 $self->{rrreq} = shift;
112 $self->{lastt} = $main::systime;
120 delete $ref->{lines};
121 delete $ref->{linesreq};
122 delete $ref->{tonode};
123 delete $ref->{fromnode};
124 delete $ref->{stream};
125 delete $ref->{lines};
127 delete $ref->{count};
128 delete $ref->{lastt} if exists $ref->{lastt};
129 delete $ref->{waitt} if exists $ref->{waitt};
134 my ($self, $line) = @_;
136 # this is periodic processing
137 if (!$self || !$line) {
139 if ($main::systime > $lastq + $queueinterval) {
141 # wander down the work queue stopping any messages that have timed out
145 if (exists $ref->{lastt} && $main::systime >= $ref->{lastt} + $timeout) {
146 dbg('msg', "Timeout, stopping msgno: $ref->{msgno} -> $node");
147 $ref->stop_msg($node);
149 # delay any outgoing messages that fail
150 $ref->{waitt} = $main::systime + $waittime + rand(120) if $node ne $main::mycall;
154 # queue some message if the interval timer has gone off
157 # import any messages in the import directory
160 $lastq = $main::systime;
163 # clean the message queue
164 clean_old() if $main::systime - $last_clean > 3600 ;
168 my @f = split /\^/, $line;
169 my ($pcno) = $f[0] =~ /^PC(\d\d)/; # just get the number
172 if ($pcno == 28) { # incoming message
174 # first look for any messages in the busy queue
175 # and cancel them this should both resolve timed out incoming messages
176 # and crossing of message between nodes, incoming messages have priority
177 if (exists $busy{$f[2]}) {
178 my $ref = $busy{$f[2]};
179 my $tonode = $ref->{tonode};
180 dbg('msg', "Busy, stopping msgno: $ref->{msgno} -> $f[2]");
181 $ref->stop_msg($self->call);
184 my $t = cltounix($f[5], $f[6]);
185 my $stream = next_transno($f[2]);
186 my $ref = DXMsg->alloc($stream, uc $f[3], $f[4], $t, $f[7], $f[8], $f[13], '0', $f[11]);
188 # fill in various forwarding state variables
189 $ref->{fromnode} = $f[2];
190 $ref->{tonode} = $f[1];
191 $ref->{rrreq} = $f[11];
192 $ref->{linesreq} = $f[10];
193 $ref->{stream} = $stream;
194 $ref->{count} = 0; # no of lines between PC31s
195 dbg('msg', "new message from $f[4] to $f[3] '$f[8]' stream $stream\n");
196 $work{"$f[2]$stream"} = $ref; # store in work
197 $busy{$f[2]} = $ref; # set interlock
198 $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack
199 $ref->{lastt} = $main::systime;
201 # look to see whether this is a non private message sent to a known callsign
202 my $uref = DXUser->get_current($ref->{to});
203 if (iscallsign($ref->{to}) && !$ref->{private} && $uref && $uref->homenode) {
205 dbg('msg', "set bull to $ref->{to} to private");
210 if ($pcno == 29) { # incoming text
211 my $ref = $work{"$f[2]$f[3]"};
213 push @{$ref->{lines}}, $f[4];
215 if ($ref->{count} >= $ref->{linesreq}) {
216 $self->send(DXProt::pc31($f[2], $f[1], $f[3]));
217 dbg('msg', "stream $f[3]: $ref->{count} lines received\n");
220 $ref->{lastt} = $main::systime;
222 dbg('msg', "PC29 from unknown stream $f[3] from $f[2]" );
223 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
228 if ($pcno == 30) { # this is a incoming subject ack
229 my $ref = $work{$f[2]}; # note no stream at this stage
232 $ref->{stream} = $f[3];
234 $ref->{linesreq} = 5;
235 $work{"$f[2]$f[3]"} = $ref; # new ref
236 dbg('msg', "incoming subject ack stream $f[3]\n");
237 $busy{$f[2]} = $ref; # interlock
239 push @{$ref->{lines}}, ($ref->read_msg_body);
240 $ref->send_tranche($self);
241 $ref->{lastt} = $main::systime;
243 dbg('msg', "PC30 from unknown stream $f[3] from $f[2]" );
244 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
249 if ($pcno == 31) { # acknowledge a tranche of lines
250 my $ref = $work{"$f[2]$f[3]"};
252 dbg('msg', "tranche ack stream $f[3]\n");
253 $ref->send_tranche($self);
254 $ref->{lastt} = $main::systime;
256 dbg('msg', "PC31 from unknown stream $f[3] from $f[2]" );
257 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
262 if ($pcno == 32) { # incoming EOM
263 dbg('msg', "stream $f[3]: EOM received\n");
264 my $ref = $work{"$f[2]$f[3]"};
266 $self->send(DXProt::pc33($f[2], $f[1], $f[3])); # acknowledge it
268 # get the next msg no - note that this has NOTHING to do with the stream number in PC protocol
269 # store the file or message
270 # remove extraneous rubbish from the hash
271 # remove it from the work in progress vector
272 # stuff it on the msg queue
275 $ref->store($ref->{lines});
278 # does an identical message already exist?
281 if ($ref->{subject} eq $m->{subject} && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from} && $ref->{to} eq $m->{to}) {
282 $ref->stop_msg($self->call);
283 my $msgno = $m->{msgno};
284 dbg('msg', "duplicate message to $msgno\n");
285 Log('msg', "duplicate message to $msgno");
291 $ref->swop_it($self->call);
293 # look for 'bad' to addresses
294 # if (grep $ref->{to} eq $_, @badmsg) {
295 if ($ref->dump_it($self->call)) {
296 $ref->stop_msg($self->call);
297 dbg('msg', "'Bad' message $ref->{to}");
298 Log('msg', "'Bad' message $ref->{to}");
302 $ref->{msgno} = next_transno("Msgno");
303 push @{$ref->{gotit}}, $f[2]; # mark this up as being received
304 $ref->store($ref->{lines});
306 my $dxchan = DXChannel->get($ref->{to});
307 $dxchan->send($dxchan->msg('m9')) if $dxchan && $dxchan->is_user;
308 Log('msg', "Message $ref->{msgno} from $ref->{from} received from $f[2] for $ref->{to}");
311 $ref->stop_msg($self->call);
313 dbg('msg', "PC32 from unknown stream $f[3] from $f[2]" );
314 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
320 if ($pcno == 33) { # acknowledge the end of message
321 my $ref = $work{"$f[2]$f[3]"};
323 if ($ref->{private}) { # remove it if it private and gone off site#
324 Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2] and deleted");
327 Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2]");
328 push @{$ref->{gotit}}, $f[2]; # mark this up as being received
329 $ref->store($ref->{lines}); # re- store the file
331 $ref->stop_msg($self->call);
333 dbg('msg', "PC33 from unknown stream $f[3] from $f[2]" );
334 $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream
337 # send next one if present
342 if ($pcno == 40) { # this is a file request
343 $f[3] =~ s/\\/\//og; # change the slashes
344 $f[3] =~ s/\.//og; # remove dots
345 $f[3] =~ s/^\///o; # remove the leading /
346 $f[3] = lc $f[3]; # to lower case;
347 dbg('msg', "incoming file $f[3]\n");
348 $f[3] = 'packclus/' . $f[3] unless $f[3] =~ /^packclus\//o;
350 # create any directories
351 my @part = split /\//, $f[3];
353 my $fn = "$main::root";
354 pop @part; # remove last part
355 foreach $part (@part) {
358 last SWITCH if !mkdir $fn, 0777;
359 dbg('msg', "created directory $fn\n");
361 my $stream = next_transno($f[2]);
362 my $ref = DXMsg->alloc($stream, "$main::root/$f[3]", $self->call, time, !$f[4], $f[3], ' ', '0', '0');
364 # forwarding variables
365 $ref->{fromnode} = $f[1];
366 $ref->{tonode} = $f[2];
367 $ref->{linesreq} = $f[5];
368 $ref->{stream} = $stream;
369 $ref->{count} = 0; # no of lines between PC31s
371 $ref->{lastt} = $main::systime;
372 $work{"$f[2]$stream"} = $ref; # store in work
373 $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack
378 if ($pcno == 42) { # abort transfer
379 dbg('msg', "stream $f[3]: abort received\n");
380 my $ref = $work{"$f[2]$f[3]"};
382 $ref->stop_msg($self->call);
388 if ($pcno == 49) { # global delete on subject
390 if ($_->{from} eq $f[1] && $_->{subject} eq $f[2]) {
392 Log('msg', "Message $_->{msgno} from $_->{from} ($_->{subject}) fully deleted");
393 DXProt::broadcast_ak1a($line, $self);
401 # store a message away on disc or whatever
403 # NOTE the second arg is a REFERENCE not a list
409 # we only proceed if there are actually any lines in the file
410 # if (!$lines || @{$lines} == 0) {
414 if ($ref->{file}) { # a file
415 dbg('msg', "To be stored in $ref->{to}\n");
417 my $fh = new IO::File "$ref->{to}", "w";
420 foreach $line (@{$lines}) {
424 dbg('msg', "file $ref->{to} stored\n");
425 Log('msg', "file $ref->{to} from $ref->{from} stored" );
427 confess "can't open file $ref->{to} $!";
429 } else { # a normal message
431 # attempt to open the message file
432 my $fn = filename($ref->{msgno});
434 dbg('msg', "To be stored in $fn\n");
436 # now save the file, overwriting what's there, YES I KNOW OK! (I will change it if it's a problem)
437 my $fh = new IO::File "$fn", "w";
439 my $rr = $ref->{rrreq} ? '1' : '0';
440 my $priv = $ref->{private} ? '1': '0';
441 print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{'read'}^$rr\n";
442 print $fh "=== ", join('^', @{$ref->{gotit}}), "\n";
445 foreach $line (@{$lines}) {
446 $ref->{size} += (length $line) + 1;
450 dbg('msg', "msg $ref->{msgno} stored\n");
451 Log('msg', "msg $ref->{msgno} from $ref->{from} to $ref->{to} stored" );
453 confess "can't open msg file $fn $!";
463 # remove it from the active message list
464 @msg = map { $_ != $self ? $_ : () } @msg;
466 # belt and braces (one day I will ask someone if this is REALLY necessary)
467 delete $self->{gotit};
468 delete $self->{list};
471 unlink filename($self->{msgno});
472 dbg('msg', "deleting $self->{msgno}\n");
475 # clean out old messages from the message queue
480 # mark old messages for deletion
481 foreach $ref (@msg) {
482 if (!$ref->{keep} && $ref->{t} < $main::systime - $maxage) {
483 $ref->{deleteme} = 1;
484 delete $ref->{gotit};
486 unlink filename($ref->{msgno});
487 dbg('msg', "deleting old $ref->{msgno}\n");
491 # remove them all from the active message list
492 @msg = map { $_->{deleteme} ? () : $_ } @msg;
493 $last_clean = $main::systime;
496 # read in a message header
506 $file = new IO::File "$fn";
508 dbg('err', "Error reading $fn $!");
512 $line = <$file>; # first line
514 $size -= length $line;
515 if (! $line =~ /^===/o) {
516 dbg('err', "corrupt first line in $fn ($line)");
520 @f = split /\^/, $line;
521 $ref = DXMsg->alloc(@f);
523 $line = <$file>; # second line
525 $size -= length $line;
526 if (! $line =~ /^===/o) {
527 dbg('err', "corrupt second line in $fn ($line)");
532 @f = split /\^/, $line;
533 push @{$ref->{gotit}}, @f;
534 $ref->{size} = $size;
541 # read in a message header
545 my $msgno = $self->{msgno};
548 my $fn = filename($msgno);
551 $file = new IO::File;
552 if (!open($file, $fn)) {
553 dbg('err' ,"Error reading $fn $!");
556 @out = map {chomp; $_} <$file>;
559 shift @out if $out[0] =~ /^=== /;
560 shift @out if $out[0] =~ /^=== /;
564 # send a tranche of lines to the other end
567 my ($self, $dxchan) = @_;
569 my $to = $self->{tonode};
570 my $from = $self->{fromnode};
571 my $stream = $self->{stream};
572 my $lines = $self->{lines};
575 for ($i = 0, $c = $self->{count}; $i < $self->{linesreq} && $c < @$lines; $i++, $c++) {
576 push @out, DXProt::pc29($to, $from, $stream, $lines->[$c]);
580 push @out, DXProt::pc32($to, $from, $stream) if $i < $self->{linesreq};
585 # find a message to send out and start the ball rolling
592 my @nodelist = DXChannel::get_all_ak1a();
594 # bat down the message list looking for one that needs to go off site and whose
595 # nearest node is not busy.
597 dbg('msg', "queue msg ($sort)\n");
598 foreach $ref (@msg) {
599 # firstly, is it private and unread? if so can I find the recipient
600 # in my cluster node list offsite?
602 # ignore 'delayed' messages until their waiting time has expired
603 if (exists $ref->{waitt}) {
604 next if $ref->{waitt} > $main::systime;
605 delete $ref->{waitt};
608 # deal with routed private messages
610 if ($ref->{private}) {
611 next if $ref->{'read'}; # if it is read, it is stuck here
612 $clref = DXCluster->get_exact($ref->{to});
613 unless ($clref) { # otherwise look for a homenode
614 my $uref = DXUser->get($ref->{to});
615 my $hnode = $uref->homenode if $uref;
616 $clref = DXCluster->get_exact($hnode) if $hnode;
618 if ($clref && !grep { $clref->{dxchan} == $_ } DXCommandmode::get_all()) {
619 next if $clref->call eq $main::mycall; # i.e. it lives here
620 $noderef = $clref->{dxchan};
621 $ref->start_msg($noderef) if !get_busy($noderef->call) && $noderef->state eq 'normal';
625 # otherwise we are dealing with a bulletin or forwarded private message
626 # compare the gotit list with
627 # the nodelist up above, if there are sites that haven't got it yet
628 # then start sending it - what happens when we get loops is anyone's
629 # guess, use (to, from, time, subject) tuple?
630 foreach $noderef (@nodelist) {
631 next if $noderef->call eq $main::mycall;
632 next if grep { $_ eq $noderef->call } @{$ref->{gotit}};
633 next unless $ref->forward_it($noderef->call); # check the forwarding file
635 # if we are here we have a node that doesn't have this message
636 $ref->start_msg($noderef) if !get_busy($noderef->call) && $noderef->state eq 'normal';
640 # if all the available nodes are busy then stop
641 last if @nodelist == scalar grep { get_busy($_->call) } @nodelist;
645 # is there a message for me?
651 foreach $ref (@msg) {
652 # is it for me, private and unread?
653 if ($ref->{to} eq $call && $ref->{private}) {
654 return 1 if !$ref->{'read'};
660 # start the message off on its travels with a PC28
663 my ($self, $dxchan) = @_;
665 dbg('msg', "start msg $self->{msgno}\n");
666 $self->{linesreq} = 5;
668 $self->{tonode} = $dxchan->call;
669 $self->{fromnode} = $main::mycall;
670 $busy{$self->{tonode}} = $self;
671 $work{$self->{tonode}} = $self;
672 $self->{lastt} = $main::systime;
673 $dxchan->send(DXProt::pc28($self->{tonode}, $self->{fromnode}, $self->{to}, $self->{from}, $self->{t}, $self->{private}, $self->{subject}, $self->{origin}, $self->{rrreq}));
676 # get the ref of a busy node
689 # get the forwarding queue
695 # stop a message from continuing, clean it out, unlock interlocks etc
700 my $stream = $self->{stream} if exists $self->{stream};
703 dbg('msg', "stop msg $self->{msgno} -> node $node\n");
705 delete $work{"$node$stream"} if $stream;
710 # get a new transaction number from the file specified
714 $name =~ s/\W//og; # remove non-word characters
715 my $fn = "$msgdir/$name";
718 my $fh = new IO::File;
719 if (sysopen($fh, $fn, O_RDWR|O_CREAT, 0666)) {
721 $msgno = $fh->getline;
725 $fh->print("$msgno\n");
726 dbg('msg', "msgno $msgno allocated for $name\n");
729 confess "can't open $fn $!";
734 # initialise the message 'system', read in all the message headers
737 my $dir = new IO::File;
741 # load various control files
742 dbg('err', "load badmsg: " . (load_badmsg() or "Ok"));
743 dbg('err', "load forward: " . (load_forward() or "Ok"));
744 dbg('err', "load swop: " . (load_swop() or "Ok"));
746 # read in the directory
747 opendir($dir, $msgdir) or confess "can't open $msgdir $!";
748 @dir = readdir($dir);
753 next unless /^m\d+$/o;
755 $ref = read_msg_header("$msgdir/$_");
758 # delete any messages to 'badmsg.pl' places
759 if (grep $ref->{to} eq $_, @badmsg) {
760 dbg('msg', "'Bad' TO address $ref->{to}");
761 Log('msg', "'Bad' TO address $ref->{to}");
766 # add the message to the available queue
771 # add the message to the directory listing
775 confess "tried to add a non-ref to the msg directory" if !ref $ref;
779 # return all the current messages
785 # get a particular message
790 return $_ if $_->{msgno} == $msgno;
791 last if $_->{msgno} > $msgno;
796 # return the official filename for a message no
799 return sprintf "$msgdir/m%06d", shift;
803 # return a list of valid elements
812 # return a prompt for a field
817 my ($self, $ele) = @_;
822 # send a message state machine
829 if ($self->state eq 'send1') {
831 confess "local var gone missing" if !ref $self->{loc};
832 my $loc = $self->{loc};
833 $loc->{subject} = $line;
835 $self->state('sendbody');
836 #push @out, $self->msg('sendbody');
837 push @out, $self->msg('m8');
838 } elsif ($self->state eq 'sendbody') {
839 confess "local var gone missing" if !ref $self->{loc};
840 my $loc = $self->{loc};
841 if ($line eq "\032" || $line eq '%1A' || uc $line eq "/EX") {
844 foreach $to (@{$loc->{to}}) {
846 my $systime = $main::systime;
847 my $mycall = $main::mycall;
848 $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'),
850 exists $loc->{from} ? $loc->{from} : $self->call,
854 exists $loc->{origin} ? $loc->{origin} : $mycall,
857 $ref->swop_it($self->call);
858 $ref->store($loc->{lines});
860 push @out, $self->msg('m11', $ref->{msgno}, $to);
861 #push @out, "msgno $ref->{msgno} sent to $to";
862 my $dxchan = DXChannel->get(uc $to);
864 if ($dxchan->is_user()) {
865 $dxchan->send($dxchan->msg('m9'));
870 delete $loc->{lines};
875 $self->state('prompt');
876 } elsif ($line eq "\031" || uc $line eq "/ABORT" || uc $line eq "/QUIT") {
877 #push @out, $self->msg('sendabort');
878 push @out, $self->msg('m10');
879 delete $loc->{lines};
883 $self->state('prompt');
886 # i.e. it ain't and end or abort, therefore store the line
887 push @{$loc->{lines}}, length($line) > 0 ? $line : " ";
893 # return the standard directory line for this ref
897 return sprintf "%6d%s%s%5d %8.8s %8.8s %-6.6s %5.5s %-30.30s",
898 $ref->msgno, $ref->read ? '-' : ' ', $ref->private ? 'p' : ' ', $ref->size,
899 $ref->to, $ref->from, cldate($ref->t), ztime($ref->t), $ref->subject;
902 # load the forward table
906 my $s = readfilestr($forwardfn);
914 # load the bad message table
918 my $s = readfilestr($badmsgfn);
926 # load the swop message table
930 my $s = readfilestr($swopfn);
939 # forward that message or not according to the forwarding table
940 # returns 1 for forward, 0 - to ignore
949 for ($i = 0; $i < @forward; $i += 5) {
950 my ($sort, $field, $pattern, $action, $bbs) = @forward[$i..($i+4)];
954 next if $ref->{private} && $sort ne 'P';
955 next if !$ref->{private} && $sort ne 'B';
958 $tested = $ref->{to} if $field eq 'T';
959 $tested = $ref->{from} if $field eq 'F';
960 $tested = $ref->{origin} if $field eq 'O';
961 $tested = $ref->{subject} if $field eq 'S';
963 if (!$pattern || $tested =~ m{$pattern}i) {
964 return 0 if $action eq 'I';
965 return 1 if !$bbs || grep $_ eq $call, @{$bbs};
977 for ($i = 0; $i < @badmsg; $i += 3) {
978 my ($sort, $field, $pattern) = @badmsg[$i..($i+2)];
982 next if $ref->{private} && $sort ne 'P';
983 next if !$ref->{private} && $sort ne 'B';
986 $tested = $ref->{to} if $field eq 'T';
987 $tested = $ref->{from} if $field eq 'F';
988 $tested = $ref->{origin} if $field eq 'O';
989 $tested = $ref->{subject} if $field eq 'S';
991 if (!$pattern || $tested =~ m{$pattern}i) {
1005 for ($i = 0; $i < @swop; $i += 5) {
1006 my ($sort, $field, $pattern, $tfield, $topattern) = @swop[$i..($i+4)];
1011 # are we interested?
1012 next if $ref->{private} && $sort ne 'P';
1013 next if !$ref->{private} && $sort ne 'B';
1016 $tested = $ref->{to} if $field eq 'T';
1017 $tested = $ref->{from} if $field eq 'F';
1018 $tested = $ref->{origin} if $field eq 'O';
1019 $tested = $ref->{subject} if $field eq 'S';
1022 $old = $swop = $ref->{to} if $tfield eq 'T';
1023 $old = $swop = $ref->{from} if $tfield eq 'F';
1024 $old = $swop = $ref->{origin} if $tfield eq 'O';
1025 $old = $swop = $ref->{subject} if $tfield eq 'S';
1027 if ($tested =~ m{$pattern}i) {
1028 if ($tested eq $swop) {
1029 $swop =~ s{$pattern}{$topattern}i;
1033 Log('msg', "Msg $ref->{msgno}: $tfield $old -> $swop");
1034 Log('dbg', "Msg $ref->{msgno}: $tfield $old -> $swop");
1035 $ref->{to} = $swop if $tfield eq 'T';
1036 $ref->{from} = $swop if $tfield eq 'F';
1037 $ref->{origin} = $swop if $tfield eq 'O';
1038 $ref->{subject} = $swop if $tfield eq 'S';
1045 # import any msgs in the import directory
1046 # the messages are in BBS format (but may have cluster extentions
1047 # so SB UK < GB7TLH is legal
1050 # are there any to do in this directory?
1051 return unless -d $importfn;
1052 unless (opendir(DIR, $importfn)) {
1053 dbg('msg', "can\'t open $importfn $!");
1054 Log('msg', "can\'t open $importfn $!");
1058 my @names = readdir(DIR);
1061 foreach $name (@names) {
1062 next if $name =~ /^\./;
1063 my $splitit = $name =~ /^split/;
1064 my $fn = "$importfn/$name";
1066 unless (open(MSG, $fn)) {
1067 dbg('msg', "can\'t open import file $fn $!");
1068 Log('msg', "can\'t open import file $fn $!");
1072 my @msg = map { chomp; $_ } <MSG>;
1075 my @out = import_one($DXProt::me, \@msg, $splitit);
1080 # import one message as a list in bbs (as extended) mode
1081 # takes a reference to an array containing the whole message
1086 my $splitit = shift;
1090 my $from = $dxchan->call;
1091 my $origin = $main::mycall;
1096 my $line = shift @$ref;
1097 my @f = split /\s+/, $line;
1098 unless (@f && $f[0] =~ /^(:?S|SP|SB|SEND)$/ ) {
1099 my $m = "invalid first line in import '$line'";
1104 my $f = uc shift @f;
1105 next if $f eq 'SEND';
1107 # private / noprivate / rr
1108 if ($notincalls && ($f eq 'B' || $f eq 'SB' || $f =~ /^NOP/oi)) {
1110 } elsif ($notincalls && ($f eq 'P' || $f eq 'SP' || $f =~ /^PRI/oi)) {
1112 } elsif ($notincalls && ($f eq 'RR')) {
1114 } elsif ($f eq '@' && @f) { # this is bbs syntax, for origin
1115 $origin = uc shift @f;
1116 } elsif ($f eq '<' && @f) { # this is bbs syntax for from call
1117 $from = uc shift @f;
1118 } elsif ($f =~ /^\$/) { # this is bbs syntax for a bid
1120 } elsif ($f =~ /^<\S+/) { # this is bbs syntax for from call
1121 ($from) = $f =~ /^<(\S+)$/;
1122 } elsif ($f =~ /^\@\S+/) { # this is bbs syntax for origin
1123 ($origin) = $f =~ /^\@(\S+)$/;
1129 # is this callsign a distro?
1130 my $fn = "$msgdir/distro/$f.pl";
1132 my $fh = new IO::File $fn;
1139 return (1, "Error in Distro $f.pl:", $@) if $@;
1147 if (grep $_ eq $f, @DXMsg::badmsg) {
1148 push @out, $dxchan->msg('m3', $f);
1155 # subject is the next line
1156 my $subject = shift @$ref;
1158 # strip off trailing lines
1159 pop @$ref while (@$ref && $$ref[-1] =~ /^\s*$/);
1161 # strip off /EX or /ABORT
1162 return ("aborted") if @$ref && $$ref[-1] =~ m{^/ABORT$}i;
1163 pop @$ref if (@$ref && $$ref[-1] =~ m{^/EX$}i);
1165 # sort out any splitting that needs to be done
1171 if ($lth >= $maxchunk || ($lth > $minchunk && /^\s*$/)) {
1172 push @chunk, $lines;
1179 push @chunk, $lines if @$lines;
1184 # write all the messages away
1186 for ( $i = 0; $i < @chunk; $i++) {
1187 my $chunk = $chunk[$i];
1190 my $num = " [" . ($i+1) . "/" . scalar @chunk . "]";
1191 $ch_subject = substr($subject, 0, 27 - length $num) . $num;
1193 $ch_subject = $subject;
1197 my $systime = $main::systime;
1198 my $mycall = $main::mycall;
1199 my $mref = DXMsg->alloc(DXMsg::next_transno('Msgno'),
1208 $mref->swop_it($main::mycall);
1209 $mref->store($chunk);
1211 push @out, $dxchan->msg('m11', $mref->{msgno}, $to);
1212 #push @out, "msgno $ref->{msgno} sent to $to";
1213 my $todxchan = DXChannel->get(uc $to);
1215 if ($todxchan->is_user()) {
1216 $todxchan->send($todxchan->msg('m9'));
1228 my $name = $AUTOLOAD;
1229 return if $name =~ /::DESTROY$/;
1232 confess "Non-existant field '$AUTOLOAD'" if !$valid{$name};
1233 @_ ? $self->{$name} = shift : $self->{$name} ;