X-Git-Url: http://dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FDXMsg.pm;h=98c41ef1d7f935f899f2688a6aca1b06f1a1c195;hb=81e9887d5b7cee45d2b810ab0f8f19d21f048eab;hp=2028f8b1da8403498950c746c2c3ab54c2408140;hpb=30dbf70f84b53174005810f64f546d2181e1a8c6;p=spider.git diff --git a/perl/DXMsg.pm b/perl/DXMsg.pm index 2028f8b1..98c41ef1 100644 --- a/perl/DXMsg.pm +++ b/perl/DXMsg.pm @@ -43,8 +43,8 @@ $maxage = 30 * 86400; # the maximum age that a message shall live for if not m $last_clean = 0; # last time we did a clean @forward = (); # msg forward table $timeout = 30*60; # forwarding timeout -$waittime = 60*60; # time an aborted outgoing message waits before trying again -$queueinterval = 5*60; # run the queue every 5 minutes +$waittime = 30*60; # time an aborted outgoing message waits before trying again +$queueinterval = 1*60; # run the queue every 1 minute $lastq = 0; @@ -130,20 +130,22 @@ sub process # this is periodic processing if (!$self || !$line) { - # wander down the work queue stopping any messages that have timed out - for (keys %busy) { - my $node = $_; - my $ref = $busy{$_}; - if (exists $ref->{lastt} && $main::systime > $ref->{lastt} + $timeout) { - $ref->stop_msg($node); + if ($main::systime > $lastq + $queueinterval) { - # delay any outgoing messages that fail - $ref->{waitt} = $main::systime + $waittime + rand(120) if $node ne $main::mycall; + # wander down the work queue stopping any messages that have timed out + for (keys %busy) { + my $node = $_; + my $ref = $busy{$_}; + if (exists $ref->{lastt} && $main::systime >= $ref->{lastt} + $timeout) { + dbg('msg', "Timeout, stopping msgno: $ref->{msgno} -> $node"); + $ref->stop_msg($node); + + # delay any outgoing messages that fail + $ref->{waitt} = $main::systime + $waittime + rand(120) if $node ne $main::mycall; + } } - } - # queue some message if the interval timer has gone off - if ($main::systime > $lastq + $queueinterval) { + # queue some message if the interval timer has gone off queue_msg(0); $lastq = $main::systime; } @@ -165,6 +167,7 @@ sub process if (exists $busy{$f[2]}) { my $ref = $busy{$f[2]}; my $tonode = $ref->{tonode}; + dbg('msg', "Busy, stopping msgno: $ref->{msgno} -> $f[2]"); $ref->stop_msg($self->call); } @@ -184,6 +187,13 @@ sub process $busy{$f[2]} = $ref; # set interlock $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack $ref->{lastt} = $main::systime; + + # look to see whether this is a non private message sent to a known callsign + my $uref = DXUser->get_current($ref->{to}); + if (iscallsign($ref->{to}) && !$ref->{private} && $uref && $uref->homenode) { + $ref->{private} = 1; + dbg('msg', "set bull to $ref->{to} to private"); + } last SWITCH; } @@ -198,6 +208,9 @@ sub process $ref->{count} = 0; } $ref->{lastt} = $main::systime; + } else { + dbg('msg', "PC29 from unknown stream $f[3] from $f[2]" ); + $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } last SWITCH; } @@ -217,6 +230,7 @@ sub process $ref->send_tranche($self); $ref->{lastt} = $main::systime; } else { + dbg('msg', "PC30 from unknown stream $f[3] from $f[2]" ); $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } last SWITCH; @@ -229,6 +243,7 @@ sub process $ref->send_tranche($self); $ref->{lastt} = $main::systime; } else { + dbg('msg', "PC31 from unknown stream $f[3] from $f[2]" ); $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } last SWITCH; @@ -253,7 +268,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}) { + if ($ref->{subject} eq $m->{subject} && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from} && $ref->{to} eq $m->{to}) { $ref->stop_msg($self->call); my $msgno = $m->{msgno}; dbg('msg', "duplicate message to $msgno\n"); @@ -281,6 +296,7 @@ sub process } $ref->stop_msg($self->call); } else { + dbg('msg', "PC32 from unknown stream $f[3] from $f[2]" ); $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } # queue_msg(0); @@ -300,6 +316,7 @@ sub process } $ref->stop_msg($self->call); } else { + dbg('msg', "PC33 from unknown stream $f[3] from $f[2]" ); $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } @@ -337,6 +354,7 @@ sub process $ref->{stream} = $stream; $ref->{count} = 0; # no of lines between PC31s $ref->{file} = 1; + $ref->{lastt} = $main::systime; $work{"$f[2]$stream"} = $ref; # store in work $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack @@ -350,7 +368,6 @@ sub process $ref->stop_msg($self->call); $ref = undef; } - last SWITCH; } @@ -376,9 +393,9 @@ sub store my $lines = shift; # we only proceed if there are actually any lines in the file - if (!$lines || @{$lines} == 0) { - return; - } +# if (!$lines || @{$lines} == 0) { +# return; +# } if ($ref->{file}) { # a file dbg('msg', "To be stored in $ref->{to}\n"); @@ -558,7 +575,6 @@ sub queue_msg my $call = shift; my $ref; my $clref; - my $dxchan; my @nodelist = DXProt::get_all_ak1a(); # bat down the message list looking for one that needs to go off site and whose @@ -571,42 +587,41 @@ sub queue_msg # ignore 'delayed' messages until their waiting time has expired if (exists $ref->{waitt}) { - next if $ref->{waitt} < $main::systime; + next if $ref->{waitt} > $main::systime; delete $ref->{waitt}; } - + + # deal with routed private messages + my $noderef; if ($ref->{private}) { - if ($ref->{'read'} == 0) { - $clref = DXCluster->get_exact($ref->{to}); - unless ($clref) { # otherwise look for a homenode - my $uref = DXUser->get($ref->{to}); - my $hnode = $uref->homenode if $uref; - $clref = DXCluster->get_exact($hnode) if $hnode; - } - if ($clref && !grep { $clref->{dxchan} == $_ } DXCommandmode::get_all) { - $dxchan = $clref->{dxchan}; - $ref->start_msg($dxchan) if $dxchan && $clref && !get_busy($dxchan->call) && $dxchan->state eq 'normal'; - } + $clref = DXCluster->get_exact($ref->{to}); + unless ($clref) { # otherwise look for a homenode + my $uref = DXUser->get($ref->{to}); + my $hnode = $uref->homenode if $uref; + $clref = DXCluster->get_exact($hnode) if $hnode; } - } elsif (!$sort) { - # otherwise we are dealing with a bulletin, compare the gotit list with - # the nodelist up above, if there are sites that haven't got it yet - # then start sending it - what happens when we get loops is anyone's - # guess, use (to, from, time, subject) tuple? - my $noderef; - foreach $noderef (@nodelist) { - next if $noderef->call eq $main::mycall; - next if grep { $_ eq $noderef->call } @{$ref->{gotit}}; - next unless $ref->forward_it($noderef->call); # check the forwarding file - # next if $noderef->isolate; # maybe add code for stuff originated here? - # next if DXUser->get( ${$ref->{gotit}}[0] )->isolate; # is the origin isolated? - - # if we are here we have a node that doesn't have this message + if ($clref && !grep { $clref->{dxchan} == $_ } DXCommandmode::get_all) { + next if $clref->call eq $main::mycall; # i.e. it lives here + $noderef = $clref->{dxchan}; $ref->start_msg($noderef) if !get_busy($noderef->call) && $noderef->state eq 'normal'; - last; } } + # otherwise we are dealing with a bulletin or forwarded private message + # compare the gotit list with + # the nodelist up above, if there are sites that haven't got it yet + # then start sending it - what happens when we get loops is anyone's + # guess, use (to, from, time, subject) tuple? + foreach $noderef (@nodelist) { + next if $noderef->call eq $main::mycall; + next if grep { $_ eq $noderef->call } @{$ref->{gotit}}; + next unless $ref->forward_it($noderef->call); # check the forwarding file + + # if we are here we have a node that doesn't have this message + $ref->start_msg($noderef) if !get_busy($noderef->call) && $noderef->state eq 'normal'; + last; + } + # if all the available nodes are busy then stop last if @nodelist == scalar grep { get_busy($_->call) } @nodelist; } @@ -639,6 +654,7 @@ sub start_msg $self->{fromnode} = $main::mycall; $busy{$self->{tonode}} = $self; $work{$self->{tonode}} = $self; + $self->{lastt} = $main::systime; $dxchan->send(DXProt::pc28($self->{tonode}, $self->{fromnode}, $self->{to}, $self->{from}, $self->{t}, $self->{private}, $self->{subject}, $self->{origin}, $self->{rrreq})); } @@ -808,7 +824,7 @@ sub do_send_stuff } elsif ($self->state eq 'sendbody') { confess "local var gone missing" if !ref $self->{loc}; my $loc = $self->{loc}; - if ($line eq "\032" || uc $line eq "/EX") { + if ($line eq "\032" || $line eq '%1A' || uc $line eq "/EX") { my $to; if (@{$loc->{lines}} > 0) { @@ -873,8 +889,11 @@ sub dir sub load_forward { my @out; - do "$forwardfn" if -e "$forwardfn"; - push @out, $@ if $@; + my $s = readfilestr($forwardfn); + if ($s) { + eval $s; + push @out, $@ if $@; + } return @out; } @@ -882,8 +901,11 @@ sub load_forward sub load_badmsg { my @out; - do "$badmsgfn" if -e "$badmsgfn"; - push @out, $@ if $@; + my $s = readfilestr($badmsgfn); + if ($s) { + eval $s; + push @out, $@ if $@; + } return @out; } @@ -903,11 +925,13 @@ sub forward_it my $tested; # are we interested? - last if $ref->{private} && $sort ne 'P'; - last if !$ref->{private} && $sort ne 'B'; + next if $ref->{private} && $sort ne 'P'; + next if !$ref->{private} && $sort ne 'B'; # select field $tested = $ref->{to} if $field eq 'T'; + my $at = $ref->{to} =~ /\@\s*(\S+)/; + $tested = $at if $field eq '\@'; $tested = $ref->{from} if $field eq 'F'; $tested = $ref->{origin} if $field eq 'O'; $tested = $ref->{subject} if $field eq 'S';