X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FDXMsg.pm;h=1bbe00ec0c8262acddbb285c200c4e81c584e0ec;hb=abd3cfb34b33d2ba1ad5bc582dfe960dbc47912c;hp=c91049ab19de50f9fb3b871d5a93520a43e1085d;hpb=0a92a4608d715cd9cd7eec3c120da651fbb4da5b;p=spider.git

diff --git a/perl/DXMsg.pm b/perl/DXMsg.pm
index c91049ab..1bbe00ec 100644
--- a/perl/DXMsg.pm
+++ b/perl/DXMsg.pm
@@ -15,8 +15,6 @@
 
 package DXMsg;
 
-@ISA = qw(DXProt DXChannel);
-
 use DXUtil;
 use DXChannel;
 use DXUser;
@@ -32,7 +30,7 @@ use Fcntl;
 use strict;
 use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean
 			@badmsg @swop $swopfn $badmsgfn $forwardfn @forward $timeout $waittime
-		    $queueinterval $lastq $importfn $minchunk $maxchunk);
+		    $queueinterval $lastq $importfn $minchunk $maxchunk $bulltopriv);
 
 %work = ();						# outstanding jobs
 @msg = ();						# messages we have
@@ -50,6 +48,8 @@ $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
+
 
 $badmsgfn = "$msgdir/badmsg.pl";    # list of TO address we wont store
 $forwardfn = "$msgdir/forward.pl";  # the forwarding table
@@ -102,8 +102,9 @@ sub alloc
 	$self->{'read'} = shift;
 	$self->{rrreq} = shift;
 	$self->{gotit} = [];
-	$self->{lastt} = $main::systime;
+#	$self->{lastt} = $main::systime;
 	$self->{lines} = [];
+	$self->{private} = 1 if $bulltopriv && DXUser->get_current($self->{to});
     
 	return $self;
 }
@@ -131,20 +132,6 @@ sub process
 
 		if ($main::systime >= $lastq + $queueinterval) {
 
-			# 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");
-					Log('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
 			queue_msg(0);
 
@@ -165,23 +152,33 @@ sub process
  SWITCH: {
 		if ($pcno == 28) {		# incoming message
 
+			# sort out various extant protocol errors that occur
+			my ($fromnode, $origin);
+			if ($self->is_arcluster && $f[13] eq $self->call) {
+				$fromnode = $f[13];
+				$origin = $f[2];
+			} else {
+				$fromnode = $f[2];
+			    $origin = $f[13];
+			}
+			$origin = $self->call unless $origin && $origin gt ' ';
+
 			# first look for any messages in the busy queue 
 			# and cancel them this should both resolve timed out incoming messages
 			# and crossing of message between nodes, incoming messages have priority
-			if (exists $busy{$f[2]}) {
-				my $ref = $busy{$f[2]};
+			if (exists $busy{$fromnode}) {
+				my $ref = $busy{$fromnode};
 				my $tonode = $ref->{tonode};
-				dbg('msg', "Busy, stopping msgno: $ref->{msgno} -> $f[2]");
+				dbg('msg', "Busy, stopping msgno: $ref->{msgno} -> $fromnode");
 				$ref->stop_msg($self->call);
 			}
 
 			my $t = cltounix($f[5], $f[6]);
-			my $stream = next_transno($f[2]);
-			$f[13] = $self->call unless $f[13] && $f[13] gt ' ';
-			my $ref = DXMsg->alloc($stream, uc $f[3], $f[4], $t, $f[7], $f[8], $f[13], '0', $f[11]);
+			my $stream = next_transno($fromnode);
+			my $ref = DXMsg->alloc($stream, uc $f[3], $f[4], $t, $f[7], $f[8], $origin, '0', $f[11]);
 			
 			# fill in various forwarding state variables
-			$ref->{fromnode} = $f[2];
+			$ref->{fromnode} = $fromnode;
 			$ref->{tonode} = $f[1];
 			$ref->{rrreq} = $f[11];
 			$ref->{linesreq} = $f[10];
@@ -189,9 +186,9 @@ sub process
 			$ref->{count} = 0;	# no of lines between PC31s
 			dbg('msg', "new message from $f[4] to $f[3] '$f[8]' stream $stream\n");
 			Log('msg', "Incoming message $f[4] to $f[3] '$f[8]'" );
-			$work{"$f[2]$stream"} = $ref; # store in work
-			$busy{$f[2]} = $ref; # set interlock
-			$self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack
+			$work{"$fromnode$stream"} = $ref; # store in work
+			$busy{$fromnode} = $ref; # set interlock
+			$self->send(DXProt::pc30($fromnode, $f[1], $stream)); # send ack
 			$ref->{lastt} = $main::systime;
 
 			# look to see whether this is a non private message sent to a known callsign
@@ -593,8 +590,6 @@ sub queue_msg
 	dbg('msg', "queue msg ($sort)\n");
 	my @nodelist = DXChannel::get_all_nodes;
 	foreach $ref (@msg) {
-		# firstly, is it private and unread? if so can I find the recipient
-		# in my cluster node list offsite?
 
 		# ignore 'delayed' messages until their waiting time has expired
 		if (exists $ref->{waitt}) {
@@ -602,6 +597,22 @@ sub queue_msg
 			delete $ref->{waitt};
 		} 
 
+		# any time outs?
+		if (exists $ref->{lastt} && $main::systime >= $ref->{lastt} + $timeout) {
+			my $node = $ref->{tonode};
+			dbg('msg', "Timeout, stopping msgno: $ref->{msgno} -> $node");
+			Log('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;
+			delete $ref->{lastt};
+			next;
+		}
+
+		# firstly, is it private and unread? if so can I find the recipient
+		# in my cluster node list offsite?
+
 		# deal with routed private messages
 		my $dxchan;
 		if ($ref->{private}) {