X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FDXMsg.pm;h=3038b75638b12308fe18bd01baafd69ff6c2de7e;hb=179ce69d2479d950536efd127fe7b11666e4da6e;hp=c87fd2e4ec5b316433e9749f28aa525a0098cb8d;hpb=b91254375e95e7931312c4177ee390866f82c648;p=spider.git

diff --git a/perl/DXMsg.pm b/perl/DXMsg.pm
index c87fd2e4..3038b756 100644
--- a/perl/DXMsg.pm
+++ b/perl/DXMsg.pm
@@ -26,8 +26,19 @@ use DXLog;
 use IO::File;
 use Fcntl;
 
+eval {
+	require Mail::Send;
+};
+
 use strict;
-use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean
+
+use vars qw($VERSION $BRANCH);
+$VERSION = sprintf( "%d.%03d", q$Revision$ =~ /(\d+)\.(\d+)/ );
+$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 $residencetime
 			@badmsg @swop $swopfn $badmsgfn $forwardfn @forward $timeout $waittime
 		    $queueinterval $lastq $importfn $minchunk $maxchunk $bulltopriv);
 
@@ -48,6 +59,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
@@ -62,10 +74,10 @@ $importfn = "$msgdir/import";       # import directory
 		  to => '0,To',
 		  from => '0,From',
 		  t => '0,Msg Time,cldatetime',
-		  private => '5,Private',
+		  private => '5,Private,yesno',
 		  subject => '0,Subject',
 		  linesreq => '0,Lines per Gob',
-		  rrreq => '5,Read Confirm',
+		  rrreq => '5,Read Confirm,yesno',
 		  origin => '0,Origin',
 		  lines => '5,Data',
 		  stream => '9,Stream No',
@@ -79,6 +91,8 @@ $importfn = "$msgdir/import";       # import directory
 		  keep => '0,Keep this?,yesno',
 		  lastt => '5,Last processed,cldatetime',
 		  waitt => '5,Wait until,cldatetime',
+		  delete => '5,Awaiting Delete,yesno',
+		  deletetime => '5,Deletion Time,cldatetime',
 		 );
 
 # allocate a new object
@@ -100,6 +114,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} = [];
@@ -108,19 +124,6 @@ sub alloc
 	return $self;
 }
 
-sub workclean
-{
-	my $ref = shift;
-	delete $ref->{lines};
-	delete $ref->{linesreq};
-	delete $ref->{tonode};
-	delete $ref->{fromnode};
-	delete $ref->{stream};
-	delete $ref->{file};
-	delete $ref->{count};
-	delete $ref->{lastt} if exists $ref->{lastt};
-	delete $ref->{waitt} if exists $ref->{waitt};
-}
 
 sub process
 {
@@ -142,53 +145,49 @@ 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) || $pcno == 42;
 
  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];
-			}
+			my $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{$fromnode}) {
-				my $ref = $busy{$fromnode};
-				my $tonode = $ref->{tonode};
-				dbg("Busy, stopping msgno: $ref->{msgno} $fromnode->$tonode") if isdbg('msg');
-				$ref->stop_msg($self->call);
+			my $ref = get_busy($fromnode);
+			if ($ref) {
+				my $otonode = $ref->{tonode} || "unknown";
+				dbg("Busy, stopping msgno: $ref->{msgno} $fromnode->$otonode") if isdbg('msg');
+				$ref->stop_msg($fromnode);
 			}
 
 			my $t = cltounix($f[5], $f[6]);
-			my $stream = next_transno($fromnode);
-			my $ref = DXMsg->alloc($stream, uc $f[3], $f[4], $t, $f[7], $f[8], $origin, '0', $f[11]);
+			$stream = next_transno($fromnode);
+			$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} = $fromnode;
-			$ref->{tonode} = $f[1];
+			$ref->{tonode} = $tonode;
 			$ref->{rrreq} = $f[11];
 			$ref->{linesreq} = $f[10];
 			$ref->{stream} = $stream;
 			$ref->{count} = 0;	# no of lines between PC31s
 			dbg("new message from $f[4] to $f[3] '$f[8]' stream $fromnode/$stream\n") if isdbg('msg');
