File Coverage

blib/lib/AnyEvent/Handle/UDP.pm

Criterion	Covered	Total	%
statement	127	172	73.8
branch	29	70	41.4
condition	8	26	30.7
subroutine	39	48	81.2
pod	22	25	88.0
total	225	341	65.9

package AnyEvent::Handle::UDP;
$AnyEvent::Handle::UDP::VERSION = '0.049';
use strict;
use warnings;

use AnyEvent ();
use AnyEvent::Util ();
use AnyEvent::Socket ();

use Carp ();
use Errno ();
use Scalar::Util ();
use Socket ();
use Symbol ();

my $subname = eval { require Sub::Name } ? \&Sub::Name::subname : sub { $_[1] };
my %non_fatal = map { ( $_ => 1 ) } Errno::EAGAIN, Errno::EWOULDBLOCK, Errno::EINTR;

sub new {
	my ($class, %args) = @_;
	my $self = bless {
		on_recv      => $args{on_recv} || Carp::croak('on_recv not given'),
		reuse_addr   => exists $args{reuse_addr} ? !!$args{reuse_addr} : 1,
		receive_size => $args{receive_size} || 1500,
		family       => $args{family}       || 0,
		autoflush    => $args{autoflush}    || 0,
		buffers      => [],
	}, $class;
	$self->{$_} = $args{$_} for grep { exists $args{$_} } qw/on_drain on_error on_timeout on_rtimeout on_wtimeout/;
	$self->{$_} = AE::now() for qw/activity ractivity wactivity/;

	$self->{fh} = bless Symbol::gensym(), 'IO::Socket';
	$self->_bind_to($self->{fh}, $args{bind}) if exists $args{bind};
	$self->_connect_to($self->{fh}, $args{connect}) if exists $args{connect};

	$self->$_($args{$_}) for grep { exists $args{$_} } qw/timeout rtimeout wtimeout/;

	$self->_drained;
	return $self;
}

sub _insert {
	my ($name, $sub) = @_;
	no strict 'refs';
	*{$name} = $subname->($name, $sub);
}

for my $name (qw/on_recv on_error on_timeout on_rtimeout on_wtimeout autoflush receive_size/) {
	_insert($name, sub {
		my $self = shift;
		$self->{$name} = shift if @_;
		return $self->{$name};
	});
}

for my $name (qw/sockname peername/) {
	_insert($name, sub {
		my $self = shift;
		return $self->{fh}->$name;
	});
}

sub fh {
	my $self = shift;
	return $self->{fh};
}

sub on_drain {
	my $self = shift;
	if (@_) {
		$self->{on_drain} = shift;
		$self->_drained if not @{ $self->{buffers} };
	}
	return $self->{on_drain};
}

sub _drained {
	my $self = shift;
	$self->{on_drain}->($self) if defined $self->{on_drain};
}

for my $dir ('', 'r', 'w') {
	my $timeout = "${dir}timeout";
	my $activity = "${dir}activity";
	my $on_timeout = "on_$timeout";
	my $timer = "${dir}timer";
	my $clear_timeout = "clear_$timeout";
	my $timeout_reset = "${timeout}_reset";

	my $callback;
	$callback = sub {
		my $self = shift;
		if (not exists $self->{$timeout} or not $self->{fh}) {
			delete $self->{$timer};
			return;
		}
		my $now = AE::now;
		my $after = $self->{$activity} + $self->{$timeout} - $now;
		if ($after <= 0) {
			$self->{$activity} = $now;
			my $time = $self->{$on_timeout};
			my $error = do { local $! = Errno::ETIMEDOUT; "$!" };
			$time ? $time->($self) : $self->_error->(0, $error);
			return if not exists $self->{$timeout};
		}
		Scalar::Util::weaken($self);
		return if not $self;
		$self->{$timer} = AE::timer($after, 0, sub {
			delete $self->{$timer};
			$callback->($self);
		});
	};

	_insert($timeout, sub {
		my $self = shift;
		if (@_) {
			my $value = shift;
			$self->{$timeout} = $value;
			if ($value == 0) {
				delete $self->{$timer};
				delete $self->{$timeout};
				return;
			}
			else {
				$callback->($self);
			}
		}
		return $self->{$timeout};
	});

	_insert($clear_timeout, sub {
		my $self = shift;
		delete $self->{$timeout};
		return;
	});

	_insert($timeout_reset, sub {
		my $self = shift;
		$self->{$activity} = AE::now;
	});
}

sub bind_to {
	my ($self, $addr) = @_;
	return $self->_bind_to($self->{fh}, $addr);
}

my $add_reader = sub {
	my $self = shift;
	$self->{reader} = AE::io($self->{fh}, 0, sub {
		while (defined (my $addr = recv $self->{fh}, my ($buffer), $self->{receive_size}, 0)) {
			$self->timeout_reset;
			$self->rtimeout_reset;
			$self->{on_recv}->($buffer, $self, $addr);
		}
		$self->_error(1, "Couldn't recv: $!") if not $non_fatal{$! + 0};
		return;
	});
};

