File Coverage

blib/lib/IO/Lambda/Socket.pm

Criterion	Covered	Total	%
statement	68	85	80.0
branch	19	48	39.5
condition	2	4	50.0
subroutine	15	17	88.2
pod			n/a
total	104	154	67.5

# $Id: Socket.pm,v 1.12 2010/03/26 20:45:45 dk Exp $
use strict;
use warnings;

package IO::Lambda::Socket;
use Carp qw(croak);
use Socket;
use Exporter;
use IO::Lambda qw(:all :dev);
use Time::HiRes qw(time);
use vars       qw(@ISA @EXPORT_OK %EXPORT_TAGS);
@ISA         = qw(Exporter);
%EXPORT_TAGS = (all => \@EXPORT_OK);
@EXPORT_OK   = qw(connect accept send recv);
use subs       qw(connect accept send recv);

sub connect(&)
{
	return this-> override_handler('connect', \&connect, shift)
		if this-> {override}->{connect};

	my $cb = _subname connect => shift;
	my ($socket, $deadline) = context;

	return this-> add_constant( $cb, \&connect, "Bad socket") unless $socket;

	my $w32 = $^O eq 'MSWin32';

	this-> watch_io(
		IO_WRITE | ( $w32 ? IO_EXCEPTION : 0),
		$socket, $deadline,
		sub {
			shift-> set_frame( \&connect, $cb, $socket, $deadline);

			my @param;
			unless ( $_[0]) {
				@param = ('timeout');
			} else {
				my $e = $w32 ? \ $^E : \ $!;
				$$e = unpack('i', getsockopt( $socket, SOL_SOCKET, SO_ERROR));
				@param = ($$e) if $$e;
			}
			local *__ANON__ = "IO::Lambda::Socket::connect::callback";
			$cb ? $cb-> (@param) : @param;
		}
	);
}

sub accept(&)
{
	return this-> override_handler('accept', \&accept, shift)
		if this-> {override}->{accept};

	my $cb = _subname accept => shift;
	my ($socket, $deadline) = context;

	return this-> add_constant( $cb, \&connect, "Bad socket") unless $socket;

	this-> watch_io(
		IO_READ, $socket, $deadline,
		sub {
			shift-> set_frame( \&accept, $cb, $socket, $deadline);

			my @param;
			unless ( $_[0]) {
				@param = ('timeout');
			} else {
				my $h = IO::Handle-> new;
				@param = (
					CORE::accept( $h, $socket) ?
					($h) : ($!)
				);
			}
			local *__ANON__ = "IO::Lambda::Socket::accept::callback";
			$cb ? $cb-> (@param) : @param;
		}
	);
}

# recv($fh, $length, $flags, $deadline) -> (address,msg|undef,error)
sub recv(&)
{
	return this-> override_handler('recv', \&recv, shift)
		if this-> {override}->{recv};

	my $cb = _subname( recv => shift );
	my ($socket, $length, $flags, $deadline) = context;

	return this-> add_constant( $cb, \&recv, undef, "Bad socket")
		unless $socket;

	this-> watch_io(
		IO_READ, $socket, $deadline,
		sub {
			shift-> set_frame( \&recv, $cb, $socket, $length, $flags, $deadline);

			my @param;
			unless ( $_[0]) {
				@param = ('timeout');
			} else {
				my $buf = '';
				my $r = CORE::recv(
					$socket, $buf, $length,
					$flags || 0
				);
				if ( defined($r)) {
					@param = defined($r) ? ($r,$buf) : (undef,$!);
				} else {
					@param = ( undef, $!);
				}
			}
			local *__ANON__ = "IO::Lambda::Socket::recv::callback";
			$cb ? $cb-> (@param) : @param;
		}
	);
}

# send($fh, $msg, $flags, $to, $deadline) -> ioresult
sub send(&)
{
	return this-> override_handler('send', \&send, shift)
		if this-> {override}->{send};

	my $cb = _subname send => shift;
	my ($socket, $msg, $flags, $to, $deadline) = context;

	return this-> add_constant( $cb, \&send, undef, "Bad socket")
		unless $socket;

	this-> watch_io(
		IO_WRITE, $socket, $deadline,
		sub {
			shift-> set_frame( \&recv, $cb, $socket, $msg, $flags, $to, $deadline);

			my @param;
			unless ( $_[0]) {
				@param = ('timeout');
			} else {
				$flags ||= 0;
				my $r = defined($to) ?
					CORE::send($socket, $msg, $flags, $to) :
					CORE::send($socket, $msg, $flags);
				@param = defined($r) ? ($r) : (undef,$!);
			}
			local *__ANON__ = "IO::Lambda::Socket::send::callback";
			$cb ? $cb-> (@param) : @param;
		}
	);
}

1;

__DATA__