From: minima <minima>
Date: Fri, 2 Mar 2001 13:37:03 +0000 (+0000)
Subject: The guts of the new Msg system
X-Git-Tag: R_1_47_PRE1~1
X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=commitdiff_plain;h=04ff9a5a6bf8f3ab55bc2922b76a44e2798af8ee;p=spider.git

The guts of the new Msg system
---

diff --git a/perl/ExtMsg.pm b/perl/ExtMsg.pm
new file mode 100644
index 00000000..dafa8566
--- /dev/null
+++ b/perl/ExtMsg.pm
@@ -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
index 00000000..150ec91a
--- /dev/null
+++ b/perl/IntMsg.pm
@@ -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
index 00000000..cc83051a
--- /dev/null
+++ b/perl/Listeners.pm
@@ -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],
+		  );
+
diff --git a/perl/Msg.pm b/perl/Msg.pm
index f4f013be..74c64ec7 100644
--- a/perl/Msg.pm
+++ b/perl/Msg.pm
@@ -15,12 +15,15 @@ use IO::Select;
 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();
+$now = time;
+@timerchain = ();
+
 my $blocking_supported = 0;
 
 BEGIN {
@@ -31,13 +34,33 @@ BEGIN {
     $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 {
-    my ($pkg, $to_host, $to_port,$rcvd_notification_proc) = @_;
-    
+    my ($pkg, $to_host, $to_port, $rproc) = @_;
+
     # Create a new internet socket
-    
     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
-    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);
     }
-    return bless $conn, $pkg;
+    return $conn;
 }
 
 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);
@@ -70,31 +97,28 @@ sub disconnect {
 
 sub send_now {
     my ($conn, $msg) = @_;
-    _enqueue ($conn, $msg);
+    $conn->enqueue($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)});
 }
 
-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);
-    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
@@ -179,21 +203,31 @@ sub handle_send_err {
 #-----------------------------------------------------------------
 # Receive side routines
 
-my ($g_login_proc,$g_pkg);
-my $main_socket = 0;
 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) = @_;
-    
-    $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);
-    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
@@ -209,13 +243,16 @@ sub _rcv {                     # Complement to _send
 	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;
 				}
+				push @{$conn->{inqueue}}, @lines if @lines;
 			} else {
 				$conn->{msg} .= $msg;
 			}
@@ -231,29 +268,21 @@ sub _rcv {                     # Complement to _send
 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
@@ -263,9 +292,9 @@ sub _new_client {
 
 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) {
-        # 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());
-        ($rset, $wset) =
+        
+		($rset, $wset) =
             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};
         }
+
+		# 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;
         }
diff --git a/perl/client.pl b/perl/client.pl
index ddf3fd3b..57a5d45b 100755
--- a/perl/client.pl
+++ b/perl/client.pl
@@ -39,6 +39,7 @@ BEGIN {
 }
 
 use Msg;
+use IntMsg;
 use DXVars;
 use DXDebug;
 use DXUtil;
@@ -78,8 +79,10 @@ sub sig_term
 # 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");
 }
 
@@ -401,11 +404,12 @@ if ($call eq $mycall) {
 
 $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');
 
@@ -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;
@@ -541,7 +545,7 @@ Msg->set_event_handler($stdin, "read" => \&rec_stdin);
 
 for (;;) {
 	my $t;
-	Msg->event_loop(1, 1);
+	Msg->event_loop(1, 0.1);
 	$t = time;
 	if ($t > $lasttime) {
 		if ($outqueue) {
diff --git a/perl/cluster.pl b/perl/cluster.pl
index 995500ca..d34a3337 100755
--- a/perl/cluster.pl
+++ b/perl/cluster.pl
@@ -39,6 +39,8 @@ BEGIN {
 }
 
 use Msg;
+use IntMsg;
+use ExtMsg;
 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
-      
-# 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
 {
@@ -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);
-	$conn->disconnect();
+	$conn->disconnect;
 }
 
 # handle incoming messages
@@ -205,19 +201,15 @@ sub cease
 	# 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);
 
 	# 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);
@@ -230,7 +222,12 @@ sub cease
 
 	# 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();
@@ -276,10 +273,10 @@ sub process_inqueue
 		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);
-		disconnect($dxchan);
+		$dxchan->disconnect;
 	} 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
-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/) {
-	$SIG{INT} = \&cease;
-	$SIG{TERM} = \&cease;
+	unless ($DB::VERSION) {
+		$SIG{INT} = \&cease;
+		$SIG{TERM} = \&cease;
+	}
 	$SIG{HUP} = 'IGNORE';
 	$SIG{CHLD} = sub { $zombies++ };
 	
diff --git a/perl/console.pl b/perl/console.pl
index 4d7040cf..208eb90a 100755
--- a/perl/console.pl
+++ b/perl/console.pl
@@ -26,6 +26,7 @@ BEGIN {
 }
 
 use Msg;
+use IntMsg;
 use DXVars;
 use DXDebug;
 use DXUtil;
@@ -422,7 +423,7 @@ if ($call eq $mycall) {
 	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;
@@ -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;
 
@@ -455,7 +459,7 @@ Msg->set_event_handler(\*STDIN, "read" => \&rec_stdin);
 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];