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(%work @msg $msgdir %valid %busy);
  27
  28 %work = ();                # outstanding jobs
  29 @msg = ();                 # messages we have
  30 %busy = ();                # station interlocks
  31 $msgdir = "$main::root/msg";              # directory contain the msgs
  32
  33 %valid = (
  34   fromnode => '9,From Node',
  35   tonode => '9,To Node',
  36   to => '0,To',
  37   from => '0,From',
  38   t => '0,Msg Time,cldatetime',
  39   private => '9,Private',
  40   subject => '0,Subject',
  41   linesreq => '0,Lines per Gob',
  42   rrreq => '9,Read Confirm',
  43   origin => '0,Origin',
  44   lines => '5,Data',
  45   stream => '9,Stream No',
  46   count => '9,Gob Linecnt',
  47   file => '9,File?,yesno',
  48   gotit => '9,Got it Nodes,parray',
  49   lines => '9,Lines,parray',
  50   read => '9,Times read',
  51   size => '0,Size',
  52   msgno => '0,Msgno',
  53 );
  54
  55 # allocate a new object
  56 # called fromnode, tonode, from, to, datetime, private?, subject, nolinesper
  57 sub alloc
  58 {
  59   my $pkg = shift;
  60   my $self = bless {}, $pkg;
  61   $self->{msgno} = shift;
  62   $self->{to} = shift;
  63   $self->{from} = shift;
  64   $self->{t} = shift;
  65   $self->{private} = shift;
  66   $self->{subject} = shift;
  67   $self->{origin} = shift;
  68   $self->{read} = shift;
  69   $self->{gotit} = [];
  70
  71   return $self;
  72 }
  73
  74 sub workclean
  75 {
  76   my $ref = shift;
  77   delete $ref->{lines};
  78   delete $ref->{linesreq};
  79   delete $ref->{tonode};
  80   delete $ref->{fromnode};
  81   delete $ref->{stream};
  82   delete $ref->{lines};
  83   delete $ref->{file};
  84   delete $ref->{count};
  85 }
  86
  87 sub process
  88 {
  89   my ($self, $line) = @_;
  90   my @f = split /[\^\~]/, $line;
  91   my ($pcno) = $f[0] =~ /^PC(\d\d)/;          # just get the number
  92
  93   SWITCH: {
  94     if ($pcno == 28) {                        # incoming message
  95           my $t = cltounix($f[5], $f[6]);
  96           my $stream = next_transno($f[2]);
  97           my $ref = DXMsg->alloc($stream, $f[3], $f[4], $t, $f[7], $f[8], $f[13], '0', $f[11]);
  98
  99           # fill in various forwarding state variables
 100       $ref->{fromnode} = $f[2];
 101       $ref->{tonode} = $f[1];
 102           $ref->{rrreq} = $f[11];
 103           $ref->{linesreq} = $f[10];
 104           $ref->{stream} = $stream;
 105           $ref->{count} = 0;                      # no of lines between PC31s
 106           dbg('msg', "new message from $f[4] to $f[3] '$f[8]' stream $stream\n");
 107       $work{"$f[2]$stream"} = $ref;         # store in work
 108           $busy{$f[2]} = $ref;                          # set interlock
 109           $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack
 110           last SWITCH;
 111         }
 112
 113     if ($pcno == 29) {                        # incoming text
 114           my $ref = $work{"$f[2]$f[3]"};
 115           if ($ref) {
 116             push @{$ref->{lines}}, $f[4];
 117                 $ref->{count}++;
 118                 if ($ref->{count} >= $ref->{linesreq}) {
 119                   $self->send(DXProt::pc31($f[2], $f[1], $f[3]));
 120                   dbg('msg', "stream $f[3]: $ref->{linereq} lines received\n");
 121                   $ref->{count} = 0;
 122                 }
 123           }
 124           last SWITCH;
 125         }
 126
 127     if ($pcno == 30) {                        # this is a incoming subject ack
 128           my $ref = $work{$f[2]};          # note no stream at this stage
 129           delete $work{$f[2]};
 130           $ref->{stream} = $f[3];
 131           $ref->{count} = 0;
 132           $ref->{linesreq} = 5;
 133           $work{"$f[2]$f[3]"} = $ref;        # new ref
 134           dbg('msg', "incoming subject ack stream $[3]\n");
 135           $busy{$f[2]} = $ref;                       # interlock
 136           $ref->{lines} = [];
 137           push @{$ref->{lines}}, ($ref->read_msg_body);
 138           $ref->send_tranche($self);
 139           last SWITCH;
 140         }
 141
 142     if ($pcno == 31) {                        # acknowledge a tranche of lines
 143           my $ref = $work{"$f[2]$f[3]"};
 144           if ($ref) {
 145             dbg('msg', "tranche ack stream $f[3]\n");
 146             $ref->send_tranche($self);
 147           } else {
 148             $self->send(DXProt::pc42($f[2], $f[1], $f[3]));       # unknown stream
 149           }
 150           last SWITCH;
 151         }
 152
 153     if ($pcno == 32) {                         # incoming EOM
 154           dbg('msg', "stream $f[3]: EOM received\n");
 155           my $ref = $work{"$f[2]$f[3]"};
 156           if ($ref) {
 157             $self->send(DXProt::pc33($f[2], $f[1], $f[3]));# acknowledge it
 158
 159                 # get the next msg no - note that this has NOTHING to do with the stream number in PC protocol
 160                 # store the file or message
 161                 # remove extraneous rubbish from the hash
 162                 # remove it from the work in progress vector
 163                 # stuff it on the msg queue
 164                 if ($ref->{lines} && @{$ref->{lines}} > 0) {            # ignore messages with 0 lines
 165                   $ref->{msgno} = next_transno("Msgno") if !$ref->{file};
 166               push @{$ref->{gotit}}, $f[2];           # mark this up as being received
 167                   $ref->store($ref->{lines});
 168                   add_dir($ref);
 169                 }
 170                 $ref->stop_msg($self);
 171                 queue_msg();
 172           } else {
 173             $self->send(DXProt::pc42($f[2], $f[1], $f[3]));       # unknown stream
 174           }
 175           queue_msg();
 176           last SWITCH;
 177         }
 178
 179     if ($pcno == 33) {                         # acknowledge the end of message
 180           my $ref = $work{"$f[2]$f[3]"};
 181           if ($ref) {
 182                 if ($ref->{private}) {                   # remove it if it private and gone off site#
 183               $ref->del_msg;
 184             } else {
 185               push @{$ref->{gotit}}, $f[2];           # mark this up as being received
 186                   $ref->store($ref->{lines});             # re- store the file
 187             }
 188         $ref->stop_msg($self);
 189           } else {
 190             $self->send(DXProt::pc42($f[2], $f[1], $f[3]));       # unknown stream
 191           }
 192           queue_msg();
 193           last SWITCH;
 194         }
 195
 196         if ($pcno == 40) {                         # this is a file request
 197           $f[3] =~ s/\\/\//og;                     # change the slashes
 198           $f[3] =~ s/\.//og;                       # remove dots
 199           $f[3] = lc $f[3];                        # to lower case;
 200           dbg('msg', "incoming file $f[3]\n");
 201           last SWITCH if $f[3] =~ /^\/(perl|cmd|local_cmd|src|lib|include|sys|msg)\//;    # prevent access to executables
 202
 203           # create any directories
 204           my @part = split /\//, $f[3];
 205           my $part;
 206           my $fn = "$main::root";
 207           pop @part;         # remove last part
 208           foreach $part (@part) {
 209             $fn .= "/$part";
 210                 next if -e $fn;
 211             last SWITCH if !mkdir $fn, 0777;
 212         dbg('msg', "created directory $fn\n");
 213           }
 214           my $stream = next_transno($f[2]);
 215           my $ref = DXMsg->alloc($stream, "$main::root/$f[3]", $self->call, time, !$f[4], $f[3], ' ', '0', '0');
 216
 217           # forwarding variables
 218       $ref->{fromnode} = $f[1];
 219       $ref->{tonode} = $f[2];
 220           $ref->{linesreq} = $f[5];
 221           $ref->{stream} = $stream;
 222           $ref->{count} = 0;                      # no of lines between PC31s
 223           $ref->{file} = 1;
 224       $work{"$f[2]$stream"} = $ref;         # store in work
 225           $self->send(DXProt::pc30($f[2], $f[1], $stream));  # send ack
 226
 227           last SWITCH;
 228         }
 229
 230         if ($pcno == 42) {                        # abort transfer
 231           dbg('msg', "stream $f[3]: abort received\n");
 232           my $ref = $work{"$f[2]$f[3]"};
 233           if ($ref) {
 234             $ref->stop_msg($self);
 235                 $ref = undef;
 236           }
 237
 238           last SWITCH;
 239         }
 240   }
 241 }
 242
 243
 244 # store a message away on disc or whatever
 245 #
 246 # NOTE the second arg is a REFERENCE not a list
 247 sub store
 248 {
 249   my $ref = shift;
 250   my $lines = shift;
 251
 252   # we only proceed if there are actually any lines in the file
 253   if (!$lines || @{$lines} == 0) {
 254         return;
 255   }
 256
 257   if ($ref->{file}) {   # a file
 258     dbg('msg', "To be stored in $ref->{to}\n");
 259
 260     my $fh = new FileHandle "$ref->{to}", "w";
 261         if (defined $fh) {
 262           my $line;
 263           foreach $line (@{$lines}) {
 264                 print $fh "$line\n";
 265           }
 266           $fh->close;
 267           dbg('msg', "file $ref->{to} stored\n");
 268     } else {
 269       confess "can't open file $ref->{to} $!";
 270     }
 271 #       push @{$ref->{gotit}}, $ref->{fromnode} if $ref->{fromnode};
 272   } else {              # a normal message
 273
 274     # attempt to open the message file
 275         my $fn = filename($ref->{msgno});
 276
 277     dbg('msg', "To be stored in $fn\n");
 278
 279         # now save the file, overwriting what's there, YES I KNOW OK! (I will change it if it's a problem)
 280     my $fh = new FileHandle "$fn", "w";
 281         if (defined $fh) {
 282           my $rr = $ref->{rrreq} ? '1' : '0';
 283           my $priv = $ref->{private} ? '1': '0';
 284       print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{read}^$rr\n";
 285           print $fh "=== ", join('^', @{$ref->{gotit}}), "\n";
 286           my $line;
 287           $ref->{size} = 0;
 288           foreach $line (@{$lines}) {
 289         $ref->{size} += (length $line) + 1;
 290                 print $fh "$line\n";
 291           }
 292           $fh->close;
 293           dbg('msg', "msg $ref->{msgno} stored\n");
 294     } else {
 295       confess "can't open msg file $fn $!";
 296     }
 297   }
 298 }
 299
 300 # delete a message
 301 sub del_msg
 302 {
 303   my $self = shift;
 304
 305   # remove it from the active message list
 306   @msg = map { $_ != $self ? $_ : () } @msg;
 307
 308   # belt and braces (one day I will ask someone if this is REALLY necessary)
 309   delete $self->{gotit};
 310   delete $self->{list};
 311
 312   # remove the file
 313   unlink filename($self->{msgno});
 314   dbg('msg', "deleting $self->{msgno}\n");
 315 }
 316
 317 # read in a message header
 318 sub read_msg_header
 319 {
 320   my $fn = shift;
 321   my $file;
 322   my $line;
 323   my $ref;
 324   my @f;
 325   my $size;
 326
 327   $file = new FileHandle;
 328   if (!open($file, $fn)) {
 329     print "Error reading $fn $!\n";
 330     return undef;
 331   }
 332   $size = -s $fn;
 333   $line = <$file>;       # first line
 334   chomp $line;
 335   $size -= length $line;
 336   if (! $line =~ /^===/o) {
 337     print "corrupt first line in $fn ($line)\n";
 338     return undef;
 339   }
 340   $line =~ s/^=== //o;
 341   @f = split /\^/, $line;
 342   $ref = DXMsg->alloc(@f);
 343
 344   $line = <$file>;       # second line
 345   chomp $line;
 346   $size -= length $line;
 347   if (! $line =~ /^===/o) {
 348     print "corrupt second line in $fn ($line)\n";
 349     return undef;
 350   }
 351   $line =~ s/^=== //o;
 352   $ref->{gotit} = [];
 353   @f = split /\^/, $line;
 354   push @{$ref->{gotit}}, @f;
 355   $ref->{size} = $size;
 356
 357   close($file);
 358
 359   return $ref;
 360 }
 361
 362 # read in a message header
 363 sub read_msg_body
 364 {
 365   my $self = shift;
 366   my $msgno = $self->{msgno};
 367   my $file;
 368   my $line;
 369   my $fn = filename($msgno);
 370   my @out;
 371
 372   $file = new FileHandle;
 373   if (!open($file, $fn)) {
 374     print "Error reading $fn $!\n";
 375     return undef;
 376   }
 377   chomp (@out = <$file>);
 378   close($file);
 379
 380   shift @out if $out[0] =~ /^=== /;
 381   shift @out if $out[0] =~ /^=== /;
 382   return @out;
 383 }
 384
 385 # send a tranche of lines to the other end
 386 sub send_tranche
 387 {
 388   my ($self, $dxchan) = @_;
 389   my @out;
 390   my $to = $self->{tonode};
 391   my $from = $self->{fromnode};
 392   my $stream = $self->{stream};
 393   my $i;
 394
 395   for ($i = 0; $i < $self->{linesreq} && $self->{count} < @{$self->{lines}}; $i++, $self->{count}++) {
 396     push @out, DXProt::pc29($to, $from, $stream, ${$self->{lines}}[$self->{count}]);
 397   }
 398   push @out, DXProt::pc32($to, $from, $stream) if $i < $self->{linesreq};
 399   $dxchan->send(@out);
 400 }
 401
 402
 403 # find a message to send out and start the ball rolling
 404 sub queue_msg
 405 {
 406   my $sort = shift;
 407   my @nodelist = DXProt::get_all_ak1a();
 408   my $ref;
 409   my $clref;
 410   my $dxchan;
 411
 412   # bat down the message list looking for one that needs to go off site and whose
 413   # nearest node is not busy.
 414
 415   dbg('msg', "queue msg ($sort)\n");
 416   foreach $ref (@msg) {
 417     # firstly, is it private and unread? if so can I find the recipient
 418         # in my cluster node list offsite?
 419         if ($ref->{private}) {
 420           if ($ref->{read} == 0) {
 421             $clref = DXCluster->get($ref->{to});
 422                 if ($clref && !grep { $clref->{dxchan} == $_ } DXCommandmode::get_all) {
 423                   $dxchan = $clref->{dxchan};
 424                   $ref->start_msg($dxchan) if $clref && !get_busy($dxchan->call);
 425                 }
 426           }
 427         } elsif ($sort == undef) {
 428       # otherwise we are dealing with a bulletin, compare the gotit list with
 429           # the nodelist up above, if there are sites that haven't got it yet
 430           # then start sending it - what happens when we get loops is anyone's
 431           # guess, use (to, from, time, subject) tuple?
 432           my $noderef;
 433           foreach $noderef (@nodelist) {
 434             next if $noderef->call eq $main::mycall;
 435                 next if grep { $_ eq $noderef->call } @{$ref->{gotit}};
 436
 437                 # if we are here we have a node that doesn't have this message
 438                 $ref->start_msg($noderef) if !get_busy($noderef->call);
 439                 last;
 440           }
 441         }
 442
 443         # if all the available nodes are busy then stop
 444         last if @nodelist == scalar grep { get_busy($_->call) } @nodelist;
 445   }
 446 }
 447
 448 # start the message off on its travels with a PC28
 449 sub start_msg
 450 {
 451   my ($self, $dxchan) = @_;
 452
 453   dbg('msg', "start msg $self->{msgno}\n");
 454   $self->{linesreq} = 5;
 455   $self->{count} = 0;
 456   $self->{tonode} = $dxchan->call;
 457   $self->{fromnode} = $main::mycall;
 458   $busy{$dxchan->call} = $self;
 459   $work{"$self->{tonode}"} = $self;
 460   $dxchan->send(DXProt::pc28($self->{tonode}, $self->{fromnode}, $self->{to}, $self->{from}, $self->{t}, $self->{private}, $self->{subject}, $self->{origin}, $self->{rrreq}));
 461 }
 462
 463 # get the ref of a busy node
 464 sub get_busy
 465 {
 466   my $call = shift;
 467   return $busy{$call};
 468 }
 469
 470 # get the busy queue
 471 sub get_all_busy
 472 {
 473   return values %busy;
 474 }
 475
 476 # get the forwarding queue
 477 sub get_fwq
 478 {
 479   return values %work;
 480 }
 481
 482 # stop a message from continuing, clean it out, unlock interlocks etc
 483 sub stop_msg
 484 {
 485   my ($self, $dxchan) = @_;
 486   my $node = $dxchan->call;
 487
 488   dbg('msg', "stop msg $self->{msgno} stream $self->{stream}\n");
 489   delete $work{$node};
 490   delete $work{"$node$self->{stream}"};
 491   $self->workclean;
 492   delete $busy{$node};
 493 }
 494
 495 # get a new transaction number from the file specified
 496 sub next_transno
 497 {
 498   my $name = shift;
 499   $name =~ s/\W//og;      # remove non-word characters
 500   my $fn = "$msgdir/$name";
 501   my $msgno;
 502
 503   my $fh = new FileHandle;
 504   if (sysopen($fh, $fn, O_RDWR|O_CREAT, 0666)) {
 505     $fh->autoflush(1);
 506         $msgno = $fh->getline;
 507         chomp $msgno;
 508         $msgno++;
 509         seek $fh, 0, 0;
 510         $fh->print("$msgno\n");
 511         dbg('msg', "msgno $msgno allocated for $name\n");
 512         $fh->close;
 513   } else {
 514     confess "can't open $fn $!";
 515   }
 516   return $msgno;
 517 }
 518
 519 # initialise the message 'system', read in all the message headers
 520 sub init
 521 {
 522   my $dir = new FileHandle;
 523   my @dir;
 524   my $ref;
 525
 526   # read in the directory
 527   opendir($dir, $msgdir) or confess "can't open $msgdir $!";
 528   @dir = readdir($dir);
 529   closedir($dir);
 530
 531   for (sort @dir) {
 532     next if /^\./o;
 533         next if ! /^m\d+/o;
 534
 535     $ref = read_msg_header("$msgdir/$_");
 536         next if !$ref;
 537
 538         # add the message to the available queue
 539         add_dir($ref);
 540
 541   }
 542 }
 543
 544 # add the message to the directory listing
 545 sub add_dir
 546 {
 547   my $ref = shift;
 548   confess "tried to add a non-ref to the msg directory" if !ref $ref;
 549   push @msg, $ref;
 550 }
 551
 552 # return all the current messages
 553 sub get_all
 554 {
 555   return @msg;
 556 }
 557
 558 # get a particular message
 559 sub get
 560 {
 561   my $msgno = shift;
 562   for (@msg) {
 563     return $_ if $_->{msgno} == $msgno;
 564     last if $_->{msgno} > $msgno;
 565   }
 566   return undef;
 567 }
 568
 569 # return the official filename for a message no
 570 sub filename
 571 {
 572   return sprintf "$msgdir/m%06d", shift;
 573 }
 574
 575 #
 576 # return a list of valid elements
 577 #
 578
 579 sub fields
 580 {
 581   return keys(%valid);
 582 }
 583
 584 #
 585 # return a prompt for a field
 586 #
 587
 588 sub field_prompt
 589 {
 590   my ($self, $ele) = @_;
 591   return $valid{$ele};
 592 }
 593
 594 no strict;
 595 sub AUTOLOAD
 596 {
 597   my $self = shift;
 598   my $name = $AUTOLOAD;
 599   return if $name =~ /::DESTROY$/;
 600   $name =~ s/.*:://o;
 601
 602   confess "Non-existant field '$AUTOLOAD'" if !$valid{$name};
 603   @_ ? $self->{$name} = shift : $self->{$name} ;
 604 }
 605
 606 sub do_send_stuff
 607 {
 608   my $self = shift;
 609   my $line = shift;
 610   my @out;
 611
 612   if ($self->state eq 'send1') {
 613 #  $DB::single = 1;
 614     confess "local var gone missing" if !ref $self->{loc};
 615         my $loc = $self->{loc};
 616         $loc->{subject} = $line;
 617         $loc->{lines} = [];
 618         $self->state('sendbody');
 619         #push @out, $self->msg('sendbody');
 620         push @out, "Enter Message /EX (^Z) to send or /ABORT (^Y) to exit";
 621   } elsif ($self->state eq 'sendbody') {
 622     confess "local var gone missing" if !ref $self->{loc};
 623         my $loc = $self->{loc};
 624         if ($line eq "\032" || uc $line eq "/EX") {
 625       my $to;
 626
 627       if (@{$loc->{lines}} > 0) {
 628             foreach $to (@{$loc->{to}}) {
 629               my $ref;
 630                   my $systime = $main::systime;
 631                   my $mycall = $main::mycall;
 632                   $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'),
 633                                 uc $to,
 634                                                         $self->call,
 635                                                         $systime,
 636                                                         $loc->{private},
 637                                                         $loc->{subject},
 638                                                         $mycall,
 639                                                         $loc->{rrreq});
 640               $ref->store($loc->{lines});
 641                   $ref->add_dir();
 642                   #push @out, $self->msg('sendsent', $to);
 643                   push @out, "msgno $ref->{msgno} sent to $to";
 644             }
 645           }
 646           delete $loc->{lines};
 647           delete $loc->{to};
 648           delete $self->{loc};
 649           $self->state('prompt');
 650           $self->func(undef);
 651           DXMsg::queue_msg();
 652     } elsif ($line eq "\031" || uc $line eq "/ABORT" || uc $line eq "/QUIT") {
 653       #push @out, $self->msg('sendabort');
 654           push @out, "aborted";
 655           delete $loc->{lines};
 656           delete $loc->{to};
 657           delete $self->{loc};
 658           $self->func(undef);
 659           $self->state('prompt');
 660     } else {
 661
 662       # i.e. it ain't and end or abort, therefore store the line
 663       push @{$loc->{lines}}, $line;
 664     }
 665   }
 666   return (1, @out);
 667 }
 668
 669 1;
 670
 671 __END__