X-Git-Url: http://www.dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FAsyncMsg.pm;h=f8cd2f9f3b60bea14be79d24029a69049b685a9c;hb=refs%2Fheads%2Fstaging;hp=f7b2bc0ddf1102520da25559887bd38456cda20a;hpb=564b5b3a0c2fa40f00e015f8b05f3a87ea4e7e26;p=spider.git

diff --git a/perl/AsyncMsg.pm b/perl/AsyncMsg.pm
index f7b2bc0d..f8cd2f9f 100644
--- a/perl/AsyncMsg.pm
+++ b/perl/AsyncMsg.pm
@@ -27,17 +27,24 @@ $deftimeout = 15;
 
 my %outstanding;
 
-#
-# standard http get handler
-#
-sub handle_get
+sub new 
 {
-	my $conn = shift;
-	my $msg = shift;
+	my $pkg = shift;
+	my $call = shift;
+	my $handler = shift;
+	
+	my $conn = $pkg->SUPER::new($handler);
+	$conn->{caller} = ref $call ? $call->call : $call;
 
-	my $state = $conn->{state};
+	# make it persistent
+	$outstanding{$conn} = $conn;
 	
-	dbg("asyncmsg: $msg") if isdbg('async');
+	return $conn;
+}
+
+sub handle_getpost
+{
+	my ($conn, $ua, $tx) = @_;
 
 	# no point in going on if there is no-one wanting the output anymore
 	my $dxchan = DXChannel::get($conn->{caller});
@@ -46,42 +53,12 @@ sub handle_get
 		return;
 	}
 	
