perl/DXMsg.pm

   1 #!/usr/bin/perl
   2 #
   3 # This module impliments the message handling for a dx cluster
   4 #
   5 # Copyright (c) 1998 Dirk Koopman G1TLH
   6 #
   7 # $Id$
   8 #
   9
  10 package DXMsg;
  11
  12 @ISA = qw(DXProt DXChannel);
  13
  14 use DXUtil;
  15 use DXChannel;
  16 use DXUser;
  17 use DXM;
  18 use DXCluster;
  19 use DXProtVars;
  20 use DXProtout;
  21 use DXDebug;
  22 use FileHandle;
  23 use Carp;
  24
  25 use strict;
  26 use vars qw($stream %work @msg $msgdir $msgnofn);
  27
  28 %work = ();                # outstanding jobs
  29 @msg = ();                 # messages we have
  30 $msgdir = "$main::data/msg";              # directory contain the msgs
  31
  32 # allocate a new object
  33 # called fromnode, tonode, from, to, datetime, private?, subject, nolinesper
  34 sub alloc
  35 {
  36   my $pkg = shift;
  37   my $self = bless {}, $pkg;
  38   $self->{fromnode} = shift;
  39   $self->{tonode} = shift;
  40   $self->{to} = shift;
  41   $self->{from} = shift;
  42   $self->{t} = shift;
  43   $self->{private} = shift;
  44   $self->{subject} = shift;
  45   $self->{linesreq} = shift;    # this the number of lines to send or receive between PC31s
  46   $self->{rrreq} = shift;       # a read receipt is required
  47   $self->{origin} = shift;
  48   $self->{stream} = shift;
  49   $self->{lines} = [];
  50
  51   return $self;
  52 }
  53
  54 sub workclean
  55 {
  56   my $ref = shift;
  57   delete $ref->{lines};
  58   delete $ref->{linesreq};
  59   delete $ref->{tonode};
  60   delete $ref->{stream};
  61 }
  62
  63 sub process
  64 {
  65   my ($self, $line) = @_;
  66   my @f = split /[\^\~]/, $line;
  67   my ($pcno) = $f[0] =~ /^PC(\d\d)/;          # just get the number
  68
  69   SWITCH: {
  70     if ($pcno == 28) {                        # incoming message
  71           my $t = cltounix($f[5], $f[6]);
  72           my $stream = next_transno($f[2]);
  73           my $ref = DXMsg->alloc($f[1], $f[2], $f[3], $f[4], $t, $f[7], $f[8], $f[10], $f[11], $f[13], $stream);
  74           dbg('msg', "new message from $f[4] to $f[3] '$f[8]' stream $stream\n");
  75       $work{"$f[1]$f[2]$stream"} = $ref;         # store in work
  76           $self->send(DXProt::pc30($f[2], $f[1], $stream));
  77           $ref->{count} = 0;                      # no of lines between PC31s
  78           last SWITCH;
  79         }
  80
  81     if ($pcno == 29) {                        # incoming text
  82           my $ref = $work{"$f[1]$f[2]$f[3]"};
  83           if ($ref) {
  84             push @{$ref->{lines}}, $f[4];
  85                 $ref->{count}++;
  86                 if ($ref->{count} >= $ref->{linesreq}) {
  87                   $self->send(DXProt::pc31($f[2], $f[1], $f[3]));
  88                   dbg('msg', "stream $f[3]: $ref->{linereq} lines received\n");
  89                   $ref->{count} = 0;
  90                 }
  91           }
  92           last SWITCH;
  93         }
  94
  95     if ($pcno == 30) {
  96           last SWITCH;
  97         }
  98
  99     if ($pcno == 31) {
 100           last SWITCH;
 101         }
 102
 103     if ($pcno == 32) {                         # incoming EOM
 104           dbg('msg', "stream $f[3]: EOM received\n");
 105           my $ref = $work{"$f[1]$f[2]$f[3]"};
 106           if ($ref) {
 107             $self->send(DXProt::pc33($f[2], $f[1], $f[3]));# acknowledge it
 108                 $ref->store();                         # store it (whatever that may mean)
 109                 delete $work{"$f[1]$f[2]$f[3]"};       # remove the reference from the work vector
 110           }
 111           last SWITCH;
 112         }
 113
 114     if ($pcno == 33) {
 115           last SWITCH;
 116         }
 117
 118         if ($pcno == 40) {                         # this is a file request
 119           $f[3] =~ s/\\/\//og;                     # change the slashes
 120           $f[3] =~ s/\.//og;                       # remove dots
 121           $f[3] = lc $f[3];                        # to lower case;
 122           dbg('msg', "incoming file $f[3]\n");
 123           last SWITCH if $f[3] =~ /^\/(perl|cmd|local_cmd|src|lib|include|sys|data\/msg)\//;    # prevent access to executables
 124
 125           # create any directories
 126           my @part = split /\//, $f[3];
 127           my $part;
 128           my $fn = "$main::root";
 129           pop @part;         # remove last part
 130           foreach $part (@part) {
 131             $fn .= "/$part";
 132                 next if -e $fn;
 133             last SWITCH if !mkdir $fn, 0777;
 134         dbg('msg', "created directory $fn\n");
 135           }
 136           my $stream = next_transno($f[2]);
 137           my $ref = DXMsg->alloc($f[1], $f[2], "$main::root/$f[3]", undef, time, !$f[4], undef, $f[5], 0, ' ', $stream);
 138           $ref->{file} = 1;
 139       $work{"$f[1]$f[2]$stream"} = $ref;         # store in work
 140           $self->send(DXProt::pc30($f[2], $f[1], $stream));
 141           $ref->{count} = 0;                      # no of lines between PC31s
 142
 143           last SWITCH;
 144         }
 145   }
 146 }
 147
 148
 149 # store a message away on disc or whatever
 150 sub store
 151 {
 152   my $ref = shift;
 153
 154   # we only proceed if there are actually any lines in the file
 155   if (@{$ref->{lines}} == 0) {
 156     delete $ref->{lines};
 157         return;
 158   }
 159
 160   if ($ref->{file}) {   # a file
 161     dbg('msg', "To be stored in $ref->{to}\n");
 162
 163     my $fh = new FileHandle "$ref->{to}", "w";
 164         if (defined $fh) {
 165           my $line;
 166           foreach $line (@{$ref->{lines}}) {
 167                 print $fh "$line\n";
 168           }
 169           $fh->close;
 170           dbg('msg', "file $ref->{to} stored\n");
 171     } else {
 172       confess "can't open file $ref->{to} $!";
 173     }
 174   } else {              # a normal message
 175
 176     # get the next msg no - note that this has NOTHING to do with the stream number in PC protocol
 177         my $msgno = next_transno("msgno");
 178
 179     # attempt to open the message file
 180         my $fn = sprintf "$msgdir/m%06d", $msgno;
 181
 182     dbg('msg', "To be stored in $fn\n");
 183
 184     my $fh = new FileHandle "$fn", "w";
 185         if (defined $fh) {
 186       print $fh "=== $ref->{to}^$ref->{from}^$ref->{private}^$ref->{subject}^$ref->{origin}\n";
 187           print $fh "=== $ref->{fromnode}\n";
 188           my $line;
 189           foreach $line (@{$ref->{lines}}) {
 190         $ref->{size} += length $line + 1;
 191                 print $fh "$line\n";
 192           }
 193           $ref->workclean();
 194           push @msg, $ref;           # add this message to the incore message list
 195           $fh->close;
 196           dbg('msg', "msg $msgno stored\n");
 197     } else {
 198       confess "can't open msg file $fn $!";
 199     }
 200   }
 201 }
 202
 203 # get a new transaction number from the file specified
 204 sub next_transno
 205 {
 206   my $name = shift;
 207   $name =~ s/\W//og;      # remove non-word characters
 208   my $fn = "$msgdir/$name";
 209   my $msgno;
 210
 211   my $fh = new FileHandle;
 212   if (sysopen($fh, $fn, O_RDWR|O_CREAT, 0666)) {
 213     $fh->autoflush(1);
 214         $msgno = $fh->getline;
 215         chomp $msgno;
 216         $msgno++;
 217         seek $fh, 0, 0;
 218         $fh->print("$msgno\n");
 219         dbg('msg', "msgno $msgno allocated for $name\n");
 220         $fh->close;
 221   } else {
 222     confess "can't open $fn $!";
 223   }
 224   return $msgno;
 225 }
 226
 227 # initialise the message 'system'
 228 sub init
 229 {
 230
 231 }
 232
 233 1;
 234
 235 __END__