-			Log('msg', "Incoming message $f[4] to $f[3] '$f[8]'" );
-			$work{"$fromnode$stream"} = $ref; # store in work
-			$busy{$fromnode} = $ref; # set interlock
-			$self->send(DXProt::pc30($fromnode, $f[1], $stream)); # send ack
+			Log('msg', "Incoming message $f[4] to $f[3] '$f[8]' origin: $origin" );
+			set_fwq($fromnode, $stream, $ref); # store in work
+			set_busy($fromnode, $ref); # set interlock
+			$self->send(DXProt::pc30($fromnode, $tonode, $stream)); # send ack
 			$ref->{lastt} = $main::systime;
 
 			# look to see whether this is a non private message sent to a known callsign
@@ -196,67 +195,83 @@ sub process
 			if (is_callsign($ref->{to}) && !$ref->{private} && $uref && $uref->homenode) {
 				$ref->{private} = 1;
 				dbg("set bull to $ref->{to} to private") if isdbg('msg');
+				Log('msg', "set bull to $ref->{to} to private");
 			}
 			last SWITCH;
 		}
 		
 		if ($pcno == 29) {		# incoming text
-			my $ref = $work{"$f[2]$f[3]"};
+			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($f[2], $f[1], $f[3]));
-					dbg("stream $f[3]: $ref->{count} lines received\n") if isdbg('msg');
+					$self->send(DXProt::pc31($fromnode, $tonode, $stream));
+					dbg("stream $stream: $ref->{count} lines received\n") if isdbg('msg');
 					$ref->{count} = 0;
 				}
 				$ref->{lastt} = $main::systime;
 			} else {
-				dbg("PC29 from unknown stream $f[3] from $f[2]") if isdbg('msg');
-				$self->send(DXProt::pc42($f[2], $f[1], $f[3]));	# unknown stream
+				dbg("PC29 from unknown stream $stream from $fromnode") if isdbg('msg');
+				$self->send(DXProt::pc42($fromnode, $tonode, $stream));	# unknown stream
 			}
 			last SWITCH;
 		}
 		
 		if ($pcno == 30) {		# this is a incoming subject ack
-			my $ref = $work{$f[2]};	# note no stream at this stage
+			my $ref = get_fwq($fromnode);	# note no stream at this stage
 			if ($ref) {
-				delete $work{$f[2]};
-				$ref->{stream} = $f[3];
+				del_fwq($fromnode);
+				$ref->{stream} = $stream;
 				$ref->{count} = 0;
 				$ref->{linesreq} = 5;
-				$work{"$f[2]$f[3]"} = $ref;	# new ref
-				dbg("incoming subject ack stream $f[3]\n") if isdbg('msg');
-				$busy{$f[2]} = $ref; # interlock
-				push @{$ref->{lines}}, ($ref->read_msg_body);
+				set_fwq($fromnode, $stream, $ref);	# new ref
+				set_busy($fromnode, $ref); # interlock
+				dbg("incoming subject ack stream $stream\n") if isdbg('msg');
+				$ref->{lines} = [ $ref->read_msg_body ];
 				$ref->send_tranche($self);
 				$ref->{lastt} = $main::systime;
 			} else {
-				dbg("PC30 from unknown stream $f[3] from $f[2]") if isdbg('msg');
-				$self->send(DXProt::pc42($f[2], $f[1], $f[3]));	# unknown stream
+				dbg("PC30 from unknown stream $stream from $fromnode") if isdbg('msg');
+				$self->send(DXProt::pc42($fromnode, $tonode, $stream));	# unknown stream
 			} 
 			last SWITCH;
 		}
 		
 		if ($pcno == 31) {		# acknowledge a tranche of lines
-			my $ref = $work{"$f[2]$f[3]"};
+			my $ref = get_fwq($fromnode, $stream);
 			if ($ref) {
-				dbg("tranche ack stream $f[3]\n") if isdbg('msg');
+				dbg("tranche ack stream $stream\n") if isdbg('msg');
 				$ref->send_tranche($self);
 				$ref->{lastt} = $main::systime;
 			} else {
-				dbg("PC31 from unknown stream $f[3] from $f[2]") if isdbg('msg');
-				$self->send(DXProt::pc42($f[2], $f[1], $f[3]));	# unknown stream
+				dbg("PC31 from unknown stream $stream from $fromnode") if isdbg('msg');
+				$self->send(DXProt::pc42($fromnode, $tonode, $stream));	# unknown stream
 			} 
 			last SWITCH;
 		}
 		
 		if ($pcno == 32) {		# incoming EOM
-			dbg("stream $f[3]: EOM received\n") if isdbg('msg');
-			my $ref = $work{"$f[2]$f[3]"};
+			dbg("stream $stream: EOM received\n") if isdbg('msg');
+			my $ref = get_fwq($fromnode, $stream);
 			if ($ref) {
-				$self->send(DXProt::pc33($f[2], $f[1], $f[3]));	# acknowledge it
+				$self->send(DXProt::pc33($fromnode, $tonode, $stream));	# acknowledge it
 				
 				# get the next msg no - note that this has NOTHING to do with the stream number in PC protocol
 				# store the file or message
@@ -271,11 +286,11 @@ 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}) {
-								$ref->stop_msg($self->call);
+							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 $msgno") if isdbg('msg');
-								Log('msg', "duplicate message from $ref->{from} -> $ref->{to} to $msgno");
+								dbg("duplicate message from $ref->{from} -> $ref->{to} to msg: $msgno") if isdbg('msg');
+								Log('msg', "duplicate message from $ref->{from} -> $ref->{to} to msg: $msgno");
 								return;
 							}
 						}