-	if ($state eq 'waitreply') {
-		# look at the reply code and decide whether it is a success
-		my ($http, $code, $ascii) = $msg =~ m|(HTTP/\d\.\d)\s+(\d+)\s+(.*)|;
-		if ($code == 200) {
-			# success
-			$conn->{state} = 'waitblank';
-		} elsif ($code == 302) {
-			# redirect
-			$conn->{state} = 'waitlocation';
-		} else {
-			$dxchan->send("$code $ascii");
-			$conn->disconnect;
-		} 
-	} elsif ($state  eq 'waitlocation') {
-		my ($path) = $msg =~ m|Location:\s*(.*)|;
-		if ($path) {
-			my @uri = split m|/+|, $path;
-			if ($uri[0] eq 'http:') {
-				shift @uri;
-				my $host = shift @uri;
-				my $newpath = '/' . join('/', @uri);
-				$newpath .= '/' if $path =~ m|/$|;
-				_getpost(ref $conn, $conn->{asyncsort}, $conn->{caller}, $host, 80, $newpath, @{$conn->{asyncargs}});
-			} elsif ($path =~ m|^/|) {
-				_getpost(ref $conn, $conn->{asyncsort}, $conn->{caller}, $conn->{peerhost}, $conn->{peerport}, $path,
-						 @{$conn->{asyncargs}});
-			}
-			delete $conn->{on_disconnect};
-			$conn->disconnect;
-		}
-	} elsif ($state eq 'waitblank') {
-		unless ($msg) {
-			$conn->{state} = 'indata';
-		}
-	} elsif ($conn->{state} eq 'indata') {
-		if (my $filter = $conn->{filter}) {
+	my @lines = split qr{\r?\n}, $tx->res->body;
+	
+	foreach my $msg(@lines) {
+		dbg("AsyncMsg: $conn->{_asstate} $msg") if isdbg('async');
+		
+		if (my $filter = $conn->{_asfilter}) {
 			no strict 'refs';
 			# this will crash if the command has been redefined and the filter is a
 			# function defined there whilst the request is in flight,
@@ -92,43 +69,8 @@ sub handle_get
 			$dxchan->send("$prefix$msg");
 		}
 	}
-}
-
-# 
-# simple raw handler
-#
-# Just outputs everything
-#
-sub handle_raw
-{
-	my $conn = shift;
-	my $msg = shift;
-
-	# no point in going on if there is no-one wanting the output anymore
-	my $dxchan = DXChannel::get($conn->{caller});
-	unless ($dxchan) {
-		$conn->disconnect;
-		return;
-	}
-
-	# send out the data
-	my $prefix = $conn->{prefix} || '';
-	$dxchan->send("$prefix$msg");
-}
-
-sub new 
-{
-	my $pkg = shift;
-	my $call = shift;
-	my $handler = shift;
 	
-	my $conn = $pkg->SUPER::new($handler);
-	$conn->{caller} = ref $call ? $call->call : $call;
-
-	# make it persistent
-	$outstanding{$conn} = $conn;
-	
-	return $conn;
+	$conn->disconnect;
 }
 
 # This does a http get on a path on a host and
@@ -158,41 +100,77 @@ sub _getpost
 	my $sort = shift;
 	my $call = shift;
 	my $host = shift;
-	my $port = shift;
 	my $path = shift;
 	my %args = @_;
 	
 
-	my $conn = $pkg->new($call, \&handle_get);
-	$conn->{asyncargs} = [@_];
-	$conn->{state} = 'waitreply';
-	$conn->{filter} = delete $args{filter} if exists $args{filter};
+	my $conn = $pkg->new($call);
+	$conn->{_asargs} = [@_];
+	$conn->{_asstate} = 'waitreply';
+	$conn->{_asfilter} = delete $args{filter} if exists $args{filter};
 	$conn->{prefix} = delete $args{prefix} if exists $args{prefix};
+	$conn->{prefix} ||= '';
 	$conn->{on_disconnect} = delete $args{on_disc} || delete $args{on_disconnect};
 	$conn->{path} = $path;
-	$conn->{asyncsort} = $sort;
+	$conn->{host} = $conn->{peerhost} = $host;
+	$conn->{port} = $conn->{peerport} = delete $args{port} || 80;
+	$conn->{sort} = 'outgoing';
+	$conn->{_assort} = $sort;
+	$conn->{csort} = 'http';
+
+	my $data = delete $args{data};
+
+	my $ua =  Mojo::UserAgent->new;
+	my $s;
+	$s .= $host;
+	$s .= ":$port" unless $conn->{port} == 80;
+	$s .= $path;
+	dbg("AsyncMsg: $sort $s") if isdbg('async');
 	
-	$r = $conn->connect($host, $port);
-	if ($r) {
-		dbg("Sending '$sort $path HTTP/1.0'") if isdbg('async');
-		$conn->send_later("$sort $path HTTP/1.0\n");
-
-		my $h = delete $args{Host} || $host;
-		my $u = delete $args{'User-Agent'} || "DxSpider;$main::version;$main::build;$^O;$main::mycall"; 
-		my $d = delete $args{data};
-		
-	    $conn->send_later("Host: $h\n");
-		$conn->send_later("User-Agent: $u\n");
-		while (my ($k,$v) = each %args) {
-			$conn->send_later("$k: $v\n");
-		}
-		$conn->send_later("\n$d") if defined $d;
-		$conn->send_later("\n");
-	}
+	my $tx = $ua->build_tx($sort => $s);
+	$ua->on(error => sub { $conn->_error(@_); });
+#	$tx->on(error => sub { $conn->_error(@_); });
+#	$tx->on(finish => sub { $conn->disconnect; });
+
+	$ua->on(start => sub {
+				my ($ua, $tx) = @_;
+				while (my ($k, $v) = each %args) {
+					dbg("AsyncMsg: attaching header $k: $v") if isdbg('async');
+					$tx->req->headers->header($k => $v);
+				}
+				if (defined $data) {
+					dbg("AsyncMsg: body ='$data'") if isdbg('async'); 
+					$tx->req->body($data);
+				}
+			});
 	
-	return $r ? $conn : undef;
+
+	$ua->start($tx => sub { $conn->handle_getpost(@_) }); 
+
+	
+	$conn->{mojo} = $ua;
+	return $conn if $tx;
+
+	$conn->disconnect;
+	return undef;
+}
+
+sub _dxchan_send
+{
+	my $conn = shift;
+	my $msg = shift;
+	my $dxchan = DXChannel::get($conn->{caller});
+	$dxchan->send($msg) if $dxchan;
 }
 
