X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FDXMsg.pm;h=04239f16b6777581ade86c2cb0f036f087d1f42b;hb=66581498212e2d0643c50369b1f39b85d7dc4dce;hp=f66f98d7f8778c22084a234a6a189e37881f6302;hpb=1283d7c748715a1bf85a781fe4b66f427d949ae9;p=spider.git diff --git a/perl/DXMsg.pm b/perl/DXMsg.pm index f66f98d7..04239f16 100644 --- a/perl/DXMsg.pm +++ b/perl/DXMsg.pm @@ -34,7 +34,7 @@ $BRANCH = sprintf( "%d.%03d", q$Revision$ =~ /\d+\.\d+\.(\d+)\.(\d+)/ ) || 0; $main::build += $VERSION; $main::branch += $BRANCH; -use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean +use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean $residencetime @badmsg @swop $swopfn $badmsgfn $forwardfn @forward $timeout $waittime $queueinterval $lastq $importfn $minchunk $maxchunk $bulltopriv); @@ -55,6 +55,7 @@ $lastq = 0; $minchunk = 4800; # minimum chunk size for a split message $maxchunk = 6000; # maximum chunk size $bulltopriv = 1; # convert msgs with callsigns to private if they are bulls +$residencetime = 2*86400; # keep deleted messages for this amount of time $badmsgfn = "$msgdir/badmsg.pl"; # list of TO address we wont store @@ -87,6 +88,7 @@ $importfn = "$msgdir/import"; # import directory lastt => '5,Last processed,cldatetime', waitt => '5,Wait until,cldatetime', delete => '5,Awaiting Delete,yesno', + deletetime => '5,Deletion Time,cldatetime', ); # allocate a new object @@ -108,6 +110,8 @@ sub alloc $self->{origin} = shift; $self->{'read'} = shift; $self->{rrreq} = shift; + $self->{delete} = shift; + $self->{deletetime} = shift; $self->{gotit} = []; # $self->{lastt} = $main::systime; $self->{lines} = []; @@ -137,13 +141,14 @@ sub process # clean the message queue clean_old() if $main::systime - $last_clean > 3600 ; + $last_clean = $main::systime; return; } my @f = split /\^/, $line; my ($pcno) = $f[0] =~ /^PC(\d\d)/; # just get the number my ($tonode, $fromnode) = @f[1, 2]; - my $stream = $f[3] if $pcno > 29 && $pcno <= 33; + my $stream = $f[3] if ($pcno >= 29 && $pcno <= 33) || $pcno == 42; SWITCH: { if ($pcno == 28) { # incoming message @@ -195,7 +200,22 @@ sub process my $ref = get_fwq($fromnode, $stream); if ($ref) { $f[4] =~ s/\%5E/^/g; - push @{$ref->{lines}}, $f[4]; + if (@{$ref->{lines}}) { + push @{$ref->{lines}}, $f[4]; + } else { + # temporarily store any R: lines so that we end up with + # only the first and last ones stored. + if ($f[4] =~ m|^R:\d{6}/\d{4}|) { + push @{$ref->{tempr}}, $f[4]; + } else { + if (exists $ref->{tempr}) { + push @{$ref->{lines}}, shift @{$ref->{tempr}}; + push @{$ref->{lines}}, pop @{$ref->{tempr}} if @{$ref->{tempr}}; + delete $ref->{tempr}; + } + push @{$ref->{lines}}, $f[4]; + } + } $ref->{count}++; if ($ref->{count} >= $ref->{linesreq}) { $self->send(DXProt::pc31($fromnode, $tonode, $stream)); @@ -262,7 +282,7 @@ sub process # does an identical message already exist? my $m; for $m (@msg) { - if ($ref->{subject} eq $m->{subject} && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from} && $ref->{to} eq $m->{to}) { + if (substr($ref->{subject},0,28) eq substr($m->{subject},0,28) && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from} && $ref->{to} eq $m->{to}) { $ref->stop_msg($fromnode); my $msgno = $m->{msgno}; dbg("duplicate message from $ref->{from} -> $ref->{to} to msg: $msgno") if isdbg('msg'); @@ -289,8 +309,8 @@ sub process } push @words, BadWords::check($ref->{subject}); if (@words) { - dbg("message with badwords '@words' $ref->{from} -> $ref->{to} '$ref->{subject}' origin: $ref->{origin} via " . $self->call) if isdbg('msg'); - Log('msg',"message with badwords '@words' $ref->{from} -> $ref->{to} origin: $ref->{origin} via " . $self->call); + dbg("$ref->{from} swore: '@words' -> $ref->{to} '$ref->{subject}' origin: $ref->{origin} via " . $self->call) if isdbg('msg'); + Log('msg',"$ref->{from} swore: '@words' -> $ref->{to} origin: $ref->{origin} via " . $self->call); Log('msg',"subject: $ref->{subject}"); for (@{$ref->{lines}}) { Log('msg', "line: $_"); @@ -322,7 +342,7 @@ sub process if ($ref) { if ($ref->{private}) { # remove it if it private and gone off site# Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $fromnode and deleted"); - $ref->{delete}++; + $ref->mark_delete; } else { Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $fromnode"); push @{$ref->{gotit}}, $fromnode; # mark this up as being received @@ -388,7 +408,7 @@ sub process if ($pcno == 49) { # global delete on subject for (@msg) { if ($_->{from} eq $f[1] && $_->{subject} eq $f[2]) { - $_->{delete}++; + $_->mark_delete; Log('msg', "Message $_->{msgno} from $_->{from} ($_->{subject}) fully deleted"); DXChannel::broadcast_nodes($line, $self); } @@ -433,7 +453,9 @@ sub store if (defined $fh) { my $rr = $ref->{rrreq} ? '1' : '0'; my $priv = $ref->{private} ? '1': '0'; - print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{'read'}^$rr\n"; + my $del = $ref->{delete} ? '1' : '0'; + my $delt = $ref->{deletetime} || '0'; + print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{'read'}^$rr^$del^$delt\n"; print $fh "=== ", join('^', @{$ref->{gotit}}), "\n"; my $line; $ref->{size} = 0; @@ -458,41 +480,45 @@ sub del_msg if ($self->{tonode}) { $self->{delete}++; + $self->{deletetime} = 0; } else { - my $call; - if ($dxchan) { - $call = " by " . $dxchan->call; - } else { - $call = ''; - } - # remove it from the active message list @msg = grep { $_ != $self } @msg; + + my $call = ''; + $call = ' by ' . $dxchan->call if $dxchan; + Log('msg', "Msgno $self->{msgno} expunged$call"); # remove the file unlink filename($self->{msgno}); } } +sub mark_delete +{ + my $ref = shift; + my $t = shift; + $t = $main::systime + $residencetime unless defined $t; + + $ref->{delete}++; + $ref->{deletetime} = $t; + $ref->store( [$ref->read_msg_body] ); +} + # clean out old messages from the message queue sub clean_old { my $ref; # mark old messages for deletion - dbg("\@msg = " . scalar @msg . " before delete") if isdbg('msg'); foreach $ref (@msg) { if (ref($ref) && !$ref->{keep} && $ref->{t} < $main::systime - $maxage) { - $ref->{deleteme} = 1; - unlink filename($ref->{msgno}); - dbg("deleting old $ref->{msgno}\n") if isdbg('msg'); + + # this is for IMMEDIATE destruction + $ref->{delete}++; + $ref->{deletetime} = 0; } } - - # remove them all from the active message list - @msg = grep { !$_->{deleteme} } @msg; - dbg("\@msg = " . scalar @msg . " after delete") if isdbg('msg'); - $last_clean = $main::systime; } # read in a message header @@ -630,10 +656,11 @@ sub queue_msg next if $ref->{tonode}; # ignore it if it already being processed # is it awaiting deletion? - if ($ref->{delete}) { + if ($ref->{delete} && $main::systime >= $ref->{deletetime}) { $ref->del_msg; next; } + next if $ref->{delete}; # firstly, is it private and unread? if so can I find the recipient # in my cluster node list offsite? @@ -796,8 +823,9 @@ sub workclean delete $ref->{stream}; delete $ref->{file}; delete $ref->{count}; - delete $ref->{lastt} if exists $ref->{lastt}; - delete $ref->{waitt} if exists $ref->{waitt}; + delete $ref->{tempr}; + delete $ref->{lastt}; + delete $ref->{waitt}; } # get a new transaction number from the file specified @@ -931,7 +959,7 @@ sub do_send_stuff if (my @ans = BadWords::check($line)) { $self->{badcount} += @ans; Log('msg', $self->call . " used badwords: @ans to @{$loc->{to}} in msg"); - return ($self->msg('e17', @ans), $self->msg('m1')); + $loc->{reject}++; } $loc->{subject} = $line; $loc->{lines} = []; @@ -943,33 +971,36 @@ sub do_send_stuff my $loc = $self->{loc}; if ($line eq "\032" || $line eq '%1A' || uc $line eq "/EX") { my $to; - - foreach $to (@{$loc->{to}}) { - my $ref; - my $systime = $main::systime; - my $mycall = $main::mycall; - $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'), - uc $to, - exists $loc->{from} ? $loc->{from} : $self->call, - $systime, - $loc->{private}, - $loc->{subject}, - exists $loc->{origin} ? $loc->{origin} : $mycall, - '0', - $loc->{rrreq}); - $ref->swop_it($self->call); - $ref->store($loc->{lines}); - $ref->add_dir(); - push @out, $self->msg('m11', $ref->{msgno}, $to); - #push @out, "msgno $ref->{msgno} sent to $to"; - my $dxchan = DXChannel->get(uc $to); - if ($dxchan) { - if ($dxchan->is_user()) { - $dxchan->send($dxchan->msg('m9')); + unless ($loc->{reject}) { + foreach $to (@{$loc->{to}}) { + my $ref; + my $systime = $main::systime; + my $mycall = $main::mycall; + $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'), + uc $to, + exists $loc->{from} ? $loc->{from} : $self->call, + $systime, + $loc->{private}, + $loc->{subject}, + exists $loc->{origin} ? $loc->{origin} : $mycall, + '0', + $loc->{rrreq}); + $ref->swop_it($self->call); + $ref->store($loc->{lines}); + $ref->add_dir(); + push @out, $self->msg('m11', $ref->{msgno}, $to); + #push @out, "msgno $ref->{msgno} sent to $to"; + my $dxchan = DXChannel->get(uc $to); + if ($dxchan) { + if ($dxchan->is_user()) { + $dxchan->send($dxchan->msg('m9')); + } } } + } else { + Log('msg', $self->call . " swore to @{$loc->{to}} subject: '$loc->{subject}' in msg, REJECTED"); } - + delete $loc->{lines}; delete $loc->{to}; delete $self->{loc}; @@ -987,9 +1018,9 @@ sub do_send_stuff } else { if (my @ans = BadWords::check($line)) { $self->{badcount} += @ans; - Log('msg', $self->call . " used badwords: @ans to @{$loc->{to}} subject: '$loc->{subject}' in msg"); + Log('msg', $self->call . " used badwords: @ans to @{$loc->{to}} subject: '$loc->{subject}' in msg") unless $loc->{reject}; Log('msg', "line: $line"); - return ($self->msg('e17', @ans)); + $loc->{reject}++; } # i.e. it ain't and end or abort, therefore store the line @@ -1206,7 +1237,9 @@ sub import_one # first line; my $line = shift @$ref; - my @f = split /\s+/, $line; + my @f = split /([\s\@\$])/, $line; + @f = map {s/\s+//g; length $_ ? $_ : ()} @f; + unless (@f && $f[0] =~ /^(:?S|SP|SB|SEND)$/ ) { my $m = "invalid first line in import '$line'"; dbg($m) if isdbg('msg'); @@ -1223,16 +1256,16 @@ sub import_one ; } elsif ($notincalls && ($f eq 'RR')) { $rr = '1'; - } elsif ($f eq '@' && @f) { # this is bbs syntax, for origin - $origin = uc shift @f; + } elsif (($f =~ /^[\@\.\#\$]$/ || $f eq '.#') && @f) { # this is bbs syntax, for AT + shift @f; } elsif ($f eq '<' && @f) { # this is bbs syntax for from call $from = uc shift @f; } elsif ($f =~ /^\$/) { # this is bbs syntax for a bid next; - } elsif ($f =~ /^<\S+/) { # this is bbs syntax for from call - ($from) = $f =~ /^<(\S+)$/; - } elsif ($f =~ /^\@\S+/) { # this is bbs syntax for origin - ($origin) = $f =~ /^\@(\S+)$/; + } elsif ($f =~ /^<(\S+)/) { # this is bbs syntax for from call + $from = $1; + } elsif ($f =~ /^\$\S+/) { # this is bbs syntax for bid + ; } else { # callsign ? @@ -1292,7 +1325,18 @@ sub import_one } else { push @chunk, $ref; } - + + # does an identical message already exist? + my $m; + for $m (@msg) { + if (substr($subject,0,28) eq substr($m->{subject},0,28) && $from eq $m->{from} && grep $m->{to} eq $_, @to) { + my $msgno = $m->{msgno}; + dbg("duplicate message from $from -> $m->{to} to msg: $msgno") if isdbg('msg'); + Log('msg', "duplicate message from $from -> $m->{to} to msg: $msgno"); + return; + } + } + # write all the messages away my $i; for ( $i = 0; $i < @chunk; $i++) {