]> scm.dxcluster.org Git - spider.git/commitdiff
The guts of the new Msg system
authorminima <minima>
Fri, 2 Mar 2001 13:37:03 +0000 (13:37 +0000)
committerminima <minima>
Fri, 2 Mar 2001 13:37:03 +0000 (13:37 +0000)
perl/ExtMsg.pm [new file with mode: 0644]
perl/IntMsg.pm [new file with mode: 0644]
perl/Listeners.pm [new file with mode: 0644]
perl/Msg.pm
perl/client.pl
perl/cluster.pl
perl/console.pl

diff --git a/perl/ExtMsg.pm b/perl/ExtMsg.pm
new file mode 100644 (file)
index 0000000..dafa856
--- /dev/null
@@ -0,0 +1,284 @@
+#
+# This class is the internal subclass that deals with the external port
+# communications for Msg.pm
+#
+# This is where the cluster handles direct connections coming both in
+# and out
+#
+# $Id$
+#
+# Copyright (c) 2001 - Dirk Koopman G1TLH
+#
+
+package ExtMsg;
+
+use strict;
+use Msg;
+use DXVars;
+use DXUtil;
+use DXDebug;
+use IO::File;
+use IO::Socket;
+
+use vars qw(@ISA $deftimeout);
+
+@ISA = qw(Msg);
+$deftimeout = 60;
+
+sub enqueue
+{
+       my ($conn, $msg) = @_;
+       unless ($msg =~ /^[ABZ]/) {
+               if ($msg =~ /^E[-\w]+\|([01])/) {
+                       $conn->{echo} = $1;
+               } else {
+                       $msg =~ s/^[-\w]+\|//;
+                       push (@{$conn->{outqueue}}, $msg . $conn->{lineend});
+               }
+       }
+}
+
+sub send_raw
+{
+       my ($conn, $msg) = @_;
+    my $sock = $conn->{sock};
+    return unless defined($sock);
+       push (@{$conn->{outqueue}}, $msg);
+       dbg('connect', $msg) unless $conn->{state} eq 'C';
+    Msg::set_event_handler ($sock, "write" => sub {$conn->_send(0)});
+}
+
+sub dequeue
+{
+       my $conn = shift;
+       my $msg;
+       
+       while ($msg = shift @{$conn->{inqueue}}){
+               dbg('connect', $msg) unless $conn->{state} eq 'C';
+               
+               $msg =~ s/\xff\xfa.*\xff\xf0|\xff[\xf0-\xfe].//g; # remove telnet options
+               $msg =~ s/[\x00-\x08\x0a-\x1f\x80-\x9f]/./g;         # immutable CSI sequence + control characters
+
+               if ($conn->{state} eq 'C') {
+                       &{$conn->{rproc}}($conn, "I$conn->{call}|$msg", $!);
+                       $! = 0;
+               } elsif ($conn->{state} eq 'WL' ) {
+                       $msg = uc $msg;
+                       if (is_callsign($msg)) {
+                               _send_file($conn, "$main::data/connected");
+                               $conn->{call} = $msg;
+                               &{$conn->{rproc}}($conn, "A$conn->{call}|telnet");
+                               $conn->{state} = 'C';
+                       } else {
+                               $conn->send_now("Sorry $msg is an invalid callsign");
+                               $conn->disconnect;
+                       }
+               } elsif ($conn->{state} eq 'WC') {
+                       if (exists $conn->{cmd} && @{$conn->{cmd}}) {
+                               $conn->_docmd($msg);
+                               unless (@{$conn->{cmd}}) {
+                                       $conn->{state} = 'C';
+                                       &{$conn->{rproc}}($conn, "O$conn->{call}|telnet");
+                                       delete $conn->{cmd};
+                                       $conn->{timeout}->del_timer if $conn->{timeout};
+                               }
+                       }
+               }
+       }
+       if ($conn->{msg} && $conn->{state} eq 'WC' && exists $conn->{cmd} && @{$conn->{cmd}}) {
+               $conn->_docmd($conn->{msg});
+               unless (@{$conn->{cmd}}) {
+                       $conn->{state} = 'C';
+                       &{$conn->{rproc}}($conn, "O$conn->{call}|telnet");
+                       delete $conn->{cmd};
+                       $conn->{timeout}->del_timer if $conn->{timeout};
+               }
+       }
+}
+
+sub new_client {
+       my $server_conn = shift;
+    my $sock = $server_conn->{sock}->accept();
+    my $conn = $server_conn->new($server_conn->{rproc});
+       $conn->{sock} = $sock;
+
+    my $rproc = &{$server_conn->{rproc}} ($conn, $sock->peerhost(), $sock->peerport());
+    if ($rproc) {
+        $conn->{rproc} = $rproc;
+        my $callback = sub {$conn->_rcv};
+               Msg::set_event_handler ($sock, "read" => $callback);
+               # send login prompt
+               $conn->{state} = 'WL';
+#              $conn->send_raw("\xff\xfe\x01\xff\xfc\x01\ff\fd\x22");
+#              $conn->send_raw("\xff\xfa\x22\x01\x01\xff\xf0");
+               _send_file($conn, "$main::data/issue");
+               $conn->send_raw("Login: ");
+    } else { 
+        $conn->disconnect();
+    }
+}
+
+sub start_connect
+{
+       my $call = shift;
+       my $fn = shift;
+       my $conn = ExtMsg->new(\&main::rec); 
+       $conn->{call} = $call;
+       
+       my $f = new IO::File $fn;
+       push @{$conn->{cmd}}, <$f>;
+       $f->close;
+       push @main::outstanding_connects, {call => $call, conn => $conn};
+       $conn->_dotimeout($deftimeout);
+       $conn->_docmd;
+}
+
+sub _docmd
+{
+       my $conn = shift;
+       my $msg = shift;
+       my $cmd;
+
+       while ($cmd = shift @{$conn->{cmd}}) {
+               chomp $cmd;
+               next if $cmd =~ /^\s*\#/o;
+               next if $cmd =~ /^\s*$/o;
+               $conn->_doabort($1) if $cmd =~ /^\s*a\w*\s+(.*)/i;
+               $conn->_dotimeout($1) if $cmd =~ /^\s*t\w*\s+(\d+)/i;
+               $conn->_dolineend($1) if $cmd =~ /^\s*[Ll]\w*\s+\'((?:\\[rn])+)\'/i;
+               if ($cmd =~ /^\s*co\w*\s+(\w+)\s+(.*)$/i) {
+                       unless ($conn->_doconnect($1, $2)) {
+                               $conn->disconnect;
+                               @{$conn->{cmd}} = [];    # empty any further commands
+                               last;
+                       }  
+               }
+               if ($cmd =~ /^\s*\'.*\'\s+\'.*\'/i) {
+                       $conn->_dochat($cmd, $msg);
+                       last;
+               }
+               if ($cmd =~ /^\s*cl\w+\s+(.*)/i) {
+                       $conn->_doclient($1);
+                       last;
+               }
+               last if $conn->{state} eq 'E';
+       }
+       unless (exists $conn->{cmd} && @{$conn->{cmd}}) {
+               @main::outstanding_connects = grep {$_->{call} ne $conn->{call}} @main::outstanding_connects;
+       }
+}
+
+sub _doconnect
+{
+       my ($conn, $sort, $line) = @_;
+       my $r;
+       
+       dbg('connect', "CONNECT sort: $sort command: $line");
+       if ($sort eq 'telnet') {
+               # this is a straight network connect
+               my ($host, $port) = split /\s+/, $line;
+               $port = 23 if !$port;
+               $r = $conn->connect($host, $port);
+               if ($r) {
+                       dbg('connect', "Connected to $host $port");
+               } else {
+                       dbg('connect', "***Connect Failed to $host $port $!");
+               }
+       } elsif ($sort eq 'ax25' || $sort eq 'prog') {
+               ;
+       } else {
+               dbg('err', "invalid type of connection ($sort)");
+               $conn->disconnect;
+       }
+       return $r;
+}
+
+sub _doabort
+{
+       my $conn = shift;
+       my $string = shift;
+       dbg('connect', "abort $string");
+       $conn->{abort} = $string;
+}
+
+sub _dotimeout
+{
+       my $conn = shift;
+       my $val = shift;
+       dbg('connect', "timeout set to $val");
+       $conn->{timeout}->del_timer if $conn->{timeout};
+       $conn->{timeout} = ExtMsg->new_timer($val, sub{ _timeout($conn); });
+       $conn->{timeval} = $val;
+}
+
+sub _dolineend
+{
+       my $conn = shift;
+       my $val = shift;
+       dbg('connect', "lineend set to $val ");
+       $val =~ s/\\r/\r/g;
+       $val =~ s/\\n/\n/g;
+       $conn->{lineend} = $val;
+}
+
+sub _dochat
+{
+       my $conn = shift;
+       my $cmd = shift;
+       my $line = shift;
+       
+       if ($line) {
+               my ($expect, $send) = $cmd =~ /^\s*\'(.*)\'\s+\'(.*)\'/;
+               if ($expect) {
+                       dbg('connect', "expecting: \"$expect\" received: \"$line\"");
+                       if ($conn->{abort} && $line =~ /$conn->{abort}/i) {
+                               dbg('connect', "aborted on /$conn->{abort}/");
+                               $conn->disconnect;
+                               return;
+                       }
+                       if ($line =~ /$expect/i) {
+                               dbg('connect', "got: \"$expect\" sending: \"$send\"");
+                               $conn->send_later($send);
+                               return;
+                       }
+               }
+       }
+       $conn->{state} = 'WC';
+       unshift @{$conn->{cmd}}, $cmd;
+}
+
+sub _timeout
+{
+       my $conn = shift;
+       dbg('connect', "timed out after $conn->{timeval} seconds");
+       $conn->disconnect;
+       @main::outstanding_connects = grep {$_->{call} ne $conn->{call}} @main::outstanding_connects;
+}
+
+# handle callsign and connection type firtling
+sub _doclient
+{
+       my $conn = shift;
+       my $line = shift;
+       my @f = split /\s+/, $line;
+       $conn->{call} = uc $f[0] if $f[0];
+       $conn->{csort} = $f[1] if $f[1];
+}
+
+sub _send_file
+{
+       my $conn = shift;
+       my $fn = shift;
+       
+       if (-e $fn) {
+               my $f = new IO::File $fn;
+               if ($f) {
+                       while (<$f>) {
+                               chomp;
+                               $conn->send_later($_);
+                       }
+                       $f->close;
+               }
+       }
+       $! = undef;
+}
diff --git a/perl/IntMsg.pm b/perl/IntMsg.pm
new file mode 100644 (file)
index 0000000..150ec91
--- /dev/null
@@ -0,0 +1,37 @@
+#
+# This class is the internal subclass that deals with the internal port 27754
+# communications for Msg.pm
+#
+# $Id$
+#
+# Copyright (c) 2001 - Dirk Koopman G1TLH
+#
+
+package IntMsg;
+
+use strict;
+use Msg;
+
+use vars qw(@ISA);
+
+@ISA = qw(Msg);
+
+sub enqueue
+{
+       my ($conn, $msg) = @_;
+       $msg =~ s/([\%\x00-\x1f\x7f-\xff])/sprintf("%%%02X", ord($1))/eg; 
+    push (@{$conn->{outqueue}}, $msg . "\n");
+}
+
+sub dequeue
+{
+       my $conn = shift;
+       my $msg;
+       
+       while ($msg = shift @{$conn->{inqueue}}){
+               $msg =~ s/\%([2-9A-F][0-9A-F])/chr(hex($1))/eg;
+               $msg =~ s/[\x00-\x08\x0a-\x1f\x80-\x9f]/./g;         # immutable CSI sequence + control characters
+               &{$conn->{rproc}}($conn, $msg, $!);
+               $! = 0;
+       }
+}
diff --git a/perl/Listeners.pm b/perl/Listeners.pm
new file mode 100644 (file)
index 0000000..cc83051
--- /dev/null
@@ -0,0 +1,19 @@
+#
+# Copy this file to /spider/local and modify it to your requirements
+#
+#
+# This file specifies which local interfaces and ports you will be
+# listening on
+#
+# You can add as many as you like
+#
+
+package main;
+
+use vars qw(@listen);
+
+@listen = (
+#                 ["localhost", 7300],
+#                 ["foo.dxcluster.net", 7300],
+                 );
+
index f4f013beeb0f659417c84cd8750ccc963a8c9348..74c64ec7c70da7e92dd7c9015010110d8bdcf9b1 100644 (file)
@@ -15,12 +15,15 @@ use IO::Select;
 use IO::Socket;
 #use DXDebug;
 
 use IO::Socket;
 #use DXDebug;
 