@@ -285,45 +300,61 @@ sub process
 						
 						# look for 'bad' to addresses 
 						if ($ref->dump_it($self->call)) {
-							$ref->stop_msg($self->call);
+							$ref->stop_msg($fromnode);
 							dbg("'Bad' message $ref->{to}") if isdbg('msg');
 							Log('msg', "'Bad' message $ref->{to}");
 							return;
 						}
 
+						# check the message for bad words 
+						my @words;
+						for (@{$ref->{lines}}) {
+							push @words, BadWords::check($_);
+						}
+						push @words, BadWords::check($ref->{subject});
+						if (@words) {
+							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: $_");
+							}
+							$ref->stop_msg($fromnode);
+							return;
+						}
+							
 						$ref->{msgno} = next_transno("Msgno");
-						push @{$ref->{gotit}}, $f[2]; # mark this up as being received
+						push @{$ref->{gotit}}, $fromnode; # mark this up as being received
 						$ref->store($ref->{lines});
+						$ref->notify;
 						add_dir($ref);
-						my $dxchan = DXChannel->get($ref->{to});
-						$dxchan->send($dxchan->msg('m9')) if $dxchan && $dxchan->is_user;
-						Log('msg', "Message $ref->{msgno} from $ref->{from} received from $f[2] for $ref->{to}");
+						Log('msg', "Message $ref->{msgno} from $ref->{from} received from $fromnode for $ref->{to}");
 					}
 				}
-				$ref->stop_msg($self->call);
+				$ref->stop_msg($fromnode);
 			} else {
-				dbg("PC32 from unknown stream $f[3] from $f[2]") if isdbg('msg');
-				$self->send(DXProt::pc42($f[2], $f[1], $f[3]));	# unknown stream
+				dbg("PC32 from unknown stream $stream from $fromnode") if isdbg('msg');
+				$self->send(DXProt::pc42($fromnode, $tonode, $stream));	# unknown stream
 			}
 			# queue_msg(0);
 			last SWITCH;
 		}
 		
 		if ($pcno == 33) {		# acknowledge the end of message
-			my $ref = $work{"$f[2]$f[3]"};
+			my $ref = get_fwq($fromnode, $stream);
 			if ($ref) {
 				if ($ref->{private}) { # remove it if it private and gone off site#
-					Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2] and deleted");
-					$ref->del_msg;
+					Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $fromnode and deleted");
+					$ref->mark_delete;
 				} else {
-					Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $f[2]");
-					push @{$ref->{gotit}}, $f[2]; # mark this up as being received
+					Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $fromnode");
+					push @{$ref->{gotit}}, $fromnode; # mark this up as being received
 					$ref->store($ref->{lines});	# re- store the file
 				}
-				$ref->stop_msg($self->call);
+				$ref->stop_msg($fromnode);
 			} else {
-				dbg("PC33 from unknown stream $f[3] from $f[2]") if isdbg('msg');
-				$self->send(DXProt::pc42($f[2], $f[1], $f[3]));	# unknown stream
+				dbg("PC33 from unknown stream $stream from $fromnode") if isdbg('msg');
+				$self->send(DXProt::pc42($fromnode, $tonode, $stream));	# unknown stream
 			} 
 
 			# send next one if present
@@ -350,28 +381,28 @@ sub process
 				last SWITCH if !mkdir $fn, 0777;
 				dbg("created directory $fn\n") if isdbg('msg');
 			}