+sub _error
+{
+	my ($conn, $e, $err);
+	dbg("Async: $conn->host:$conn->port path $conn->{path} error $err") if isdbg('chan');
+	$conn->_dxchan_send("$conn->{prefix}$msg");
+	$conn->disconnect;
+}
+	
 sub get
 {
 	my $pkg = shift;
@@ -225,12 +203,51 @@ sub raw
 	my %args = @_;
 
 	my $handler = delete $args{handler} || \&handle_raw;
+	
 	my $conn = $pkg->new($call, $handler);
 	$conn->{prefix} = delete $args{prefix} if exists $args{prefix};
-	$r = $conn->connect($host, $port);
+	$conn->{prefix} ||= '';
+	$conn->{on_disconnect} = delete $args{on_disc} || delete $args{on_disconnect};
+	$r = $conn->connect($host, $port, on_connect => &_on_raw_connect);
 	return $r ? $conn : undef;
 }
 
+# 
+# simple raw handler
+#
+# Just outputs everything
+#
+sub handle_raw
+{
+	my $conn = shift;
+	my $msg = shift;
+
+	# no point in going on if there is no-one wanting the output anymore
+	my $dxchan = DXChannel::get($conn->{caller});
+	unless ($dxchan) {
+		$conn->disconnect;
+		return;
+	}
+
+	# send out the data
+	$dxchan->send("$conn->{prefix}$msg");
+}
+
+
+sub _on_raw_connect
+{
+	my $conn = shift;
+	my $handle = shift;
+	dbg("AsyncMsg: Connected $conn->{cnum} to $conn->{host}:$conn->{port}") if isdbg('async');
+}
+
+sub _on_error
+{
+	my $conn = shift;
+	my $msg = shift;
+	dbg("AsyncMsg: ***Connect $conn->{cnum} Failed to $conn->{host}:$conn->{port} $!") if isdbg('async');	
+}
+
 sub connect
 {
 	my $conn = shift;
@@ -238,13 +255,8 @@ sub connect
 	my $port = shift;
 	
 	# start a connection
-	my $r = $conn->SUPER::connect($host, $port);
-	if ($r) {
-		dbg("AsyncMsg: Connected $conn->{cnum} to $host $port") if isdbg('async');
-	} else {
-		dbg("AsyncMsg: ***Connect $conn->{cnum} Failed to $host $port $!") if isdbg('async');
-	}
-	
+	my $r = $conn->SUPER::connect($host, $port, @_);
+
 	return $r;
 }
 
@@ -256,13 +268,28 @@ sub disconnect
 		my $dxchan = DXChannel::get($conn->{caller});
 		if ($dxchan) {
 			no strict 'refs';
-			$ondisc->($conn, $dxchan)
+			$ondisc->($conn, $dxchan);
+			delete $conn->{on_disconnect};
 		}
 	}
+	delete $conn->{mojo};
 	delete $outstanding{$conn};
 	$conn->SUPER::disconnect;
 }
 
+sub _send_later
+{
+	my $conn = shift;
+	my $m = shift;
+	
+	if (isdbg('async')) {
+		my $s = $m;
+		$s =~ s/([\%\x00-\x1f\x7f-\xff])/sprintf("%%%02X", ord($1))/eg;
+		dbg("AsyncMsg: send $s");
+	}
+	$conn->send_later($m);
+}
+
 sub DESTROY
 {
 	my $conn = shift;