-use vars qw(%rd_callbacks %wt_callbacks $rd_handles $wt_handles);
+use vars qw(%rd_callbacks %wt_callbacks $rd_handles $wt_handles $now @timerchain);
 
 %rd_callbacks = ();
 %wt_callbacks = ();
 $rd_handles   = IO::Select->new();
 $wt_handles   = IO::Select->new();
 
 %rd_callbacks = ();
 %wt_callbacks = ();
 $rd_handles   = IO::Select->new();
 $wt_handles   = IO::Select->new();
+$now = time;
+@timerchain = ();
+
 my $blocking_supported = 0;
 
 BEGIN {
 my $blocking_supported = 0;
 
 BEGIN {
@@ -31,13 +34,33 @@ BEGIN {
     $blocking_supported = 1 unless $@;
 }
 
     $blocking_supported = 1 unless $@;
 }
 
+#
+#-----------------------------------------------------------------
+# Generalised initializer
+
+sub new
+{
+    my ($pkg, $rproc) = @_;
+       my $obj = ref($pkg);
+       my $class = $obj || $pkg;
+
+    my $conn = {
+        rproc => $rproc,
+               inqueue => [],
+               state => 0,
+               lineend => "\r\n",
+               csort => 'telnet',
+    };
+
+       return bless $conn, $class;
+}
+
 #-----------------------------------------------------------------
 # Send side routines
 sub connect {
 #-----------------------------------------------------------------
 # Send side routines
 sub connect {
-    my ($pkg, $to_host, $to_port,$rcvd_notification_proc) = @_;
-    
+    my ($pkg, $to_host, $to_port, $rproc) = @_;
+
     # Create a new internet socket
     # Create a new internet socket
-    
     my $sock = IO::Socket::INET->new (
                                       PeerAddr => $to_host,
                                       PeerPort => $to_port,
     my $sock = IO::Socket::INET->new (
                                       PeerAddr => $to_host,
                                       PeerPort => $to_port,
@@ -47,22 +70,26 @@ sub connect {
     return undef unless $sock;
 
     # Create a connection end-point object
     return undef unless $sock;
 
     # Create a connection end-point object
-    my $conn = {
-        sock                   => $sock,
-        rcvd_notification_proc => $rcvd_notification_proc,
-    };
+    my $conn = $pkg;
+       unless (ref $pkg) {
+               $conn = $pkg->new($rproc);
+       }
+       $conn->{sock} = $sock;
     
     
-    if ($rcvd_notification_proc) {
+    if ($conn->{rproc}) {
         my $callback = sub {_rcv($conn)};
         set_event_handler ($sock, "read" => $callback);
     }
         my $callback = sub {_rcv($conn)};
         set_event_handler ($sock, "read" => $callback);
     }
-    return bless $conn, $pkg;
+    return $conn;
 }
 
 sub disconnect {
     my $conn = shift;
     my $sock = delete $conn->{sock};
 }
 
 sub disconnect {
     my $conn = shift;
     my $sock = delete $conn->{sock};
-    return unless defined($sock);
+       $conn->{state} = 'E';
+       delete $conn->{cmd};
+       $conn->{timeout}->del_timer if $conn->{timeout};
+       return unless defined($sock);
     set_event_handler ($sock, "read" => undef, "write" => undef);
     shutdown($sock, 3);
        close($sock);
     set_event_handler ($sock, "read" => undef, "write" => undef);
     shutdown($sock, 3);
        close($sock);
@@ -70,31 +97,28 @@ sub disconnect {
 
 sub send_now {
     my ($conn, $msg) = @_;
 
 sub send_now {
     my ($conn, $msg) = @_;
-    _enqueue ($conn, $msg);
+    $conn->enqueue($msg);
     $conn->_send (1); # 1 ==> flush
 }
 
 sub send_later {
     my ($conn, $msg) = @_;
     $conn->_send (1); # 1 ==> flush
 }
 
 sub send_later {
     my ($conn, $msg) = @_;
-    _enqueue($conn, $msg);
+    $conn->enqueue($msg);
     my $sock = $conn->{sock};
     return unless defined($sock);
     set_event_handler ($sock, "write" => sub {$conn->_send(0)});
 }
 
     my $sock = $conn->{sock};
     return unless defined($sock);
     set_event_handler ($sock, "write" => sub {$conn->_send(0)});
 }
 
-sub _enqueue {
-    my ($conn, $msg) = @_;
-    # prepend length (encoded as network long)
-    my $len = length($msg);
-       $msg =~ s/([\%\x00-\x1f\x7f-\xff])/sprintf("%%%02X", ord($1))/eg; 
-    push (@{$conn->{queue}}, $msg . "\n");
+sub enqueue {
+    my $conn = shift;
+    push (@{$conn->{outqueue}}, $_[0]);
 }
 
 sub _send {
     my ($conn, $flush) = @_;
     my $sock = $conn->{sock};
     return unless defined($sock);
 }
 
 sub _send {
     my ($conn, $flush) = @_;
     my $sock = $conn->{sock};
     return unless defined($sock);
-    my ($rq) = $conn->{queue};
+    my ($rq) = $conn->{outqueue};
 
     # If $flush is set, set the socket to blocking, and send all
     # messages in the queue - return only if there's an error
 
     # If $flush is set, set the socket to blocking, and send all
     # messages in the queue - return only if there's an error
@@ -179,21 +203,31 @@ sub handle_send_err {
 #-----------------------------------------------------------------
 # Receive side routines
 
 #-----------------------------------------------------------------
 # Receive side routines
 
-my ($g_login_proc,$g_pkg);
-my $main_socket = 0;
 sub new_server {
 sub new_server {
-    @_ == 4 || die "Msg->new_server (myhost, myport, login_proc)\n";
+    @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
     my ($pkg, $my_host, $my_port, $login_proc) = @_;
     my ($pkg, $my_host, $my_port, $login_proc) = @_;
-    
-    $main_socket = IO::Socket::INET->new (
+       my $self = $pkg->new($login_proc);
+       
+    $self->{sock} = IO::Socket::INET->new (
                                           LocalAddr => $my_host,
                                           LocalPort => $my_port,
                                           Listen    => 5,
                                           Proto     => 'tcp',
                                           Reuse     => 1);
                                           LocalAddr => $my_host,
                                           LocalPort => $my_port,
                                           Listen    => 5,
                                           Proto     => 'tcp',
                                           Reuse     => 1);
-    die "Could not create socket: $! \n" unless $main_socket;
-    set_event_handler ($main_socket, "read" => \&_new_client);
-    $g_login_proc = $login_proc; $g_pkg = $pkg;
+    die "Could not create socket: $! \n" unless $self->{sock};
+    set_event_handler ($self->{sock}, "read" => sub { $self->new_client }  );
+       return $self;
+}
+
+sub dequeue
+{
+       my $conn = shift;
+       my $msg;
+       
+       while ($msg = shift @{$conn->{inqueue}}){
+               &{$conn->{rproc}}($conn, $msg, $!);
+               $! = 0;
+       }
 }
 
 sub _rcv {                     # Complement to _send
 }
 
 sub _rcv {                     # Complement to _send
@@ -209,13 +243,16 @@ sub _rcv {                     # Complement to _send
        if (defined ($bytes_read)) {
                if ($bytes_read > 0) {
                        if ($msg =~ /\n/) {
        if (defined ($bytes_read)) {
                if ($bytes_read > 0) {
                        if ($msg =~ /\n/) {
-                               @lines = split /\n/, $msg;
-                               $lines[0] = $conn->{msg} . $lines[0] if $conn->{msg};
+                               @lines = split /\r?\n/, $msg;
+                               push @lines, ' ' unless @lines;
+                               
+                               $lines[0] = $conn->{msg} . $lines[0] if exists $conn->{msg};
                                if ($msg =~ /\n$/) {
                                        delete $conn->{msg};
                                } else {
                                        $conn->{msg} = pop @lines;
                                }
                                if ($msg =~ /\n$/) {
                                        delete $conn->{msg};
                                } else {
                                        $conn->{msg} = pop @lines;
                                }
+                               push @{$conn->{inqueue}}, @lines if @lines;
                        } else {
                                $conn->{msg} .= $msg;
                        }
                        } else {
                                $conn->{msg} .= $msg;
                        }
@@ -231,29 +268,21 @@ sub _rcv {                     # Complement to _send
 FINISH:
     if (defined $bytes_read && $bytes_read == 0) {
 #              $conn->disconnect();
 FINISH:
     if (defined $bytes_read && $bytes_read == 0) {
 #              $conn->disconnect();
-               &{$conn->{rcvd_notification_proc}}($conn, undef, $!);
-               @lines = ();
-    } 
-
-       while (@lines){
-               $msg = shift @lines;
-               $msg =~ s/\%([2-9A-F][0-9A-F])/chr(hex($1))/eg;
-               $msg =~ s/[\x00-\x08\x0a-\x1f\x80-\x9f]/./g;         # immutable CSI sequence + control characters
-               &{$conn->{rcvd_notification_proc}}($conn, $msg, $!);
-               $! = 0;
+               &{$conn->{rproc}}($conn, undef, $!);
+               delete $conn->{inqueue};
+    } else {
+               $conn->dequeue;
        }
 }
 
        }
 }
 
-sub _new_client {
-    my $sock = $main_socket->accept();
-    my $conn = bless {
-        'sock' =>  $sock,
-        'state' => 'connected'
-    }, $g_pkg;
-    my $rcvd_notification_proc =
-        &$g_login_proc ($conn, $sock->peerhost(), $sock->peerport());
-    if ($rcvd_notification_proc) {
-        $conn->{rcvd_notification_proc} = $rcvd_notification_proc;
+sub new_client {
+       my $server_conn = shift;
+    my $sock = $server_conn->{sock}->accept();
+    my $conn = $server_conn->new($server_conn->{rproc});
+       $conn->{sock} = $sock;
+    my $rproc = &{$server_conn->{rproc}} ($conn, $sock->peerhost(), $sock->peerport());
+    if ($rproc) {
+        $conn->{rproc} = $rproc;
         my $callback = sub {_rcv($conn)};
         set_event_handler ($sock, "read" => $callback);
     } else {  # Login failed
         my $callback = sub {_rcv($conn)};
         set_event_handler ($sock, "read" => $callback);
     } else {  # Login failed
@@ -263,9 +292,9 @@ sub _new_client {
 
 sub close_server
 {
 
 sub close_server
 {
-       set_event_handler ($main_socket, "read" => undef);
-       $main_socket->close;
-       $main_socket = 0;
+       my $conn = shift;
+       set_event_handler ($conn->{sock}, "read" => undef);
+       $conn->{sock}->close;
 }
 
 #----------------------------------------------------
 }
 
 #----------------------------------------------------
@@ -297,20 +326,53 @@ sub set_event_handler {
     }
 }
 
     }
 }
 
+sub new_timer
+{
+    my ($pkg, $time, $proc, $recur) = @_;
+       my $obj = ref($pkg);
+       my $class = $obj || $pkg;
+       my $self = bless { t=>$time + time, proc=>$proc }, $class;
+       $self->{interval} = $time if $recur;
+       push @timerchain, $self;
+       return $self;
+}
+
+sub del_timer
+{
+       my $self = shift;
+       @timerchain = grep {$_ != $self} @timerchain;
+}
+
 sub event_loop {
     my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
     my ($conn, $r, $w, $rset, $wset);
     while (1) {
 sub event_loop {
     my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
     my ($conn, $r, $w, $rset, $wset);
     while (1) {
-        # Quit the loop if no handles left to process
+       # Quit the loop if no handles left to process
         last unless ($rd_handles->count() || $wt_handles->count());
         last unless ($rd_handles->count() || $wt_handles->count());
-        ($rset, $wset) =
+        
+               ($rset, $wset) =
             IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
             IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
+               $now = time;
+               
         foreach $r (@$rset) {
             &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
         }
         foreach $w (@$wset) {
             &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
         }
         foreach $r (@$rset) {
             &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
         }
         foreach $w (@$wset) {
             &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
         }
+
+               # handle things on the timer chain
+               for (@timerchain) {
+                       if ($now >= $_->{t}) {
+                               &{$_->{proc}}();
+                               $_->{t} = $now + $_->{interval} if exists $_->{interval};
+                       }
+               }
+
+               # remove dead timers
+               @timerchain = grep { $_->{t} > $now } @timerchain;
+               
         if (defined($loop_count)) {
             last unless --$loop_count;
         }
         if (defined($loop_count)) {
             last unless --$loop_count;
         }
index ddf3fd3ba6adba52f6ed6f19ccab464d613dbd05..57a5d45b0fa4ea373ac434a08d9c2b623687febe 100755 (executable)
@@ -39,6 +39,7 @@ BEGIN {
 }
 
 use Msg;
 }
 
 use Msg;
+use IntMsg;
 use DXVars;
 use DXDebug;
 use DXUtil;
 use DXVars;
 use DXDebug;
 use DXUtil;
@@ -78,8 +79,10 @@ sub sig_term
 # terminate a child
 sub sig_chld
 {
 # terminate a child
 sub sig_chld
 {
-       $SIG{CHLD} = \&sig_chld;
-       $waitedpid = wait;
+       unless ($^O =~ /^MS/i) {
+               $SIG{CHLD} = \&sig_chld;
+               $waitedpid = wait;
+       }
        dbg('connect', "caught $pid");
 }
 
        dbg('connect', "caught $pid");
 }
 
@@ -401,11 +404,12 @@ if ($call eq $mycall) {
 
 $stdout->autoflush(1);
 
 
 $stdout->autoflush(1);
 
-$SIG{'INT'} = \&sig_term;
-$SIG{'TERM'} = \&sig_term;
-$SIG{'HUP'} = \&sig_term;
-$SIG{'CHLD'} = \&sig_chld;
-$SIG{'ALRM'} = \&timeout;
+unless ($^O =~ /^MS/i) {
+       $SIG{'INT'} = \&sig_term;
+       $SIG{'TERM'} = \&sig_term;
+       $SIG{'HUP'} = \&sig_term;
+       $SIG{'CHLD'} = \&sig_chld;
+}
 
 dbgadd('connect');
 
 
 dbgadd('connect');
 
@@ -520,7 +524,7 @@ if ($ssid) {
 }
 
 
 }
 
 
-$conn = Msg->connect("$clusteraddr", $clusterport, \&rec_socket);
+$conn = IntMsg->connect("$clusteraddr", $clusterport, \&rec_socket);
 if (! $conn) {
        if (-r "$data/offline") {
                open IN, "$data/offline" or die;
 if (! $conn) {
        if (-r "$data/offline") {
                open IN, "$data/offline" or die;
@@ -541,7 +545,7 @@ Msg->set_event_handler($stdin, "read" => \&rec_stdin);
 
 for (;;) {
        my $t;
 
 for (;;) {
        my $t;
-       Msg->event_loop(1, 1);
+       Msg->event_loop(1, 0.1);
        $t = time;
        if ($t > $lasttime) {
                if ($outqueue) {
        $t = time;
        if ($t > $lasttime) {
                if ($outqueue) {
index 995500ca934239b4ca18ec6ba580d9ef96f8e3c8..d34a3337b270cf27a1c912e2a17594f1047ed77e 100755 (executable)
@@ -39,6 +39,8 @@ BEGIN {
 }
 
 use Msg;
 }
 
 use Msg;
+use IntMsg;
+use ExtMsg;
 use DXVars;
 use DXDebug;
 use DXLog;
 use DXVars;
 use DXDebug;
 use DXLog;
@@ -82,15 +84,9 @@ $version = "1.47";                           # the version no of the software
 $starttime = 0;                 # the starting time of the cluster   
 $lockfn = "cluster.lock";       # lock file name
 @outstanding_connects = ();     # list of outstanding connects
 $starttime = 0;                 # the starting time of the cluster   
 $lockfn = "cluster.lock";       # lock file name
 @outstanding_connects = ();     # list of outstanding connects
-      
-# handle disconnections
-sub disconnect
-{
-       my $dxchan = shift;
-       return if !defined $dxchan;
-       $dxchan->disconnect();
-}
+@listeners = ();                               # list of listeners
 
 
+      
 # send a message to call on conn and disconnect
 sub already_conn
 {
 # send a message to call on conn and disconnect
 sub already_conn
 {
@@ -102,7 +98,7 @@ sub already_conn
        dbg('chan', "-> Z $call bye\n");
        $conn->send_now("Z$call|bye"); # this will cause 'client' to disconnect
        sleep(1);
        dbg('chan', "-> Z $call bye\n");
        $conn->send_now("Z$call|bye"); # this will cause 'client' to disconnect
        sleep(1);
-       $conn->disconnect();
+       $conn->disconnect;
 }
 
 # handle incoming messages
 }
 
 # handle incoming messages
@@ -205,19 +201,15 @@ sub cease
        # disconnect nodes
        foreach $dxchan (DXChannel->get_all()) {
                next unless $dxchan->is_node;
        # disconnect nodes
        foreach $dxchan (DXChannel->get_all()) {
                next unless $dxchan->is_node;
-               disconnect($dxchan) unless $dxchan == $DXProt::me;
+           $dxchan->disconnect unless $dxchan == $DXProt::me;
        }
        Msg->event_loop(1, 0.05);
        Msg->event_loop(1, 0.05);
        }
        Msg->event_loop(1, 0.05);
        Msg->event_loop(1, 0.05);
-       Msg->event_loop(1, 0.05);
-       Msg->event_loop(1, 0.05);
-       Msg->event_loop(1, 0.05);
-       Msg->event_loop(1, 0.05);
 
        # disconnect users
        foreach $dxchan (DXChannel->get_all()) {
                next if $dxchan->is_node;
 
        # disconnect users
        foreach $dxchan (DXChannel->get_all()) {
                next if $dxchan->is_node;
-               disconnect($dxchan) unless $dxchan == $DXProt::me;
+               $dxchan->disconnect unless $dxchan == $DXProt::me;
        }
        Msg->event_loop(1, 0.05);
        Msg->event_loop(1, 0.05);
        }
        Msg->event_loop(1, 0.05);
        Msg->event_loop(1, 0.05);
@@ -230,7 +222,12 @@ sub cease
 
        # close all databases
        DXDb::closeall;
 
        # close all databases
        DXDb::closeall;
-       
+
+       # close all listeners
+       for (@listeners) {
+               $_->close_server;
+       }
+
        dbg('chan', "DXSpider version $version ended");
        Log('cluster', "DXSpider V$version stopped");
        dbgclose();
        dbg('chan', "DXSpider version $version ended");
        Log('cluster', "DXSpider V$version stopped");
        dbgclose();
@@ -276,10 +273,10 @@ sub process_inqueue
                die "\$user not defined for $call" if !defined $user;
                # normal input
                $dxchan->normal($line);
                die "\$user not defined for $call" if !defined $user;
                # normal input
                $dxchan->normal($line);
-               disconnect($dxchan) if ($dxchan->{state} eq 'bye');
+               $dxchan->disconnect if ($dxchan->{state} eq 'bye');
        } elsif ($sort eq 'Z') {
                $dxchan->conn(undef);
        } elsif ($sort eq 'Z') {
                $dxchan->conn(undef);
-               disconnect($dxchan);
+               $dxchan->disconnect;
        } elsif ($sort eq 'D') {
                ;                       # ignored (an echo)
        } else {
        } elsif ($sort eq 'D') {
                ;                       # ignored (an echo)
        } else {
@@ -331,16 +328,25 @@ dbg('err', "loading user file system ...");
 DXUser->init($userfn, 1);
 
 # start listening for incoming messages/connects
 DXUser->init($userfn, 1);
 
 # start listening for incoming messages/connects
-dbg('err', "starting listener ...");
-Msg->new_server("$clusteraddr", $clusterport, \&login);
+use Listeners;
+
+dbg('err', "starting listeners ...");
+push @listeners, IntMsg->new_server("$clusteraddr", $clusterport, \&login);
+dbg('err', "Internal port: $clusteraddr $clusterport");
+for (@main::listen) {
+       push @listeners, ExtMsg->new_server($_->[0], $_->[1], \&login);
+       dbg('err', "External Port: $_->[0] $_->[1]");
+}
 
 # load bad words
 dbg('err', "load badwords: " . (BadWords::load or "Ok"));
 
 # prime some signals
 unless ($^O =~ /^MS/) {
 
 # load bad words
 dbg('err', "load badwords: " . (BadWords::load or "Ok"));
 
 # prime some signals
 unless ($^O =~ /^MS/) {
-       $SIG{INT} = \&cease;
-       $SIG{TERM} = \&cease;
+       unless ($DB::VERSION) {
+               $SIG{INT} = \&cease;
+               $SIG{TERM} = \&cease;
+       }
        $SIG{HUP} = 'IGNORE';
        $SIG{CHLD} = sub { $zombies++ };
        
        $SIG{HUP} = 'IGNORE';
        $SIG{CHLD} = sub { $zombies++ };
        
index 4d7040cf2374f9c97b58b76dd9484be131afbd9d..208eb90a4b8969fd8fbd8e95d4386fdd07f39f85 100755 (executable)
@@ -26,6 +26,7 @@ BEGIN {
 }
 
 use Msg;
 }
 
 use Msg;
+use IntMsg;
 use DXVars;
 use DXDebug;
 use DXUtil;
 use DXVars;
 use DXDebug;
 use DXUtil;
@@ -422,7 +423,7 @@ if ($call eq $mycall) {
        exit(0);
 }
 
        exit(0);
 }
 
-$conn = Msg->connect("$clusteraddr", $clusterport, \&rec_socket);
+$conn = IntMsg->connect("$clusteraddr", $clusterport, \&rec_socket);
 if (! $conn) {
        if (-r "$data/offline") {
                open IN, "$data/offline" or die;
 if (! $conn) {
        if (-r "$data/offline") {
                open IN, "$data/offline" or die;
@@ -437,8 +438,11 @@ if (! $conn) {
 }
 
 
 }
 
 
-$SIG{'INT'} = \&sig_term;
-$SIG{'TERM'} = \&sig_term;
+unless ($DB::VERSION) {
+       $SIG{'INT'} = \&sig_term;
+       $SIG{'TERM'} = \&sig_term;
+}
+
 #$SIG{'WINCH'} = \&do_resize;
 $SIG{'HUP'} = \&sig_term;
 
 #$SIG{'WINCH'} = \&do_resize;
 $SIG{'HUP'} = \&sig_term;
 
@@ -455,7 +459,7 @@ Msg->set_event_handler(\*STDIN, "read" => \&rec_stdin);
 my $lastmin = 0;
 for (;;) {
        my $t;
 my $lastmin = 0;
 for (;;) {
        my $t;
-       Msg->event_loop(1, 1);
+       Msg->event_loop(1, 0.1);
        $t = time;
        if ($t > $lasttime) {
                my ($min)= (gmtime($t))[1];
        $t = time;
        if ($t > $lasttime) {
                my ($min)= (gmtime($t))[1];