-			my $stream = next_transno($f[2]);
+			my $stream = next_transno($fromnode);
 			my $ref = DXMsg->alloc($stream, "$main::root/$f[3]", $self->call, time, !$f[4], $f[3], ' ', '0', '0');
 			
 			# forwarding variables
-			$ref->{fromnode} = $f[1];
-			$ref->{tonode} = $f[2];
+			$ref->{fromnode} = $tonode;
+			$ref->{tonode} = $fromnode;
 			$ref->{linesreq} = $f[5];
 			$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 
+			set_fwq($fromnode, $stream, $ref); # store in work
+			$self->send(DXProt::pc30($fromnode, $tonode, $stream)); # send ack 
 			
 			last SWITCH;
 		}
 		
 		if ($pcno == 42) {		# abort transfer
-			dbg("stream $f[3]: abort received\n") if isdbg('msg');
-			my $ref = $work{"$f[2]$f[3]"};
+			dbg("stream $stream: abort received\n") if isdbg('msg');
+			my $ref = get_fwq($fromnode, $stream);
 			if ($ref) {
-				$ref->stop_msg($self->call);
+				$ref->stop_msg($fromnode);
 				$ref = undef;
 			}
 			last SWITCH;
@@ -380,9 +411,9 @@ sub process
 		if ($pcno == 49) {      # global delete on subject
 			for (@msg) {
 				if ($_->{from} eq $f[1] && $_->{subject} eq $f[2]) {
-					$_->del_msg();
+					$_->mark_delete;
 					Log('msg', "Message $_->{msgno} from $_->{from} ($_->{subject}) fully deleted");
-					DXProt::broadcast_ak1a($line, $self);
+					DXChannel::broadcast_nodes($line, $self);
 				}
 			}
 		}
@@ -390,6 +421,34 @@ sub process
 }
 
 
+sub notify
+{
+	my $ref = shift;
+	my $to = $ref->{to};
+	my $uref = DXUser->get($to);
+	my $dxchan = DXChannel->get($to);
+	if (*Mail::Send && $uref && $uref->wantemail) {
+		my $email = $uref->email;
+		if ($email) {
+			my @list = ref $email ? @{$email} : $email;
+			my $msg = new Mail::Send Subject=>"[DXSpider: $ref->{from}] $ref->{subject}";
+			$msg->to(@list);
+			my $fh = $msg->open;
+			if ($fh) {
+				print $fh "From: $ref->{from} To: $to On Node: $main::mycall Origin: $ref->{origin} Msgno: $ref->{msgno}\r\n\r\n";
+				print $fh map {"$_\r\n"} $ref->read_msg_body;
+				$fh->close;
+				for (@list) {
+					Log('msg', "Msgno $ref->{msgno} from $ref->{from} emailed to $_");
+				}
+			} else {
+				dbg("email forwarding error $!") if isdbg('msg'); 
+			}
+		}
+	}
+	$dxchan->send($dxchan->msg('m9')) if $dxchan && $dxchan->is_user;
+}
+
 # store a message away on disc or whatever
 #
 # NOTE the second arg is a REFERENCE not a list
