+#!/usr/bin/perl
+#
+# This module impliments the message handling for a dx cluster
+#
+# Copyright (c) 1998 Dirk Koopman G1TLH
+#
+# $Id$
+#
+
+package DXMsg;
+
+@ISA = qw(DXProt DXChannel);
+
+use DXUtil;
+use DXChannel;
+use DXUser;
+use DXM;
+use DXCluster;
+use DXProtVars;
+use DXProtout;
+use DXDebug;
+use FileHandle;
+use Carp;
+
+use strict;
+use vars qw($stream %work @msg $msgdir $msgnofn);
+
+%work = (); # outstanding jobs
+@msg = (); # messages we have
+$msgdir = "$main::data/msg"; # directory contain the msgs
+
+# allocate a new object
+# called fromnode, tonode, from, to, datetime, private?, subject, nolinesper
+sub alloc
+{
+ my $pkg = shift;
+ my $self = bless {}, $pkg;
+ $self->{fromnode} = shift;
+ $self->{tonode} = shift;
+ $self->{to} = shift;
+ $self->{from} = shift;
+ $self->{t} = shift;
+ $self->{private} = shift;
+ $self->{subject} = shift;
+ $self->{linesreq} = shift; # this the number of lines to send or receive between PC31s
+ $self->{rrreq} = shift; # a read receipt is required
+ $self->{origin} = shift;
+ $self->{stream} = shift;
+ $self->{lines} = [];
+
+ return $self;
+}
+
+sub workclean
+{
+ my $ref = shift;
+ delete $ref->{lines};
+ delete $ref->{linesreq};
+ delete $ref->{tonode};
+ delete $ref->{stream};
+}
+
+sub process
+{
+ my ($self, $line) = @_;
+ my @f = split /[\^\~]/, $line;
+ my ($pcno) = $f[0] =~ /^PC(\d\d)/; # just get the number
+
+ SWITCH: {
+ if ($pcno == 28) { # incoming message
+ my $t = cltounix($f[5], $f[6]);
+ my $stream = next_transno($f[2]);
+ my $ref = DXMsg->alloc($f[1], $f[2], $f[3], $f[4], $t, $f[7], $f[8], $f[10], $f[11], $f[13], $stream);
+ dbg('msg', "new message from $f[4] to $f[3] '$f[8]' stream $stream\n");
+ $work{"$f[1]$f[2]$stream"} = $ref; # store in work
+ $self->send(DXProt::pc30($f[2], $f[1], $stream));
+ $ref->{count} = 0; # no of lines between PC31s
+ last SWITCH;
+ }
+
+ if ($pcno == 29) { # incoming text
+ my $ref = $work{"$f[1]$f[2]$f[3]"};
+ if ($ref) {
+ push @{$ref->{lines}}, $f[4];
+ $ref->{count}++;
+ if ($ref->{count} >= $ref->{linesreq}) {
+ $self->send(DXProt::pc31($f[2], $f[1], $f[3]));
+ dbg('msg', "stream $f[3]: $ref->{linereq} lines received\n");
+ $ref->{count} = 0;
+ }
+ }
+ last SWITCH;
+ }
+
+ if ($pcno == 30) {
+ last SWITCH;
+ }
+
+ if ($pcno == 31) {
+ last SWITCH;
+ }
+
+ if ($pcno == 32) { # incoming EOM
+ dbg('msg', "stream $f[3]: EOM received\n");
+ my $ref = $work{"$f[1]$f[2]$f[3]"};
+ if ($ref) {
+ $self->send(DXProt::pc33($f[2], $f[1], $f[3]));# acknowledge it
+ $ref->store(); # store it (whatever that may mean)
+ delete $work{"$f[1]$f[2]$f[3]"}; # remove the reference from the work vector
+ }
+ last SWITCH;
+ }
+
+ if ($pcno == 33) {
+ last SWITCH;
+ }
+
+ if ($pcno == 40) { # this is a file request
+ $f[3] =~ s/\\/\//og; # change the slashes
+ $f[3] =~ s/\.//og; # remove dots
+ $f[3] = lc $f[3]; # to lower case;
+ dbg('msg', "incoming file $f[3]\n");
+ last SWITCH if $f[3] =~ /^\/(perl|cmd|local_cmd|src|lib|include|sys|data\/msg)\//; # prevent access to executables
+
+ # create any directories
+ my @part = split /\//, $f[3];
+ my $part;
+ my $fn = "$main::root";
+ pop @part; # remove last part
+ foreach $part (@part) {
+ $fn .= "/$part";
+ next if -e $fn;
+ last SWITCH if !mkdir $fn, 0777;
+ dbg('msg', "created directory $fn\n");
+ }
+ my $stream = next_transno($f[2]);
+ my $ref = DXMsg->alloc($f[1], $f[2], "$main::root/$f[3]", undef, time, !$f[4], undef, $f[5], 0, ' ', $stream);
+ $ref->{file} = 1;
+ $work{"$f[1]$f[2]$stream"} = $ref; # store in work
+ $self->send(DXProt::pc30($f[2], $f[1], $stream));
+ $ref->{count} = 0; # no of lines between PC31s
+
+ last SWITCH;
+ }
+ }
+}
+
+
+# store a message away on disc or whatever
+sub store
+{
+ my $ref = shift;
+
+ # we only proceed if there are actually any lines in the file
+ if (@{$ref->{lines}} == 0) {
+ delete $ref->{lines};
+ return;
+ }
+
+ if ($ref->{file}) { # a file
+ dbg('msg', "To be stored in $ref->{to}\n");
+
+ my $fh = new FileHandle "$ref->{to}", "w";
+ if (defined $fh) {
+ my $line;
+ foreach $line (@{$ref->{lines}}) {
+ print $fh "$line\n";
+ }
+ $fh->close;
+ dbg('msg', "file $ref->{to} stored\n");
+ } else {
+ confess "can't open file $ref->{to} $!";
+ }
+ } else { # a normal message
+
+ # get the next msg no - note that this has NOTHING to do with the stream number in PC protocol
+ my $msgno = next_transno("msgno");
+
+ # attempt to open the message file
+ my $fn = sprintf "$msgdir/m%06d", $msgno;
+
+ dbg('msg', "To be stored in $fn\n");
+
+ my $fh = new FileHandle "$fn", "w";
+ if (defined $fh) {
+ print $fh "=== $ref->{to}^$ref->{from}^$ref->{private}^$ref->{subject}^$ref->{origin}\n";
+ print $fh "=== $ref->{fromnode}\n";
+ my $line;
+ foreach $line (@{$ref->{lines}}) {
+ $ref->{size} += length $line + 1;
+ print $fh "$line\n";
+ }
+ $ref->workclean();
+ push @msg, $ref; # add this message to the incore message list
+ $fh->close;
+ dbg('msg', "msg $msgno stored\n");
+ } else {
+ confess "can't open msg file $fn $!";
+ }
+ }
+}
+
+# get a new transaction number from the file specified
+sub next_transno
+{
+ my $name = shift;
+ $name =~ s/\W//og; # remove non-word characters
+ my $fn = "$msgdir/$name";
+ my $msgno;
+
+ my $fh = new FileHandle;
+ if (sysopen($fh, $fn, O_RDWR|O_CREAT, 0666)) {
+ $fh->autoflush(1);
+ $msgno = $fh->getline;
+ chomp $msgno;
+ $msgno++;
+ seek $fh, 0, 0;
+ $fh->print("$msgno\n");
+ dbg('msg', "msgno $msgno allocated for $name\n");
+ $fh->close;
+ } else {
+ confess "can't open $fn $!";
+ }
+ return $msgno;
+}
+
+# initialise the message 'system'
+sub init
+{
+
+}
+
+1;
+
+__END__
use DXProtVars;
use DXCommandmode;
use Spot;
-use Date::Parse;
use DXProtout;
use strict;
last SWITCH if grep $field[7] =~ /^$_/, @DXProt::nodx_node;
# convert the date to a unix date
- my $date = $field[3];
- my $time = $field[4];
- $date =~ s/^\s*(\d+)-(\w\w\w)-(19\d\d)$/$1 $2 $3/;
- $time =~ s/^(\d\d)(\d\d)Z$/$1:$2 +0000/;
- my $d = str2time("$date $time");
+ my $d = cltounix($field[3], $field[4]);
+# my $date = $field[3];
+# my $time = $field[4];
+# $date =~ s/^\s*(\d+)-(\w\w\w)-(19\d\d)$/$1 $2 $3/;
+# $time =~ s/^(\d\d)(\d\d)Z$/$1:$2 +0000/;
+# my $d = str2time("$date $time");
return if !$d; # bang out (and don't pass on) if date is invalid
# strip off the leading & trailing spaces from the comment
if ($pcno == 25) {last SWITCH;}
if ($pcno == 26) {last SWITCH;}
if ($pcno == 27) {last SWITCH;}
- if ($pcno == 28) {last SWITCH;}
- if ($pcno == 29) {last SWITCH;}
- if ($pcno == 30) {last SWITCH;}
- if ($pcno == 31) {last SWITCH;}
- if ($pcno == 32) {last SWITCH;}
- if ($pcno == 33) {last SWITCH;}
- if ($pcno == 34) {last SWITCH;}
- if ($pcno == 35) {last SWITCH;}
- if ($pcno == 36) {last SWITCH;}
+
+ if (($pcno >= 28 && $pcno <= 33) || $pcno == 40 || $pcno == 42) { # mail/file handling
+ DXMsg::process($self, $line);
+ return;
+ }
+
+ if ($pcno == 34 || $pcno == 36) { # remote commands (incoming)
+ last SWITCH;
+ }
+
+ if ($pcno == 35) { # remote command replies
+ last SWITCH;
+ }
+
if ($pcno == 37) {last SWITCH;}
- if ($pcno == 38) {last SWITCH;}
+
+ if ($pcno == 38) { # node connected list from neighbour
+ return;
+ }
if ($pcno == 39) { # incoming disconnect
$self->disconnect();
return;
}
- if ($pcno == 40) {last SWITCH;}
if ($pcno == 41) { # user info
# add this station to the user database, if required
my $user = DXUser->get_current($field[1]);
$user->put;
last SWITCH;
}
- if ($pcno == 42) {last SWITCH;}
if ($pcno == 43) {last SWITCH;}
if ($pcno == 44) {last SWITCH;}
if ($pcno == 45) {last SWITCH;}
my $ref = DXCluster->get($self->call);
# broadcast to all other nodes that all the nodes connected to via me are gone
- my @nodes = map { $_->dxchan == $self ? $_ : () } DXNode::get_all();
+ my @gonenodes = map { $_->dxchan == $self ? $_ : () } DXNode::get_all();
my $node;
- foreach $node (@nodes) {
+ foreach $node (@gonenodes) {
next if $node->call eq $self->call;
- broadcast_ak1a(DXProt::pc21($node, 'Gone'), $self); # done like this 'cos DXNodes don't have a pc21 method
+ broadcast_ak1a(pc21($node->call, 'Gone'), $self); # done like this 'cos DXNodes don't have a pc21 method
+ $node->del();
}
# now broadcast to all other ak1a nodes that I have gone
- broadcast_ak1a($self->pc21('Gone.'), $self);
+ broadcast_ak1a(pc21($self->call, 'Gone.'), $self);
$ref->del() if $ref;
}
# delete a node
sub pc21
{
- my ($ref, $reason) = @_;
- my $call = $ref->call;
+ my ($call, $reason) = @_;
my $hops = get_hops(21);
$reason = "Gone." if !$reason;
return "PC21^$call^$reason^$hops^";
return "PC24^$call^$flag^$hops^";
}
+# message start (fromnode, tonode, to, from, t, private, subject, origin)
+sub pc28
+{
+ my ($fromnode, $tonode, $to, $from, $t, $private, $subject, $origin) = @_;
+ my $date = cldate($t);
+ my $time = ztime($t);
+ $private = $private ? '1' : '0';
+ return "PC28^$fromnode^$tonode^$to^from^$date^$time^$private^$subject^ ^5^0^ ^$origin^~";
+}
+
+# message text (from and to node same way round as pc29)
+sub pc29
+{
+ my ($fromnode, $tonode, $stream, $text) = @_;
+ $text =~ s/\^//og; # remove ^
+ return "PC29^$fromnode^$tonode^$stream^text^~";
+}
+
+# subject acknowledge (will have to and from node reversed to pc28)
+sub pc30
+{
+ my ($fromnode, $tonode, $stream) = @_;
+ return "PC30^$fromnode^$tonode^$stream^";
+}
+
+# acknowledge this tranche of lines (to and from nodes reversed to pc29 and pc28
+sub pc31
+{
+ my ($fromnode, $tonode, $stream) = @_;
+ return "PC31^$fromnode^$tonode^$stream^";
+}
+
+# end of message from the sending end (pc28 node order)
+sub pc32
+{
+ my ($fromnode, $tonode, $stream) = @_;
+ return "PC32^$fromnode^$tonode^$stream^";
+}
+
+# acknowledge end of message from receiving end (opposite pc28 node order)
+sub pc33
+{
+ my ($fromnode, $tonode, $stream) = @_;
+ return "PC33^$fromnode^$tonode^$stream^";
+}
+
+
# send all the DX clusters I reckon are connected
sub pc38
{
projects / spider.git / commitdiff