sub _bind_to {
	my ($self, $fh, $addr) = @_;
	my $bind_to = sub {
		my ($domain, $type, $proto, $sockaddr) = @_;
		if (!Scalar::Util::openhandle($fh)) {
			socket $fh, $domain, $type, $proto or redo;
			AnyEvent::Util::fh_nonblocking $fh, 1;
			setsockopt $fh, Socket::SOL_SOCKET, Socket::SO_REUSEADDR, 1 or $self->_error(1, "Couldn't set so_reuseaddr: $!") if $self->{reuse_addr};
			$add_reader->($self);
		}
		bind $fh, $sockaddr or $self->_error(1, "Could not bind: $!");
	};
	if (ref $addr) {
		my ($host, $port) = @{$addr};
		_on_addr($self, $fh, $host, $port, $bind_to);
	}
	else {
		$bind_to->(Socket::sockaddr_family($addr), Socket::SOCK_DGRAM, 0, $addr);
	}
	return;
}

sub connect_to {
	my ($self, $addr) = @_;
	return $self->_connect_to($self->{fh}, $addr);
}

sub _connect_to {
	my ($self, $fh, $addr) = @_;
	my $connect_to = sub {
		my ($domain, $type, $proto, $sockaddr) = @_;
		if (!Scalar::Util::openhandle($fh)) {
			socket $fh, $domain, $type, $proto or redo;
			AnyEvent::Util::fh_nonblocking $fh, 1;
			$add_reader->($self);
		}
		connect $fh, $sockaddr or $self->_error(1, "Could not connect: $!");
	};
	if (ref $addr) {
		my ($host, $port) = @{$addr};
		_on_addr($self, $fh, $host, $port, $connect_to);
	}
	else {
		$connect_to->(Socket::sockaddr_family($addr), Socket::SOCK_DGRAM, 0, $addr);
	}
	return;
}

my $get_family = sub {
	my ($self, $fh) = @_;
	return $self->{family} if !Scalar::Util::openhandle($fh) || !getsockname $fh;
	my $family = Socket::sockaddr_family(getsockname $fh);
	return +($family == Socket::AF_INET) ? 4 : ($family == Socket::AF_INET6) ? 6 : $self->{family};
};

sub _on_addr {
	my ($self, $fh, $host, $port, $on_success) = @_;

	AnyEvent::Socket::resolve_sockaddr($host, $port, 'udp', $get_family->($self, $fh), Socket::SOCK_DGRAM, sub {
		my @targets = @_;
		while (1) {
			my $target = shift @targets or $self->_error(1, "Could not resolve $host:$port");
			$on_success->(@{$target});
			last;
		}
	});
	return;
}

sub _error {
	my ($self, $fatal, $message) = @_;

	if (exists $self->{on_error}) {
		$self->{on_error}->($self, $fatal, $message);
		$self->destroy if $fatal;
	} else {
		$self->destroy;
		Carp::croak("AnyEvent::Handle::UDP uncaught error: $message");
	}
	return;
}

sub push_send {
	my ($self, $message, $to, $cv) = @_;
	$to = AnyEvent::Socket::pack_sockaddr($to->[1], defined $to->[0] ? AnyEvent::Socket::parse_address($to->[0]) : Socket::INADDR_ANY) if ref $to;
	$cv ||= AnyEvent::CondVar->new if defined wantarray;
	if ($self->{autoflush} and ! @{ $self->{buffers} }) {
		my $ret = $self->_send($message, $to, $cv);
		$self->_push_writer($message, $to, $cv) if not defined $ret and $non_fatal{$! + 0};
		$self->_drained if $ret;
	}
	else {
		$self->_push_writer($message, $to, $cv);
	}
	return $cv;
}

sub _send {
	my ($self, $message, $to, $cv) = @_;
	my $ret = defined $to ? send $self->{fh}, $message, 0, $to : send $self->{fh}, $message, 0;
	$self->_error(1, "Could not send: $!") if not defined $ret and !$non_fatal{$! + 0};
	if (defined $ret) {
		$self->timeout_reset;
		$self->wtimeout_reset;
		$cv->($ret) if defined $cv;
	}
	return $ret;
}

sub _push_writer {
	my ($self, $message, $to, $condvar) = @_;
	push @{ $self->{buffers} }, [ $message, $to, $condvar ];
	$self->{writer} ||= AE::io $self->{fh}, 1, sub {
		if (@{ $self->{buffers} }) {
			while (my $entry = shift @{ $self->{buffers} }) {
				my ($msg, $to, $cv) = @{$entry};
				my $ret = $self->_send($msg, $to, $cv);
				if (not defined $ret) {
					unshift @{ $self->{buffers} }, $entry if $self->{buffers};
					last;
				}
			}
		}
		else {
			$self->_drained;
		}
		delete $self->{writer} if not @{ $self->{buffers} };
	};
	return $condvar;
}

sub destroy {
	my $self = shift;
	delete @{$self}{qw/reader writer/};
	close $self->{fh};
	$self->timeout(0);
	$self->rtimeout(0);
	$self->wtimeout(0);
	return;
}

1;

# ABSTRACT: client/server UDP handles for AnyEvent

__END__