@@ -425,7 +484,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;
@@ -446,15 +507,33 @@ sub store
 sub del_msg
 {
 	my $self = shift;
+	my $dxchan = shift;
 	
-	# remove it from the active message list
-	dbg("\@msg = " . scalar @msg . " before delete") if isdbg('msg');
-	@msg = grep { $_ != $self } @msg;
+	if ($self->{tonode}) {
+		$self->{delete}++;
+		$self->{deletetime} = 0;
+	} else {
+		# 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;
 	
-	# remove the file
-	unlink filename($self->{msgno});
-	dbg("deleting $self->{msgno}\n") if isdbg('msg');
-	dbg("\@msg = " . scalar @msg . " after delete") if isdbg('msg');
+	$ref->{delete}++;
+	$ref->{deletetime} = $t;
+	$ref->store( [$ref->read_msg_body] );
 }
 
 # clean out old messages from the message queue
@@ -463,19 +542,14 @@ 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
@@ -545,7 +619,7 @@ sub read_msg_body
 	if (!open($file, $fn)) {
 		dbg("Error reading $fn $!");
 		Log('err' ,"Error reading $fn $!");
-		return undef;
+		return ();
 	}
 	@out = map {chomp; $_} <$file>;
 	close($file);
@@ -580,7 +654,6 @@ sub send_tranche
 sub queue_msg
 {
 	my $sort = shift;
-	my $call = shift;
 	my $ref;
 	my $clref;
 	
@@ -610,6 +683,16 @@ sub queue_msg
 			next;
 		}
 
+		# is it being sent anywhere currently?
+		next if $ref->{tonode};	          # ignore it if it already being processed
+		
+		# is it awaiting deletion?
+		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?
 
@@ -618,11 +701,6 @@ sub queue_msg
 		if ($ref->{private}) {
 			next if $ref->{'read'};           # if it is read, it is stuck here
 			$clref = Route::get($ref->{to});
-#			unless ($clref) {             # otherwise look for a homenode
-#				my $uref = DXUser->get_current($ref->{to});
-#				my $hnode =  $uref->homenode if $uref;
-#				$clref = Route::Node::get($hnode) if $hnode;
-#			}
 			if ($clref) {
 				$dxchan = $clref->dxchan;
 				if ($dxchan) {
@@ -634,23 +712,27 @@ sub queue_msg
 					dbg("Route: No dxchan for $ref->{to} " . ref($clref) ) if isdbg('msg');
 				}
 			}
-		}
-		
-		# 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 $dxchan (@nodelist) {
-			my $call = $dxchan->call;
-			next unless $call;
-			next if $call eq $main::mycall;
-			next if ref $ref->{gotit} && grep $_ eq $call, @{$ref->{gotit}};
-			next unless $ref->forward_it($call);           # check the forwarding file
-
-			# if we are here we have a node that doesn't have this message
-			$ref->start_msg($dxchan) if !get_busy($call)  && $dxchan->state eq 'normal';
-			last;
+		} else {
+			
+			# 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 $dxchan (@nodelist) {
+				my $call = $dxchan->call;
+				next unless $call;
+				next if $call eq $main::mycall;
+				next if ref $ref->{gotit} && grep $_ eq $call, @{$ref->{gotit}};
+				next unless $ref->forward_it($call);           # check the forwarding file
+				next if $ref->{tonode};	          # ignore it if it already being processed
+				
+				# if we are here we have a node that doesn't have this message
+				if (!get_busy($call)  && $dxchan->state eq 'normal') {
+					$ref->start_msg($dxchan);
+					last;
+				}
+			}
 		}
 
 		# if all the available nodes are busy then stop
@@ -678,22 +760,18 @@ sub start_msg
 {
 	my ($self, $dxchan) = @_;
 	
+	confess("trying to start started msg $self->{msgno} nodes: $self->{fromnode} -> $self->{tonode}") if $self->{tonode};
 	dbg("start msg $self->{msgno}\n") if isdbg('msg');
 	$self->{linesreq} = 10;
 	$self->{count} = 0;
 	$self->{tonode} = $dxchan->call;
 	$self->{fromnode} = $main::mycall;
-	$busy{$self->{tonode}} = $self;
-	$work{$self->{tonode}} = $self;
+	set_busy($self->{tonode}, $self);
+	set_fwq($self->{tonode}, undef, $self);
 	$self->{lastt} = $main::systime;
 	my ($fromnode, $origin);
-	if ($dxchan->is_arcluster) {
-		$fromnode = $self->{origin};
-		$origin = $self->{fromnode};
-	} else {
-		$fromnode = $self->{fromnode};
-		$origin = $self->{origin};
-	}
+	$fromnode = $self->{fromnode};
+	$origin = $self->{origin};
 	$dxchan->send(DXProt::pc28($self->{tonode}, $fromnode, $self->{to}, $self->{from}, $self->{t}, $self->{private}, $self->{subject}, $origin, $self->{rrreq}));
 }
 
@@ -704,16 +782,52 @@ sub get_busy
 	return $busy{$call};
 }
 
-# get the busy queue
+sub set_busy
+{
+	my $call = shift;
+	return $busy{$call} = shift;
+}
+
+sub del_busy
+{
+	my $call = shift;
+	return delete $busy{$call};
+}
+
+# get the whole busy queue
 sub get_all_busy
 {
-	return values %busy;
+	return keys %busy;
 }
 
-# get the forwarding queue
+# get a forwarding queue entry
 sub get_fwq
 {
-	return values %work;
+	my $call = shift;
+	my $stream = shift || '0';
+	return $work{"$call,$stream"};
+}
+
+# delete a forwarding queue entry
+sub del_fwq
+{
+	my $call = shift;
+	my $stream = shift || '0';
+	return delete $work{"$call,$stream"};
+}
+
+# set a fwq entry
+sub set_fwq
+{
+	my $call = shift;
+	my $stream = shift || '0';
+	return $work{"$call,$stream"} = shift;
+}
+
+# get the whole forwarding queue
+sub get_all_fwq
+{
+	return keys %work;
 }
 
 # stop a message from continuing, clean it out, unlock interlocks etc
@@ -721,14 +835,28 @@ sub stop_msg
 {
 	my $self = shift;
 	my $node = shift;
-	my $stream = $self->{stream} if exists $self->{stream};
+	my $stream = $self->{stream};
 	
 	
 	dbg("stop msg $self->{msgno} -> node $node\n") if isdbg('msg');
-	delete $work{$node};
-	delete $work{"$node$stream"} if $stream;
+	del_fwq($node, $stream);
 	$self->workclean;
-	delete $busy{$node};
+	del_busy($node);
+}
+
+sub workclean
+{
+	my $ref = shift;
+	delete $ref->{lines};
+	delete $ref->{linesreq};
+	delete $ref->{tonode};
+	delete $ref->{fromnode};
+	delete $ref->{stream};
+	delete $ref->{file};
+	delete $ref->{count};
+	delete $ref->{tempr};
+	delete $ref->{lastt};
+	delete $ref->{waitt};
 }
 
 # get a new transaction number from the file specified
@@ -859,6 +987,11 @@ sub do_send_stuff
 		#  $DB::single = 1;
 		confess "local var gone missing" if !ref $self->{loc};
 		my $loc = $self->{loc};
+		if (my @ans = BadWords::check($line)) {
+			$self->{badcount} += @ans;
+			Log('msg', $self->call . " used badwords: @ans to @{$loc->{to}} in msg");
+			$loc->{reject}++;
+		}
 		$loc->{subject} = $line;
 		$loc->{lines} = [];
 		$self->state('sendbody');
@@ -869,33 +1002,31 @@ 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";
+					$ref->notify;
 				}
+			} else {
+				Log('msg', $self->call . " swore to @{$loc->{to}} subject: '$loc->{subject}' in msg, REJECTED");
 			}
-
+			
 			delete $loc->{lines};
 			delete $loc->{to};
 			delete $self->{loc};
@@ -911,20 +1042,28 @@ sub do_send_stuff
 			$self->func(undef);
 			$self->state('prompt');
 		} 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") unless $loc->{reject};
+				Log('msg', "line: $line");
+				$loc->{reject}++;
+			}
 			
 			# i.e. it ain't and end or abort, therefore store the line
 			push @{$loc->{lines}}, length($line) > 0 ? $line : " ";
 		}
 	}
-	return (1, @out);
+	return @out;
 }
 
 # return the standard directory line for this ref 
 sub dir
 {
 	my $ref = shift;
+	my $flag = $ref->read ? '-' : ' ';
+	$flag = 'D' if $ref->delete;
 	return sprintf "%6d%s%s%5d %8.8s %8.8s %-6.6s %5.5s %-30.30s", 
-		$ref->msgno, $ref->read ? '-' : ' ', $ref->private ? 'p' : ' ', $ref->size,
+		$ref->msgno, $flag, $ref->private ? 'p' : ' ', $ref->size,
 			$ref->to, $ref->from, cldate($ref->t), ztime($ref->t), $ref->subject;
 }
 
@@ -1102,7 +1241,7 @@ sub import_msgs
 		my @msg = map { chomp; $_ } <MSG>;
 		close(MSG);
 		unlink($fn);
-		my @out = import_one($DXProt::me, \@msg, $splitit);
+		my @out = import_one($main::me, \@msg, $splitit);
 		Log('msg', @out);
 	}
 }
@@ -1124,7 +1263,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');
@@ -1141,16 +1282,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 ?
@@ -1210,7 +1351,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++) {
@@ -1240,12 +1392,7 @@ sub import_one
 			$mref->add_dir();
 			push @out, $dxchan->msg('m11', $mref->{msgno}, $to);
 			#push @out, "msgno $ref->{msgno} sent to $to";
-			my $todxchan = DXChannel->get(uc $to);
-			if ($todxchan) {
-				if ($todxchan->is_user()) {
-					$todxchan->send($todxchan->msg('m9'));
-				}
-			}
+			$mref->notify;
 		}
 	}
 	return @out;