diff options
Diffstat (limited to 't')
62 files changed, 9674 insertions, 0 deletions
diff --git a/t/00use.t b/t/00use.t new file mode 100644 index 0000000..268ef56 --- /dev/null +++ b/t/00use.t @@ -0,0 +1,35 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; + +use_ok( "IO::Async::Notifier" ); +use_ok( "IO::Async::Handle" ); +use_ok( "IO::Async::Stream" ); +use_ok( "IO::Async::Timer" ); +use_ok( "IO::Async::Timer::Absolute" ); +use_ok( "IO::Async::Timer::Countdown" ); +use_ok( "IO::Async::Timer::Periodic" ); +use_ok( "IO::Async::Signal" ); +use_ok( "IO::Async::Listener" ); +use_ok( "IO::Async::Socket" ); +use_ok( "IO::Async::File" ); +use_ok( "IO::Async::FileStream" ); + +use_ok( "IO::Async::OS" ); + +use_ok( "IO::Async::Loop::Select" ); +use_ok( "IO::Async::Loop::Poll" ); + +use_ok( "IO::Async::Test" ); + +use_ok( "IO::Async::Function" ); +use_ok( "IO::Async::Resolver" ); + +use_ok( "IO::Async::Protocol" ); +use_ok( "IO::Async::Protocol::Stream" ); +use_ok( "IO::Async::Protocol::LineStream" ); + +done_testing; diff --git a/t/01timequeue.t b/t/01timequeue.t new file mode 100644 index 0000000..0b05f10 --- /dev/null +++ b/t/01timequeue.t @@ -0,0 +1,94 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Fatal; + +use IO::Async::Internals::TimeQueue; + +my $queue = IO::Async::Internals::TimeQueue->new; + +ok( defined $queue, '$queue defined' ); +isa_ok( $queue, "IO::Async::Internals::TimeQueue", '$queue isa IO::Async::Internals::TimeQueue' ); + +is( $queue->next_time, undef, '->next_time when empty is undef' ); + +ok( exception { $queue->enqueue( code => sub { "DUMMY" } ) }, + 'enqueue no time fails' ); + +ok( exception { $queue->enqueue( time => 123 ) }, + 'enqueue no code fails' ); + +ok( exception { $queue->enqueue( time => 123, code => 'HELLO' ) }, + 'enqueue code not CODE ref fails' ); + +$queue->enqueue( time => 1000, code => sub { "DUMMY" } ); +is( $queue->next_time, 1000, '->next_time after single enqueue' ); + +my $fired = 0; + +$queue->enqueue( time => 500, code => sub { $fired = 1; } ); +is( $queue->next_time, 500, '->next_time after second enqueue' ); + +my $count = $queue->fire( now => 700 ); + +is( $fired, 1, '$fired after fire at time 700' ); +is( $count, 1, '$count after fire at time 700' ); +is( $queue->next_time, 1000, '->next_time after fire at time 700' ); + +$count = $queue->fire( now => 900 ); + +is( $count, 0, '$count after fire at time 900' ); +is( $queue->next_time, 1000, '->next_time after fire at time 900' ); + +$count = $queue->fire( now => 1200 ); + +is( $count, 1, '$count after fire at time 1200' ); +is( $queue->next_time, undef, '->next_time after fire at time 1200' ); + +$queue->enqueue( time => 1300, code => sub{ $fired++; } ); +$queue->enqueue( time => 1301, code => sub{ $fired++; } ); + +$count = $queue->fire( now => 1400 ); + +is( $fired, 3, '$fired after fire at time 1400' ); +is( $count, 2, '$count after fire at time 1400' ); +is( $queue->next_time, undef, '->next_time after fire at time 1400' ); + +my $id = $queue->enqueue( time => 1500, code => sub { $fired++ } ); +$queue->enqueue( time => 1505, code => sub { $fired++ } ); + +is( $queue->next_time, 1500, '->next_time before cancel' ); + +$queue->cancel( $id ); + +is( $queue->next_time, 1505, '->next_time after cancel' ); + +$fired = 0; +$count = $queue->fire( now => 1501 ); + +is( $fired, 0, '$fired after fire at time 1501' ); +is( $count, 0, '$count after fire at time 1501' ); + +$count = $queue->fire( now => 1510 ); + +is( $fired, 1, '$fired after fire at time 1510' ); +is( $count, 1, '$count after fire at time 1510' ); + +# Performance for large collections +{ + foreach my $t ( 2000 .. 2100 ) { + $queue->enqueue( time => $t, code => sub {} ); + } + + foreach my $t ( 2000 .. 2100 ) { + $queue->next_time == $t or fail( "Failed for large collection - expected $t" ), last; + $queue->fire( now => $t ); + } + + ok( "Large collection" ); +} + +done_testing; diff --git a/t/02os.t b/t/02os.t new file mode 100644 index 0000000..b2d5a94 --- /dev/null +++ b/t/02os.t @@ -0,0 +1,170 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Fatal; + +use IO::Async::OS; + +use Socket qw( + AF_INET AF_INET6 AF_UNIX SOCK_STREAM SOCK_DGRAM SO_TYPE + pack_sockaddr_in pack_sockaddr_in6 pack_sockaddr_un inet_aton inet_pton + INADDR_ANY +); + +use POSIX qw( SIGTERM ); + +SKIP: { + skip "No IO::Socket::IP", 2 unless eval { require IO::Socket::IP }; + + my $S_inet = IO::Async::OS->socket( "inet", "stream" ); + isa_ok( $S_inet, "IO::Socket::IP", 'IO::Async::OS->socket("inet")' ); + + SKIP: { + skip "No AF_INET6", 1 unless eval { socket( my $fh, AF_INET6, SOCK_STREAM, 0 ) }; + + my $S_inet6 = IO::Async::OS->socket( "inet6", "stream" ); + isa_ok( $S_inet6, "IO::Socket::IP", 'IO::Async::OS->socket("inet6")' ); + } +} + +foreach my $family ( undef, "inet" ) { + my ( $S1, $S2 ) = IO::Async::OS->socketpair( $family, "stream" ) + or die "Could not socketpair - $!"; + + isa_ok( $S1, "IO::Socket", '$S1 isa IO::Socket' ); + isa_ok( $S2, "IO::Socket", '$S2 isa IO::Socket' ); + + # Due to a bug in IO::Socket, ->socktype may not be set + + is( $S1->sockopt(SO_TYPE), SOCK_STREAM, 'SO_TYPE of $S1 is SOCK_STREAM' ); + is( $S2->sockopt(SO_TYPE), SOCK_STREAM, 'SO_TYPE of $S2 is SOCK_STREAM' ); + + $S1->syswrite( "Hello" ); + is( do { my $b; $S2->sysread( $b, 8192 ); $b }, "Hello", '$S1 --writes-> $S2' ); + + $S2->syswrite( "Goodbye" ); + is( do { my $b; $S1->sysread( $b, 8192 ); $b }, "Goodbye", '$S2 --writes-> $S1' ); + + ( $S1, $S2 ) = IO::Async::OS->socketpair( $family, "dgram" ) + or die "Could not socketpair - $!"; + + isa_ok( $S1, "IO::Socket", '$S1 isa IO::Socket' ); + isa_ok( $S2, "IO::Socket", '$S2 isa IO::Socket' ); + + is( $S1->socktype, SOCK_DGRAM, '$S1->socktype is SOCK_DGRAM' ); + is( $S2->socktype, SOCK_DGRAM, '$S2->socktype is SOCK_DGRAM' ); + + $S1->syswrite( "Hello" ); + is( do { my $b; $S2->sysread( $b, 8192 ); $b }, "Hello", '$S1 --writes-> $S2' ); + + $S2->syswrite( "Goodbye" ); + is( do { my $b; $S1->sysread( $b, 8192 ); $b }, "Goodbye", '$S2 --writes-> $S1' ); +} + +{ + my ( $Prd, $Pwr ) = IO::Async::OS->pipepair or die "Could not pipepair - $!"; + + $Pwr->syswrite( "Hello" ); + is( do { my $b; $Prd->sysread( $b, 8192 ); $b }, "Hello", '$Pwr --writes-> $Prd' ); + + # Writing to $Prd _may_ fail, but some systems might implement this as a + # socketpair instead. We won't test it just in case +} + +{ + my ( $rdA, $wrA, $rdB, $wrB ) = IO::Async::OS->pipequad or die "Could not pipequad - $!"; + + $wrA->syswrite( "Hello" ); + is( do { my $b; $rdA->sysread( $b, 8192 ); $b }, "Hello", '$wrA --writes-> $rdA' ); + + $wrB->syswrite( "Goodbye" ); + is( do { my $b; $rdB->sysread( $b, 8192 ); $b }, "Goodbye", '$wrB --writes-> $rdB' ); +} + +is( IO::Async::OS->signame2num( 'TERM' ), SIGTERM, 'signame2num' ); + +is( IO::Async::OS->getfamilybyname( "inet" ), AF_INET, 'getfamilybyname "inet"' ); +is( IO::Async::OS->getfamilybyname( AF_INET ), AF_INET, 'getfamilybyname AF_INET' ); + +is( IO::Async::OS->getsocktypebyname( "stream" ), SOCK_STREAM, 'getsocktypebyname "stream"' ); +is( IO::Async::OS->getsocktypebyname( SOCK_STREAM ), SOCK_STREAM, 'getsocktypebyname SOCK_STREAM' ); + +{ + my $sinaddr = pack_sockaddr_in( 56, inet_aton( "1.2.3.4" ) ); + + is_deeply( [ IO::Async::OS->extract_addrinfo( [ "inet", "stream", 0, $sinaddr ] ) ], + [ AF_INET, SOCK_STREAM, 0, $sinaddr ], + 'extract_addrinfo( ARRAY )' ); + + is_deeply( [ IO::Async::OS->extract_addrinfo( { + family => "inet", + socktype => "stream", + addr => $sinaddr + } ) ], + [ AF_INET, SOCK_STREAM, 0, $sinaddr ], + 'extract_addrinfo( HASH )' ); + + is_deeply( [ IO::Async::OS->extract_addrinfo( { + family => "inet", + socktype => "stream", + ip => "1.2.3.4", + port => "56", + } ) ], + [ AF_INET, SOCK_STREAM, 0, $sinaddr ], + 'extract_addrinfo( HASH ) with inet, ip+port' ); + + is_deeply( [ IO::Async::OS->extract_addrinfo( { + family => "inet", + socktype => "stream", + port => "56", + } ) ], + [ AF_INET, SOCK_STREAM, 0, pack_sockaddr_in( 56, INADDR_ANY ) ], + 'extract_addrinfo( HASH ) with inet, port' ); + + is_deeply( [ IO::Async::OS->extract_addrinfo( { + family => "inet", + socktype => "stream", + } ) ], + [ AF_INET, SOCK_STREAM, 0, pack_sockaddr_in( 0, INADDR_ANY ) ], + 'extract_addrinfo( HASH ) with inet only' ); + + ok( exception { IO::Async::OS->extract_addrinfo( { + family => "inet", + host => "foobar.com", + } ) }, 'extract_addrinfo for inet complains about unrecognised key' ); +} + +SKIP: { + my $sin6addr = eval { Socket::pack_sockaddr_in6( 1234, inet_pton( AF_INET6, "fe80::5678" ) ) }; + skip "No pack_sockaddr_in6", 1 unless defined $sin6addr; + + is_deeply( [ IO::Async::OS->extract_addrinfo( { + family => "inet6", + socktype => "stream", + ip => "fe80::5678", + port => "1234", + } ) ], + [ AF_INET6, SOCK_STREAM, 0, $sin6addr ], + 'extract_addrinfo( HASH ) with inet6, ip+port' ); +} + +SKIP: { + skip "No pack_sockaddr_un", 1 unless IO::Async::OS->HAVE_SOCKADDR_UN; + my $sunaddr = pack_sockaddr_un( "foo.sock" ); + + is_deeply( [ IO::Async::OS->extract_addrinfo( { + family => "unix", + socktype => "stream", + path => "foo.sock", + } ) ], + [ AF_UNIX, SOCK_STREAM, 0, $sunaddr ], + 'extract_addrinfo( HASH ) with unix, path' ); +} + +ok( exception { IO::Async::OS->extract_addrinfo( { family => "hohum" } ) }, + 'extract_addrinfo on unrecognised family complains' ); + +done_testing; diff --git a/t/03loop-magic.t b/t/03loop-magic.t new file mode 100644 index 0000000..ede4105 --- /dev/null +++ b/t/03loop-magic.t @@ -0,0 +1,51 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; + +use IO::Async::Loop; + +$IO::Async::Loop::LOOP_NO_OS = 1; +delete $ENV{IO_ASYNC_LOOP}; # Just in case it was already set + +my $loop; + +my $LOOPCLASS = "IO::Async::Loop::" . ( IO::Async::OS->LOOP_BUILTIN_CLASSES )[0]; + +$loop = IO::Async::Loop->new; + +isa_ok( $loop, $LOOPCLASS, 'Magic constructor in default mode' ) or + diag( 'ref($loop) is ' . ref $loop ); + +is( IO::Async::Loop->new, $loop, 'IO::Async::Loop->new again yields same loop' ); + +{ + local $ENV{IO_ASYNC_LOOP} = "t::StupidLoop"; + undef $IO::Async::Loop::ONE_TRUE_LOOP; + + $loop = IO::Async::Loop->new; + + isa_ok( $loop, "t::StupidLoop", 'Magic constructor obeys $ENV{IO_ASYNC_LOOP}' ); +} + +{ + local $IO::Async::Loop::LOOP = "t::StupidLoop"; + undef $IO::Async::Loop::ONE_TRUE_LOOP; + + $loop = IO::Async::Loop->new; + + isa_ok( $loop, "t::StupidLoop", 'Magic constructor obeys $IO::Async::Loop::LOOP' ); +} + +{ + local $IO::Async::Loop::LOOP = "Select"; + undef $IO::Async::Loop::ONE_TRUE_LOOP; + + $loop = IO::Async::Loop->new; + + isa_ok( $loop, "IO::Async::Loop::Select", 'Magic constructor expands unqualified package names' ); +} + +done_testing; diff --git a/t/04notifier.t b/t/04notifier.t new file mode 100644 index 0000000..8016583 --- /dev/null +++ b/t/04notifier.t @@ -0,0 +1,194 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use IO::Async::Notifier; + +{ + my $notifier = IO::Async::Notifier->new( + notifier_name => "test1", + ); + + ok( defined $notifier, '$notifier defined' ); + isa_ok( $notifier, "IO::Async::Notifier", '$notifier isa IO::Async::Notifier' ); + + is_oneref( $notifier, '$notifier has refcount 1 initially' ); + + is( $notifier->notifier_name, "test1", '$notifier->notifier_name' ); + + ok( !exception { $notifier->configure; }, + '$notifier->configure no params succeeds' ); + + ok( exception { $notifier->configure( oranges => 1 ) }, + '$notifier->configure an unknown parameter fails' ); + + my %other; + no warnings 'redefine'; + local *IO::Async::Notifier::configure_unknown = sub { + shift; + %other = @_; + }; + + ok( !exception { $notifier->configure( oranges => 3 ) }, + '$notifier->configure with configure_unknown succeeds' ); + + is_deeply( \%other, { oranges => 3 }, '%other after configure_unknown' ); +} + +# weaseling +{ + my $notifier = IO::Async::Notifier->new; + + my @args; + my $mref = $notifier->_capture_weakself( sub { @args = @_ } ); + + is_oneref( $notifier, '$notifier has refcount 1 after _capture_weakself' ); + + $mref->( 123 ); + is_deeply( \@args, [ $notifier, 123 ], '@args after invoking $mref' ); + + my @callstack; + $notifier->_capture_weakself( sub { + my $level = 0; + push @callstack, [ (caller $level++)[0,3] ] while defined caller $level; + } )->(); + + is_deeply( \@callstack, + [ [ "main", "main::__ANON__" ] ], + 'trampoline does not appear in _capture_weakself callstack' ); + + undef @args; + + $mref = $notifier->_replace_weakself( sub { @args = @_ } ); + + is_oneref( $notifier, '$notifier has refcount 1 after _replace_weakself' ); + + my $outerself = bless [], "OtherClass"; + $mref->( $outerself, 456 ); + is_deeply( \@args, [ $notifier, 456 ], '@args after invoking replacer $mref' ); + + isa_ok( $outerself, "OtherClass", '$outerself unchanged' ); + + ok( exception { $notifier->_capture_weakself( 'cannotdo' ) }, + '$notifier->_capture_weakself on unknown method name fails' ); +} + +# Subclass +{ + my @subargs; + { + package TestNotifier; + use base qw( IO::Async::Notifier ); + + sub frobnicate { @subargs = @_ } + } + + my $subn = TestNotifier->new; + + my $mref = $subn->_capture_weakself( 'frobnicate' ); + + is_oneref( $subn, '$subn has refcount 1 after _capture_weakself on named method' ); + + $mref->( 456 ); + is_deeply( \@subargs, [ $subn, 456 ], '@subargs after invoking $mref on named method' ); + + undef @subargs; + + # Method capture + { + my @newargs; + + no warnings 'redefine'; + local *TestNotifier::frobnicate = sub { @newargs = @_; }; + + $mref->( 321 ); + + is_deeply( \@subargs, [], '@subargs empty after TestNotifier::frobnicate replacement' ); + is_deeply( \@newargs, [ $subn, 321 ], '@newargs after TestNotifier::frobnicate replacement' ); + } + + undef @subargs; + + $subn->invoke_event( 'frobnicate', 78 ); + is_deeply( \@subargs, [ $subn, 78 ], '@subargs after ->invoke_event' ); + + undef @subargs; + + is_deeply( $subn->maybe_invoke_event( 'frobnicate', 'a'..'c' ), + [ $subn, 'a'..'c' ], + 'return value from ->maybe_invoke_event' ); + + is( $subn->maybe_invoke_event( 'mangle' ), undef, 'return value from ->maybe_invoke_event on missing event' ); + + undef @subargs; + + my $cb = $subn->make_event_cb( 'frobnicate' ); + + is( ref $cb, "CODE", '->make_event_cb returns a CODE reference' ); + is_oneref( $subn, '$subn has refcount 1 after ->make_event_cb' ); + + $cb->( 90 ); + is_deeply( \@subargs, [ $subn, 90 ], '@subargs after ->make_event_cb->()' ); + + isa_ok( $subn->maybe_make_event_cb( 'frobnicate' ), "CODE", '->maybe_make_event_cb yields CODE ref' ); + is( $subn->maybe_make_event_cb( 'mangle' ), undef, '->maybe_make_event_cb on missing event yields undef' ); + + undef @subargs; + + is_oneref( $subn, '$subn has refcount 1 finally' ); +} + +# parent/child +{ + my $parent = IO::Async::Notifier->new; + my $child = IO::Async::Notifier->new; + + is_oneref( $parent, '$parent has refcount 1 initially' ); + is_oneref( $child, '$child has refcount 1 initially' ); + + $parent->add_child( $child ); + + is( $child->parent, $parent, '$child->parent is $parent' ); + is_deeply( [ $parent->children ], [ $child ], '$parent->children' ); + + is_oneref( $parent, '$parent has refcount 1 after add_child' ); + is_refcount( $child, 2, '$child has refcount 2 after add_child' ); + + ok( exception { $parent->add_child( $child ) }, 'Adding child again fails' ); + + $parent->remove_child( $child ); + + is_oneref( $child, '$child has refcount 1 after remove_child' ); + is_deeply( [ $parent->children ], [], '$parent->children now empty' ); +} + +# invoke_error +{ + my $parent = IO::Async::Notifier->new; + my $child = IO::Async::Notifier->new; + + $parent->add_child( $child ); + + # invoke_error no handler + ok( exception { $parent->invoke_error( "It went wrong", wrong => ) }, + 'Exception thrown from ->invoke_error with no handler' ); + + # invoke_error handler + my $err; + $parent->configure( on_error => sub { $err = $_[1] } ); + + ok( !exception { $parent->invoke_error( "It's still wrong", wrong => ) }, + 'Exception not thrown from ->invoke_error with handler' ); + is( $err, "It's still wrong", '$message to on_error' ); + + ok( !exception { $child->invoke_error( "Wrong on child", wrong => ) }, + 'Exception not thrown from ->invoke_error on child' ); + is( $err, "Wrong on child", '$message to parent on_error' ); +} + +done_testing; diff --git a/t/05notifier-loop.t b/t/05notifier-loop.t new file mode 100644 index 0000000..ceaea6e --- /dev/null +++ b/t/05notifier-loop.t @@ -0,0 +1,129 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use IO::Async::Notifier; + +use IO::Async::Loop; + +my $loop = IO::Async::Loop->new; +is_refcount( $loop, 2, '$loop has refcount 2 initially' ); + +{ + package TestNotifier; + use base qw( IO::Async::Notifier ); + + sub new + { + my $self = shift->SUPER::new; + ( $self->{varref} ) = @_; + return $self; + } + + sub _add_to_loop + { + my $self = shift; + ${ $self->{varref} } = 1; + } + + sub _remove_from_loop + { + my $self = shift; + ${ $self->{varref} } = 0; + } +} + +# $loop->add +{ + my $notifier = TestNotifier->new( \my $in_loop ); + + is_deeply( [ $loop->notifiers ], + [], + '$loop->notifiers empty' ); + is( $notifier->loop, undef, 'loop undef' ); + + $loop->add( $notifier ); + + is_refcount( $loop, 2, '$loop has refcount 2 adding Notifier' ); + is_refcount( $notifier, 2, '$notifier has refcount 2 after adding to Loop' ); + + is( $notifier->loop, $loop, 'loop $loop' ); + + is_deeply( [ $loop->notifiers ], + [ $notifier ], + '$loop->notifiers contains new Notifier' ); + + ok( $in_loop, '_add_to_loop called' ); + + ok( exception { $loop->add( $notifier ) }, 'adding again produces error' ); + + $loop->remove( $notifier ); + + is( $notifier->loop, undef, '$notifier->loop is undef' ); + + is_deeply( [ $loop->notifiers ], + [], + '$loop->notifiers empty once more' ); + + ok( !$in_loop, '_remove_from_loop called' ); + + is_oneref( $notifier, '$notifier has refcount 1 finally' ); +} + +# parent/child in Loop +{ + my $parent = TestNotifier->new( \my $parent_in_loop ); + my $child = TestNotifier->new( \my $child_in_loop ); + + $loop->add( $parent ); + + $parent->add_child( $child ); + + is_refcount( $child, 3, '$child has refcount 3 after add_child within loop' ); + + is( $parent->loop, $loop, '$parent->loop is $loop' ); + is( $child->loop, $loop, '$child->loop is $loop' ); + + ok( $parent_in_loop, '$parent now in loop' ); + ok( $child_in_loop, '$child now in loop' ); + + ok( exception { $loop->remove( $child ) }, 'Directly removing a child from the loop fails' ); + + $loop->remove( $parent ); + + is_deeply( [ $parent->children ], [ $child ], '$parent->children after $loop->remove' ); + + is_oneref( $parent, '$parent has refcount 1 after removal from loop' ); + is_refcount( $child, 2, '$child has refcount 2 after removal of parent from loop' ); + + is( $parent->loop, undef, '$parent->loop is undef' ); + is( $child->loop, undef, '$child->loop is undef' ); + + ok( !$parent_in_loop, '$parent no longer in loop' ); + ok( !$child_in_loop, '$child no longer in loop' ); + + ok( exception { $loop->add( $child ) }, 'Directly adding a child to the loop fails' ); + + $loop->add( $parent ); + + is( $child->loop, $loop, '$child->loop is $loop after remove/add parent' ); + + ok( $parent_in_loop, '$parent now in loop' ); + ok( $child_in_loop, '$child now in loop' ); + + $loop->remove( $parent ); + + $parent->remove_child( $child ); + + is_oneref( $parent, '$parent has refcount 1 finally' ); + is_oneref( $child, '$child has refcount 1 finally' ); +} + +is_refcount( $loop, 2, '$loop has refcount 2 finally' ); + +done_testing; diff --git a/t/06notifier-mixin.t b/t/06notifier-mixin.t new file mode 100644 index 0000000..7554fc4 --- /dev/null +++ b/t/06notifier-mixin.t @@ -0,0 +1,50 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Refcount; + +use IO::Async::Loop; + +my $loop = IO::Async::Loop->new; + +is_refcount( $loop, 2, '$loop has refcount 2 initially' ); + +my $notifier = SomeEventSource::Async->new; +my $in_loop; + +isa_ok( $notifier, "SomeEventSource", '$notifier isa SomeEventSource' ); +isa_ok( $notifier, "IO::Async::Notifier", '$notifier isa IO::Async::Notifier' ); + +$loop->add( $notifier ); + +is_refcount( $loop, 2, '$loop has refcount 2 adding Notifier' ); +is_refcount( $notifier, 2, '$notifier has refcount 2 after adding to Loop' ); + +is( $notifier->loop, $loop, 'loop $loop' ); + +ok( $in_loop, 'SomeEventSource::Async added to Loop' ); + +$loop->remove( $notifier ); + +is( $notifier->loop, undef, '$notifier->loop is undef' ); + +ok( !$in_loop, 'SomeEventSource::Async removed from Loop' ); + +done_testing; + +package SomeEventSource; + +sub new +{ + my $class = shift; + return bless {}, $class; +} + +package SomeEventSource::Async; +use base qw( SomeEventSource IO::Async::Notifier ); + +sub _add_to_loop { $in_loop = 1 } +sub _remove_from_loop { $in_loop = 0 } diff --git a/t/07notifier-future.t b/t/07notifier-future.t new file mode 100644 index 0000000..066fbb9 --- /dev/null +++ b/t/07notifier-future.t @@ -0,0 +1,58 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use IO::Async::Notifier; +use Future; + +my ( $err, $name, @detail ); +my $notifier = IO::Async::Notifier->new( + on_error => sub { + ( undef, $err, $name, @detail ) = @_; + }, +); + +# done +{ + my $f = Future->new; + + $notifier->adopt_future( $f ); + + is_refcount( $f, 2, '$f has refcount 2 after ->adopt_future' ); + is_oneref( $notifier, '$notifier still has refcount 1 after ->adopt_future' ); + + $f->done( "result" ); + + is_refcount( $f, 1, '$f has refcount 1 after $f->done' ); +} + +# fail +{ + my $f = Future->new; + + $notifier->adopt_future( $f ); + + $f->fail( "It failed", name => 1, 2, 3 ); + + is( $err, "It failed", '$err after $f->fail' ); + is( $name, "name", '$name after $f->fail' ); + is_deeply( \@detail, [ 1, 2, 3 ], '@detail after $f->fail' ); + + is_refcount( $f, 1, '$f has refcount 1 after $f->fail' ); + + undef $err; + + $f = Future->new; + $notifier->adopt_future( $f->else_done() ); + + $f->fail( "Not captured" ); + + ok( !defined $err, '$err not defined after ->else_done suppressed failure' ); +} + +done_testing; diff --git a/t/10loop-poll-io.t b/t/10loop-poll-io.t new file mode 100644 index 0000000..8c34ffa --- /dev/null +++ b/t/10loop-poll-io.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Poll', 'io' ); diff --git a/t/10loop-select-io.t b/t/10loop-select-io.t new file mode 100644 index 0000000..b3e3916 --- /dev/null +++ b/t/10loop-select-io.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Select', 'io' ); diff --git a/t/11loop-poll-timer.t b/t/11loop-poll-timer.t new file mode 100644 index 0000000..748ba2a --- /dev/null +++ b/t/11loop-poll-timer.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Poll', 'timer' ); diff --git a/t/11loop-select-timer.t b/t/11loop-select-timer.t new file mode 100644 index 0000000..314bd38 --- /dev/null +++ b/t/11loop-select-timer.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Select', 'timer' ); diff --git a/t/12loop-poll-signal.t b/t/12loop-poll-signal.t new file mode 100644 index 0000000..4c8ef24 --- /dev/null +++ b/t/12loop-poll-signal.t @@ -0,0 +1,9 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +Test::More::plan skip_all => "This OS does not have signals" unless IO::Async::OS->HAVE_SIGNALS; + +run_tests( 'IO::Async::Loop::Poll', 'signal' ); diff --git a/t/12loop-select-signal.t b/t/12loop-select-signal.t new file mode 100644 index 0000000..66a6c91 --- /dev/null +++ b/t/12loop-select-signal.t @@ -0,0 +1,9 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +Test::More::plan skip_all => "This OS does not have signals" unless IO::Async::OS->HAVE_SIGNALS; + +run_tests( 'IO::Async::Loop::Select', 'signal' ); diff --git a/t/13loop-poll-idle.t b/t/13loop-poll-idle.t new file mode 100644 index 0000000..3ce457b --- /dev/null +++ b/t/13loop-poll-idle.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Poll', 'idle' ); diff --git a/t/13loop-select-idle.t b/t/13loop-select-idle.t new file mode 100644 index 0000000..8cea7a7 --- /dev/null +++ b/t/13loop-select-idle.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Select', 'idle' ); diff --git a/t/14loop-poll-child.t b/t/14loop-poll-child.t new file mode 100644 index 0000000..5337166 --- /dev/null +++ b/t/14loop-poll-child.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Poll', 'child' ); diff --git a/t/14loop-select-child.t b/t/14loop-select-child.t new file mode 100644 index 0000000..8e31e23 --- /dev/null +++ b/t/14loop-select-child.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Select', 'child' ); diff --git a/t/15loop-poll-control.t b/t/15loop-poll-control.t new file mode 100644 index 0000000..74839dd --- /dev/null +++ b/t/15loop-poll-control.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Poll', 'control' ); diff --git a/t/15loop-select-control.t b/t/15loop-select-control.t new file mode 100644 index 0000000..ff62634 --- /dev/null +++ b/t/15loop-select-control.t @@ -0,0 +1,7 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::LoopTests; +run_tests( 'IO::Async::Loop::Select', 'control' ); diff --git a/t/18loop-poll-legacy.t b/t/18loop-poll-legacy.t new file mode 100644 index 0000000..3c20dfb --- /dev/null +++ b/t/18loop-poll-legacy.t @@ -0,0 +1,95 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; + +use IO::Poll; + +use IO::Async::OS; + +use IO::Async::Loop::Poll; + +my $poll = IO::Poll->new; +my $loop = IO::Async::Loop::Poll->new( poll => $poll ); + +my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + +# Need sockets in nonblocking mode +$S1->blocking( 0 ); +$S2->blocking( 0 ); + +# Empty + +is_deeply( [ $poll->handles ], [], '$poll->handles empty initially' ); + +# watch_io + +my $readready = 0; +$loop->watch_io( + handle => $S1, + on_read_ready => sub { $readready = 1 }, +); + +is_deeply( [ $poll->handles ], [ $S1 ], '$poll->handles after watch_io read_ready' ); + +$S2->syswrite( "data\n" ); + +# We should still wait a little while even thought we expect to be ready +# immediately, because talking to ourself with 0 poll timeout is a race +# condition - we can still race with the kernel. + +$poll->poll( 0.1 ); + +is( $readready, 0, '$readready before post_poll' ); +$loop->post_poll; +is( $readready, 1, '$readready after post_poll' ); + +# Ready $S1 to clear the data +$S1->getline; # ignore return + +$loop->unwatch_io( + handle => $S1, + on_read_ready => 1, +); + +is_deeply( [ $poll->handles ], [], '$poll->handles empty after unwatch_io read_ready' ); + +my $writeready = 0; +$loop->watch_io( + handle => $S1, + on_write_ready => sub { $writeready = 1 }, +); + +is_deeply( [ $poll->handles ], [ $S1 ], '$poll->handles after watch_io write_ready' ); + +$poll->poll( 0.1 ); + +is( $writeready, 0, '$writeready before post_poll' ); +$loop->post_poll; +is( $writeready, 1, '$writeready after post_poll' ); + +$loop->unwatch_io( + handle => $S1, + on_write_ready => 1, +); + +is_deeply( [ $poll->handles ], [], '$poll->handles empty after unwatch_io write_ready' ); + +# Removal is clean (tests for workaround to bug in IO::Poll version 0.05) + +my ( $P1, $P2 ) = IO::Async::OS->pipepair or die "Cannot pipepair - $!"; + +# Just to make the loop non-empty +$loop->watch_io( handle => $P2, on_read_ready => sub {} ); + +$loop->watch_io( handle => \*STDOUT, on_write_ready => sub {} ); + +is( scalar $poll->handles, 2, '$poll->handles before removal in clean removal test' ); + +$loop->unwatch_io( handle => \*STDOUT, on_write_ready => 1 ); + +is( scalar $poll->handles, 1, '$poll->handles after removal in clean removal test' ); + +done_testing; diff --git a/t/18loop-select-legacy.t b/t/18loop-select-legacy.t new file mode 100644 index 0000000..d5796b0 --- /dev/null +++ b/t/18loop-select-legacy.t @@ -0,0 +1,152 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; + +use Time::HiRes qw( time ); + +use IO::Async::Loop::Select; + +use IO::Async::OS; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop::Select->new; + +my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + +# Need sockets in nonblocking mode +$S1->blocking( 0 ); +$S2->blocking( 0 ); + +my $testvec = ''; +vec( $testvec, $S1->fileno, 1 ) = 1; + +my ( $rvec, $wvec, $evec ) = ('') x 3; +my $timeout; + +# Empty + +$loop->pre_select( \$rvec, \$wvec, \$evec, \$timeout ); +is( $rvec, '', '$rvec idling pre_select' ); +is( $wvec, '', '$wvec idling pre_select' ); +is( $evec, '', '$evec idling pre_select' ); +is( $timeout, undef, '$timeout idling pre_select' ); + +# watch_io + +my $readready = 0; +$loop->watch_io( + handle => $S1, + on_read_ready => sub { $readready = 1 }, +); + +$loop->pre_select( \$rvec, \$wvec, \$evec, \$timeout ); + +is( $rvec, $testvec, '$rvec readready pre_select' ); +is( $wvec, '', '$wvec readready pre_select' ); +is( $evec, '', '$evec readready pre_select' ); +is( $timeout, undef, '$timeout readready pre_select' ); + +is( $readready, 0, '$readready readready pre_select' ); + +$rvec = $testvec; +$wvec = ''; +$evec = ''; + +$loop->post_select( $rvec, $wvec, $evec ); + +is( $readready, 1, '$readready readready post_select' ); + +$loop->unwatch_io( + handle => $S1, + on_read_ready => 1, +); + +my $writeready = 0; +$loop->watch_io( + handle => $S1, + on_write_ready => sub { $writeready = 1 }, +); + +$loop->pre_select( \$rvec, \$wvec, \$evec, \$timeout ); + +is( $rvec, $testvec, '$rvec writeready pre_select' ); +is( $wvec, $testvec, '$wvec writeready pre_select' ); +is( $evec, IO::Async::OS->HAVE_SELECT_CONNECT_EVEC ? $testvec : '', '$evec writeready pre_select' ); +is( $timeout, undef, '$timeout writeready pre_select' ); + +is( $writeready, 0, '$writeready writeready pre_select' ); + +$rvec = ''; +$wvec = $testvec; +$evec = ''; + +$loop->post_select( $rvec, $wvec, $evec ); + +is( $writeready, 1, '$writeready writeready post_select' ); + +$loop->unwatch_io( + handle => $S1, + on_write_ready => 1, +); + +# watch_time + +$rvec = $wvec = $evec = ''; +$timeout = 5 * AUT; + +$loop->pre_select( \$rvec, \$wvec, \$evec, \$timeout ); +is( $timeout, 5 * AUT, '$timeout idling pre_select with timeout' ); + +my $done = 0; +$loop->watch_time( after => 2 * AUT, code => sub { $done = 1; } ); + +$loop->pre_select( \$rvec, \$wvec, \$evec, \$timeout ); +cmp_ok( $timeout/AUT, '>', 1.7, '$timeout while timer waiting pre_select at least 1.7' ); +cmp_ok( $timeout/AUT, '<', 2.5, '$timeout while timer waiting pre_select at least 2.5' ); + +my ( $now, $took ); + +$now = time; +select( $rvec, $wvec, $evec, $timeout ); +$took = (time - $now) / AUT; + +cmp_ok( $took, '>', 1.7, 'loop_once(5) while waiting for timer takes at least 1.7 seconds' ); +cmp_ok( $took, '<', 10, 'loop_once(5) while waiting for timer no more than 10 seconds' ); +if( $took > 2.5 ) { + diag( "took more than 2.5 seconds to select(2).\n" . + "This is not itself a bug, and may just be an indication of a busy testing machine" ); +} + +$loop->post_select( $rvec, $evec, $wvec ); + +# select might have returned just a little early, such that the TimerQueue +# doesn't think anything is ready yet. We need to handle that case. +while( !$done ) { + die "It should have been ready by now" if( time - $now > 5 * AUT ); + + $timeout = 0.1 * AUT; + + $loop->pre_select( \$rvec, \$wvec, \$evec, \$timeout ); + select( $rvec, $wvec, $evec, $timeout ); + $loop->post_select( $rvec, $evec, $wvec ); +} + +is( $done, 1, '$done after post_select while waiting for timer' ); + +my $id = $loop->watch_time( after => 1 * AUT, code => sub { $done = 2; } ); +$loop->unwatch_time( $id ); + +$done = 0; +$now = time; + +$loop->pre_select( \$rvec, \$wvec, \$evec, \$timeout ); +select( $rvec, $wvec, $evec, 1.5 * AUT ); +$loop->post_select( $rvec, $evec, $wvec ); + +is( $done, 0, '$done still 0 before cancelled timeout' ); + +done_testing; diff --git a/t/19loop-future.t b/t/19loop-future.t new file mode 100644 index 0000000..45ab441 --- /dev/null +++ b/t/19loop-future.t @@ -0,0 +1,107 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Identity; +use t::TimeAbout; + +use IO::Async::Loop; + +use Future; +use IO::Async::Future; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop->new_builtin; + +{ + my $future = Future->new; + + $loop->later( sub { $future->done( "result" ) } ); + + my $ret = $loop->await( $future ); + identical( $ret, $future, '$loop->await( $future ) returns $future' ); + + is_deeply( [ $future->get ], [ "result" ], '$future->get' ); +} + +{ + my @futures = map { Future->new } 0 .. 2; + + do { my $id = $_; $loop->later( sub { $futures[$id]->done } ) } for 0 .. 2; + + $loop->await_all( @futures ); + + ok( 1, '$loop->await_all' ); + ok( $futures[$_]->is_ready, "future $_ ready" ) for 0 .. 2; +} + +{ + my $future = IO::Async::Future->new( $loop ); + + identical( $future->loop, $loop, '$future->loop yields $loop' ); + + $loop->later( sub { $future->done( "result" ) } ); + + is_deeply( [ $future->get ], [ "result" ], '$future->get on IO::Async::Future' ); +} + +{ + my $future = $loop->new_future; + + $loop->later( sub { $future->done( "result" ) } ); + + is_deeply( [ $future->get ], [ "result" ], '$future->get on IO::Async::Future from $loop->new_future' ); +} + +# done_later +{ + my $future = $loop->new_future; + + identical( $future->done_later( "deferred result" ), $future, '->done_later returns $future' ); + ok( !$future->is_ready, '$future not yet ready after ->done_later' ); + + is_deeply( [ $future->get ], [ "deferred result" ], '$future now ready after ->get' ); +} + +# fail_later +{ + my $future = $loop->new_future; + + identical( $future->fail_later( "deferred exception\n" ), $future, '->fail_later returns $future' ); + ok( !$future->is_ready, '$future not yet ready after ->fail_later' ); + + $loop->await( $future ); + + is_deeply( [ $future->failure ], [ "deferred exception\n" ], '$future now ready after $loop->await' ); +} + +# delay_future +{ + my $future = $loop->delay_future( after => 1 * AUT ); + + time_about( sub { $loop->await( $future ) }, 1, '->delay_future is ready' ); + + ok( $future->is_ready, '$future is ready from delay_future' ); + is_deeply( [ $future->get ], [], '$future->get returns empty list on delay_future' ); + + # Check that ->cancel does not crash + $loop->delay_future( after => 1 * AUT )->cancel; +} + +# timeout_future +{ + my $future = $loop->timeout_future( after => 1 * AUT ); + + time_about( sub { $loop->await( $future ) }, 1, '->timeout_future is ready' ); + + ok( $future->is_ready, '$future is ready from timeout_future' ); + is( scalar $future->failure, "Timeout", '$future failed with "Timeout" for timeout_future' ); + + # Check that ->cancel does not crash + $loop->timeout_future( after => 1 * AUT )->cancel; +} + +done_testing; diff --git a/t/19test.t b/t/19test.t new file mode 100644 index 0000000..cc24d45 --- /dev/null +++ b/t/19test.t @@ -0,0 +1,69 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; +use Test::Refcount; +use IO::Async::Test; + +use IO::Async::OS; + +use IO::Async::Loop; + +my $loop = IO::Async::Loop->new_builtin; + +is_refcount( $loop, 2, '$loop has refcount 2 initially' ); + +testing_loop( $loop ); + +is_refcount( $loop, 3, '$loop has refcount 3 after adding to IO::Async::Test' ); + +my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + +my $readbuffer = ""; + +$loop->watch_io( + handle => $S1, + on_read_ready => sub { + $S1->sysread( $readbuffer, 8192, length $readbuffer ) or die "Test failed early"; + }, +); + +# This is just a token "does it run once?" test. A test of a test script. +# Mmmmmm. Meta-testing. +# Coming up with a proper test that would guarantee multiple loop_once +# cycles, etc.. is difficult. TODO for later I feel. +# In any case, the wait_for function is effectively tested to death in later +# test scripts which use it. If it fails to work, they'd notice it. + +$S2->syswrite( "A line\n" ); + +wait_for { $readbuffer =~ m/\n/ }; + +is( $readbuffer, "A line\n", 'Single-wait' ); + +$loop->unwatch_io( + handle => $S1, + on_read_ready => 1, +); + +# Now the automatic version + +$readbuffer = ""; + +$S2->syswrite( "Another line\n" ); + +wait_for_stream { $readbuffer =~ m/\n/ } $S1 => $readbuffer; + +is( $readbuffer, "Another line\n", 'Automatic stream read wait' ); + +$readbuffer = ""; + +$S2->syswrite( "Some dynamic data\n" ); + +wait_for_stream { $readbuffer =~ m/\n/ } $S1 => sub { $readbuffer .= shift }; + +is( $readbuffer, "Some dynamic data\n" ); + +done_testing; diff --git a/t/20handle.t b/t/20handle.t new file mode 100644 index 0000000..194a621 --- /dev/null +++ b/t/20handle.t @@ -0,0 +1,422 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Identity; +use Test::Refcount; + +use IO::Async::Loop; + +use IO::Async::Handle; + +use IO::Async::OS; + +use Socket qw( AF_INET SOCK_STREAM SOCK_DGRAM SO_TYPE unpack_sockaddr_in ); + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +sub mkhandles +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + + # Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + return ( $S1, $S2 ); +} + +ok( exception { IO::Async::Handle->new( handle => "Hello" ) }, 'Not a filehandle' ); + +# Read readiness +{ + my ( $S1, $S2 ) = mkhandles; + my $fd1 = $S1->fileno; + + my $readready = 0; + my @rrargs; + + my $handle = IO::Async::Handle->new( + read_handle => $S1, + on_read_ready => sub { @rrargs = @_; $readready = 1 }, + ); + + ok( defined $handle, '$handle defined' ); + isa_ok( $handle, "IO::Async::Handle", '$handle isa IO::Async::Handle' ); + + is( $handle->notifier_name, "r=$fd1", '$handle->notifier_name for read_handle' ); + + is_oneref( $handle, '$handle has refcount 1 initially' ); + + is( $handle->read_handle, $S1, '->read_handle returns S1' ); + is( $handle->read_fileno, $S1->fileno, '->read_fileno returns fileno(S1)' ); + + is( $handle->write_handle, undef, '->write_handle returns undef' ); + + ok( $handle->want_readready, 'want_readready true' ); + + $loop->add( $handle ); + + is_refcount( $handle, 2, '$handle has refcount 2 after adding to Loop' ); + + $loop->loop_once( 0.1 ); # nothing happens + + is( $readready, 0, '$readready while idle' ); + + $S2->syswrite( "data\n" ); + + wait_for { $readready }; + + is( $readready, 1, '$readready while readable' ); + is_deeply( \@rrargs, [ $handle ], 'on_read_ready args while readable' ); + + $S1->getline; # ignore return + + $readready = 0; + my $new_readready = 0; + + $handle->configure( on_read_ready => sub { $new_readready = 1 } ); + + $loop->loop_once( 0.1 ); # nothing happens + + is( $readready, 0, '$readready while idle after on_read_ready replace' ); + is( $new_readready, 0, '$new_readready while idle after on_read_ready replace' ); + + $S2->syswrite( "data\n" ); + + wait_for { $new_readready }; + + is( $readready, 0, '$readready while readable after on_read_ready replace' ); + is( $new_readready, 1, '$new_readready while readable after on_read_ready replace' ); + + $S1->getline; # ignore return + + ok( exception { $handle->want_writeready( 1 ); }, + 'setting want_writeready with write_handle == undef dies' ); + ok( !$handle->want_writeready, 'wantwriteready write_handle == undef false' ); + + undef @rrargs; + + is_refcount( $handle, 2, '$handle has refcount 2 before removing from Loop' ); + + $loop->remove( $handle ); + + is_oneref( $handle, '$handle has refcount 1 finally' ); +} + +# Write readiness +{ + my ( $S1, $S2 ) = mkhandles; + my $fd1 = $S1->fileno; + + my $writeready = 0; + my @wrargs; + + my $handle = IO::Async::Handle->new( + write_handle => $S1, + on_write_ready => sub { @wrargs = @_; $writeready = 1 }, + ); + + ok( defined $handle, '$handle defined' ); + isa_ok( $handle, "IO::Async::Handle", '$handle isa IO::Async::Handle' ); + + is( $handle->notifier_name, "w=$fd1", '$handle->notifier_name for write_handle' ); + + is_oneref( $handle, '$handle has refcount 1 initially' ); + + is( $handle->write_handle, $S1, '->write_handle returns S1' ); + is( $handle->write_fileno, $S1->fileno, '->write_fileno returns fileno(S1)' ); + + is( $handle->read_handle, undef, '->read_handle returns undef' ); + + ok( !$handle->want_writeready, 'want_writeready false' ); + + $loop->add( $handle ); + + is_refcount( $handle, 2, '$handle has refcount 2 after adding to Loop' ); + + $loop->loop_once( 0.1 ); # nothing happens + + is( $writeready, 0, '$writeready while idle' ); + + $handle->want_writeready( 1 ); + + wait_for { $writeready }; + + is( $writeready, 1, '$writeready while writeable' ); + is_deeply( \@wrargs, [ $handle ], 'on_write_ready args while writeable' ); + + $writeready = 0; + my $new_writeready = 0; + + $handle->configure( on_write_ready => sub { $new_writeready = 1 } ); + + wait_for { $new_writeready }; + + is( $writeready, 0, '$writeready while writeable after on_write_ready replace' ); + is( $new_writeready, 1, '$new_writeready while writeable after on_write_ready replace' ); + + undef @wrargs; + + is_refcount( $handle, 2, '$handle has refcount 2 before removing from Loop' ); + + $loop->remove( $handle ); + + is_oneref( $handle, '$handle has refcount 1 finally' ); +} + +# Combined handle +{ + my ( $S1, $S2 ) = mkhandles; + my $fd1 = $S1->fileno; + + my $handle = IO::Async::Handle->new( + handle => $S1, + on_read_ready => sub {}, + on_write_ready => sub {}, + ); + + is( $handle->read_handle, $S1, '->read_handle returns S1' ); + is( $handle->write_handle, $S1, '->write_handle returns S1' ); + + is( $handle->notifier_name, "rw=$fd1", '$handle->notifier_name for handle' ); +} + +# Subclass +my $sub_readready = 0; +my $sub_writeready = 0; + +{ + my ( $S1, $S2 ) = mkhandles; + + my $handle = TestHandle->new( + handle => $S1, + ); + + ok( defined $handle, 'subclass $handle defined' ); + isa_ok( $handle, "IO::Async::Handle", 'subclass $handle isa IO::Async::Handle' ); + + is_oneref( $handle, 'subclass $handle has refcount 1 initially' ); + + is( $handle->read_handle, $S1, 'subclass ->read_handle returns S1' ); + is( $handle->write_handle, $S1, 'subclass ->write_handle returns S1' ); + + $loop->add( $handle ); + + is_refcount( $handle, 2, 'subclass $handle has refcount 2 after adding to Loop' ); + + $S2->syswrite( "data\n" ); + + wait_for { $sub_readready }; + + is( $sub_readready, 1, '$sub_readready while readable' ); + is( $sub_writeready, 0, '$sub_writeready while readable' ); + + $S1->getline; # ignore return + $sub_readready = 0; + + $handle->want_writeready( 1 ); + + wait_for { $sub_writeready }; + + is( $sub_readready, 0, '$sub_readready while writeable' ); + is( $sub_writeready, 1, '$sub_writeready while writeable' ); + + $loop->remove( $handle ); +} + +# Close +{ + my ( $S1, $S2 ) = mkhandles; + + my $closed = 0; + + my $handle = IO::Async::Handle->new( + read_handle => $S1, + want_writeready => 0, + on_read_ready => sub {}, + on_closed => sub { $closed = 1 }, + ); + + $loop->add( $handle ); + + my $close_future = $handle->new_close_future; + + my $closed_by_future; + $close_future->on_done( sub { $closed_by_future++ } ); + + $handle->close; + + is( $closed, 1, '$closed after ->close' ); + + ok( $close_future->is_ready, '$close_future is now ready' ); + is( $closed_by_future, 1, '$closed_by_future after ->close' ); + + # removed itself +} + +# Close read/write +{ + my ( $Srd1, $Srd2 ) = mkhandles; + my ( $Swr1, $Swr2 ) = mkhandles; + + local $SIG{PIPE} = "IGNORE"; + + my $readready = 0; + my $writeready = 0; + + my $closed = 0; + + my $handle = IO::Async::Handle->new( + read_handle => $Srd1, + write_handle => $Swr1, + on_read_ready => sub { $readready++ }, + on_write_ready => sub { $writeready++ }, + on_closed => sub { $closed++ }, + want_writeready => 1, + ); + + $loop->add( $handle ); + + $handle->close_read; + + wait_for { $writeready }; + is( $writeready, 1, '$writeready after ->close_read' ); + + $handle->write_handle->syswrite( "Still works\n" ); + is( $Swr2->getline, "Still works\n", 'write handle still works' ); + + is( $closed, 0, 'not $closed after ->close_read' ); + + is( $handle->loop, $loop, 'Handle still member of Loop after ->close_read' ); + + ( $Srd1, $Srd2 ) = mkhandles; + + $handle->configure( read_handle => $Srd1 ); + + $handle->close_write; + + $Srd2->syswrite( "Also works\n" ); + + wait_for { $readready }; + is( $readready, 1, '$readready after ->close_write' ); + + is( $handle->read_handle->getline, "Also works\n", 'read handle still works' ); + is( $Swr2->getline, undef, 'sysread from EOF write handle' ); + + is( $handle->loop, $loop, 'Handle still member of Loop after ->close_write' ); + + is( $closed, 0, 'not $closed after ->close_read' ); + + $handle->close_read; + + is( $closed, 1, '$closed after ->close_read + ->close_write' ); + + is( $handle->loop, undef, '$handle no longer member of Loop' ); +} + +# Late-binding of handle +{ + my $readready; + my $writeready; + + my $handle = IO::Async::Handle->new( + want_writeready => 0, + on_read_ready => sub { $readready = 1 }, + on_write_ready => sub { $writeready = 1 }, + ); + + ok( defined $handle, '$handle defined' ); + + ok( !defined $handle->read_handle, '->read_handle not defined' ); + ok( !defined $handle->write_handle, '->write_handle not defined' ); + + is_oneref( $handle, '$handle latebound has refcount 1 initially' ); + + is( $handle->notifier_name, "no", '$handle->notifier_name for late bind before handles' ); + + $loop->add( $handle ); + + is_refcount( $handle, 2, '$handle latebound has refcount 2 after $loop->add' ); + + my ( $S1, $S2 ) = mkhandles; + my $fd1 = $S1->fileno; + + $handle->set_handle( $S1 ); + + is( $handle->read_handle, $S1, '->read_handle now S1' ); + is( $handle->write_handle, $S1, '->write_handle now S1' ); + + is_refcount( $handle, 2, '$handle latebound still has refcount 2 after set_handle' ); + + is( $handle->notifier_name, "rw=$fd1", '$handle->notifier_name for late bind after handles' ); + + $S2->syswrite( "readable" ); + + wait_for { $readready }; + pass( '$handle latebound still invokes on_read_ready' ); + + $loop->remove( $handle ); +} + +# ->socket and ->bind +{ + my $handle = IO::Async::Handle->new( on_read_ready => sub {}, on_write_ready => sub {} ); + + $handle->socket( [ 'inet', 'stream', 0 ] ); + + ok( defined $handle->read_handle, '->socket sets handle' ); + + is( $handle->read_handle->sockdomain, AF_INET, 'handle->sockdomain is AF_INET' ); + is( $handle->read_handle->sockopt(SO_TYPE), SOCK_STREAM, 'handle->socktype is SOCK_STREAM' ); + + $handle->bind( { family => "inet", socktype => "dgram" } )->get; + + is( $handle->read_handle->sockopt(SO_TYPE), SOCK_DGRAM, 'handle->socktype is SOCK_DGRAM' ); + # Not sure what port number but it should be nonzero + ok( ( unpack_sockaddr_in( $handle->read_handle->sockname ) )[0], 'handle->sockname has nonzero port' ); +} + +# Construction of IO::Handle from fileno +{ + my $handle = IO::Async::Handle->new( + read_fileno => 0, + on_read_ready => sub { }, + ); + + ok( defined $handle->read_handle, '->new with read_fileno creates read_handle' ); + is( $handle->read_handle->fileno, 0, '->fileno of read_handle' ); + + $handle = IO::Async::Handle->new( + write_fileno => 1, + on_write_ready => sub { }, + ); + + ok( defined $handle->write_handle, '->new with write_fileno creates write_handle' ); + is( $handle->write_handle->fileno, 1, '->fileno of write_handle' ); + + $handle = IO::Async::Handle->new( + read_fileno => 2, + write_fileno => 2, + on_read_ready => sub { }, + on_write_ready => sub { }, + ); + + identical( $handle->read_handle, $handle->write_handle, + '->new with equal read and write fileno only creates one handle' ); +} + +done_testing; + +package TestHandle; +use base qw( IO::Async::Handle ); + +sub on_read_ready { $sub_readready = 1 } +sub on_write_ready { $sub_writeready = 1 } diff --git a/t/21stream-1read.t b/t/21stream-1read.t new file mode 100644 index 0000000..aa49453 --- /dev/null +++ b/t/21stream-1read.t @@ -0,0 +1,637 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use IO::File; +use POSIX qw( ECONNRESET ); + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Stream; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +sub mkhandles +{ + my ( $rd, $wr ) = IO::Async::OS->pipepair or die "Cannot pipe() - $!"; + # Need handles in nonblocking mode + $rd->blocking( 0 ); + $wr->blocking( 0 ); + + return ( $rd, $wr ); +} + +{ + my ( $rd, $wr ) = mkhandles; + + my @lines; + + my $stream = IO::Async::Stream->new( + read_handle => $rd, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + ok( defined $stream, 'reading $stream defined' ); + isa_ok( $stream, "IO::Async::Stream", 'reading $stream isa IO::Async::Stream' ); + + is_oneref( $stream, 'reading $stream has refcount 1 initially' ); + + $loop->add( $stream ); + + is_refcount( $stream, 2, 'reading $stream has refcount 2 after adding to Loop' ); + + $wr->syswrite( "message\n" ); + + is_deeply( \@lines, [], '@lines before wait' ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "message\n" ], '@lines after wait' ); + + undef @lines; + + $wr->syswrite( "return" ); + + $loop->loop_once( 0.1 ); # nothing happens + + is_deeply( \@lines, [], '@lines partial still empty' ); + + $wr->syswrite( "\n" ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "return\n" ], '@lines partial completed now received' ); + + undef @lines; + + $wr->syswrite( "hello\nworld\n" ); + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "hello\n", "world\n" ], '@lines two at once' ); + + undef @lines; + my @new_lines; + $stream->configure( + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @new_lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + $wr->syswrite( "new\nlines\n" ); + + wait_for { scalar @new_lines }; + + is( scalar @lines, 0, '@lines still empty after on_read replace' ); + is_deeply( \@new_lines, [ "new\n", "lines\n" ], '@new_lines after on_read replace' ); + + is_refcount( $stream, 2, 'reading $stream has refcount 2 before removing from Loop' ); + + $loop->remove( $stream ); + + is_oneref( $stream, 'reading $stream refcount 1 finally' ); +} + +# Abstract reading with reader function +{ + my ( $rd, $wr ) = mkhandles; + my $buffer = "Here is the contents\n"; + + my @lines; + my $stream = IO::Async::Stream->new( + read_handle => $rd, + reader => sub { + my $self = shift; + my $more = substr( $buffer, 0, $_[2], "" ); + $_[1] .= $more; + return length $more; + }, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + $loop->add( $stream ); + + # make it readready + $wr->syswrite( "1" ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "Here is the contents\n" ], '@lines from stream with abstract reader' ); + + $loop->remove( $stream ); +} + +# ->want_readready_for_write +{ + my ( $rd, $wr ) = mkhandles; + + my $reader_called; + my $writer_called; + my $stream = IO::Async::Stream->new( + handle => $rd, + on_read => sub { return 0; }, # ignore reading + reader => sub { $reader_called++; sysread( $rd, $_[2], $_[3] ) }, + writer => sub { $writer_called++; return 1 }, + ); + + $loop->add( $stream ); + + # Hacky hack - make the stream want to write, but don't mark the stream write-ready + $stream->write( "A" ); + $stream->want_writeready_for_write( 0 ); + # End hack + + # make it readready + $wr->syswrite( "1" ); + + wait_for { $reader_called }; + + ok( !$writer_called, 'writer not yet called before ->want_readready_for_write' ); + + $stream->want_readready_for_write( 1 ); + + undef $reader_called; + $wr->syswrite( "2" ); + wait_for { $reader_called && $writer_called }; + + ok( $writer_called, 'writer now invoked with ->want_readready_for_write' ); + + $loop->remove( $stream ); +} + +{ + my ( $rd, $wr ) = mkhandles; + + my @chunks; + + my $stream = IO::Async::Stream->new( + read_handle => $rd, + read_len => 2, + on_read => sub { + my ( $self, $buffref, $eof ) = @_; + push @chunks, $$buffref; + $$buffref = ""; + }, + ); + + $loop->add( $stream ); + + $wr->syswrite( "partial" ); + + wait_for { scalar @chunks }; + + is_deeply( \@chunks, [ "pa" ], '@lines with read_len=2 without read_all' ); + + wait_for { @chunks == 4 }; + + is_deeply( \@chunks, [ "pa", "rt", "ia", "l" ], '@lines finally with read_len=2 without read_all' ); + + undef @chunks; + $stream->configure( read_all => 1 ); + + $wr->syswrite( "partial" ); + + wait_for { scalar @chunks }; + + is_deeply( \@chunks, [ "pa", "rt", "ia", "l" ], '@lines with read_len=2 with read_all' ); + + $loop->remove( $stream ); +} + +{ + my ( $rd, $wr ) = mkhandles; + + my $no_on_read_stream; + ok( !exception { $no_on_read_stream = IO::Async::Stream->new( read_handle => $rd ) }, + 'Allowed to construct a Stream without an on_read handler' ); + ok( exception { $loop->add( $no_on_read_stream ) }, + 'Not allowed to add an on_read-less Stream to a Loop' ); +} + +# Subclass +my @sub_lines; + +{ + my ( $rd, $wr ) = mkhandles; + + my $stream = TestStream->new( + read_handle => $rd, + ); + + ok( defined $stream, 'reading subclass $stream defined' ); + isa_ok( $stream, "IO::Async::Stream", 'reading $stream isa IO::Async::Stream' ); + + is_oneref( $stream, 'subclass $stream has refcount 1 initially' ); + + $loop->add( $stream ); + + is_refcount( $stream, 2, 'subclass $stream has refcount 2 after adding to Loop' ); + + $wr->syswrite( "message\n" ); + + is_deeply( \@sub_lines, [], '@sub_lines before wait' ); + + wait_for { scalar @sub_lines }; + + is_deeply( \@sub_lines, [ "message\n" ], '@sub_lines after wait' ); + + $loop->remove( $stream ); +} + +# Dynamic on_read chaining +{ + my ( $rd, $wr ) = mkhandles; + + my $outer_count = 0; + my $inner_count = 0; + + my $record; + + my $stream = IO::Async::Stream->new( + read_handle => $rd, + on_read => sub { + my ( $self, $buffref, $eof ) = @_; + $outer_count++; + + return 0 unless $$buffref =~ s/^(.*\n)//; + + my $length = $1; + + return sub { + my ( $self, $buffref, $eof ) = @_; + $inner_count++; + + return 0 unless length $$buffref >= $length; + + $record = substr( $$buffref, 0, $length, "" ); + + return undef; + } + }, + ); + + is_oneref( $stream, 'dynamic reading $stream has refcount 1 initially' ); + + $loop->add( $stream ); + + $wr->syswrite( "11" ); # No linefeed yet + wait_for { $outer_count > 0 }; + is( $outer_count, 1, '$outer_count after idle' ); + is( $inner_count, 0, '$inner_count after idle' ); + + $wr->syswrite( "\n" ); + wait_for { $inner_count > 0 }; + is( $outer_count, 2, '$outer_count after received length' ); + is( $inner_count, 1, '$inner_count after received length' ); + + $wr->syswrite( "Hello " ); + wait_for { $inner_count > 1 }; + is( $outer_count, 2, '$outer_count after partial body' ); + is( $inner_count, 2, '$inner_count after partial body' ); + + $wr->syswrite( "world" ); + wait_for { $inner_count > 2 }; + is( $outer_count, 3, '$outer_count after complete body' ); + is( $inner_count, 3, '$inner_count after complete body' ); + is( $record, "Hello world", '$record after complete body' ); + + $loop->remove( $stream ); + + is_oneref( $stream, 'dynamic reading $stream has refcount 1 finally' ); +} + +# ->push_on_read +{ + my ( $rd, $wr ) = mkhandles; + + my $base; + my $stream = IO::Async::Stream->new( read_handle => $rd, + on_read => sub { + my ( $self, $buffref ) = @_; + $base = $$buffref; $$buffref = ""; + return 0; + }, + ); + + $loop->add( $stream ); + + my $firstline; + $stream->push_on_read( + sub { + my ( $stream, $buffref, $eof ) = @_; + return 0 unless $$buffref =~ s/(.*)\n//; + $firstline = $1; + return undef; + } + ); + + my $eightbytes; + $stream->push_on_read( + sub { + my ( $stream, $buffref, $eof ) = @_; + return 0 unless length $$buffref >= 8; + $eightbytes = substr( $$buffref, 0, 8, "" ); + return undef; + } + ); + + $wr->syswrite( "The first line\nABCDEFGHIJK" ); + + wait_for { defined $firstline and defined $eightbytes }; + + is( $firstline, "The first line", '$firstline from ->push_on_read CODE' ); + is( $eightbytes, "ABCDEFGH", '$eightbytes from ->push_on_read CODE' ); + is( $base, "IJK", '$base from ->push_on_read CODE' ); + + $loop->remove( $stream ); +} + +# EOF +{ + my ( $rd, $wr ) = mkhandles; + + my $eof = 0; + my $partial; + + my $stream = IO::Async::Stream->new( read_handle => $rd, + on_read => sub { + my ( undef, $buffref, $eof ) = @_; + $partial = $$buffref if $eof; + return 0; + }, + on_read_eof => sub { $eof++ }, + ); + + $loop->add( $stream ); + + $wr->syswrite( "Incomplete" ); + + $wr->close; + + ok( !$stream->is_read_eof, '$stream ->is_read_eof before wait' ); + is( $eof, 0, 'EOF indication before wait' ); + + wait_for { $eof }; + + ok( $stream->is_read_eof, '$stream ->is_read_eof after wait' ); + is( $eof, 1, 'EOF indication after wait' ); + is( $partial, "Incomplete", 'EOF stream retains partial input' ); + + ok( !defined $stream->loop, 'EOF stream no longer member of Loop' ); + ok( !defined $stream->read_handle, 'Stream no longer has a read_handle' ); +} + +# Disabled close_on_read_eof +{ + my ( $rd, $wr ) = mkhandles; + + my $eof = 0; + my $partial; + + my $stream = IO::Async::Stream->new( read_handle => $rd, + on_read => sub { + my ( undef, $buffref, $eof ) = @_; + $partial = $$buffref if $eof; + return 0; + }, + on_read_eof => sub { $eof++ }, + close_on_read_eof => 0, + ); + + $loop->add( $stream ); + + $wr->syswrite( "Incomplete" ); + + $wr->close; + + is( $eof, 0, 'EOF indication before wait' ); + + wait_for { $eof }; + + is( $eof, 1, 'EOF indication after wait' ); + is( $partial, "Incomplete", 'EOF stream retains partial input' ); + + ok( defined $stream->loop, 'EOF stream still member of Loop' ); + ok( defined $stream->read_handle, 'Stream still has a read_handle' ); +} + +# Close +{ + my ( $rd, $wr ) = mkhandles; + + my $closed = 0; + my $loop_during_closed; + + my $stream = IO::Async::Stream->new( read_handle => $rd, + on_read => sub { }, + on_closed => sub { + my ( $self ) = @_; + $closed = 1; + $loop_during_closed = $self->loop; + }, + ); + + is_oneref( $stream, 'closing $stream has refcount 1 initially' ); + + $loop->add( $stream ); + + is_refcount( $stream, 2, 'closing $stream has refcount 2 after adding to Loop' ); + + is( $closed, 0, 'closed before close' ); + + $stream->close; + + is( $closed, 1, 'closed after close' ); + is( $loop_during_closed, $loop, 'loop during closed' ); + + ok( !defined $stream->loop, 'Stream no longer member of Loop' ); + + is_oneref( $stream, 'closing $stream refcount 1 finally' ); +} + +# ->read Futures +{ + my ( $rd, $wr ) = mkhandles; + + my $stream = IO::Async::Stream->new( read_handle => $rd, + on_read => sub { + my ( $self, $buffref ) = @_; + die "Base on_read invoked with data in the buffer" if length $$buffref; + return 0; + }, + ); + + $loop->add( $stream ); + + my $f_atmost = $stream->read_atmost( 256 ); + + $wr->syswrite( "Some data\n" ); + wait_for { $f_atmost->is_ready }; + + is( scalar $f_atmost->get, "Some data\n", '->read_atmost' ); + + my $f_exactly = $stream->read_exactly( 4 ); + my $f_until_qr = $stream->read_until( qr/[A-Z][a-z]*/ ); + my $f_until_str = $stream->read_until( "\n" ); + + $wr->syswrite( "Here is the First line of input\n" ); + + wait_for { $f_exactly->is_ready and $f_until_qr->is_ready and $f_until_str->is_ready }; + + is( scalar $f_exactly->get, "Here", '->read_exactly' ); + is( scalar $f_until_qr->get, " is the First", '->read_until regexp' ); + is( scalar $f_until_str->get, " line of input\n", '->read_until str' ); + + my $f_first = $stream->read_until( "\n" ); + my $f_second = $stream->read_until( "\n" ); + $f_first->cancel; + + $wr->syswrite( "For the second\n" ); + + wait_for { $f_second->is_ready }; + + is( scalar $f_second->get, "For the second\n", 'Second ->read_until recieves data after first is ->cancelled' ); + + my $f_until_eof = $stream->read_until_eof; + + $wr->syswrite( "And the rest of it" ); + $wr->close; + + wait_for { $f_until_eof->is_ready }; + + is( scalar $f_until_eof->get, "And the rest of it", '->read_until_eof' ); + + # No need to remove as ->close did it +} + +# RT101774 +{ + my ( $rd, $wr ) = mkhandles; + + my $stream = IO::Async::Stream->new( read_handle => $rd, + on_read => sub { 0 }, + ); + + $loop->add( $stream ); + + $wr->syswrite( "lalaLALA" ); + + my $f = $stream->read_exactly( 4 )->then( sub { + $stream->read_exactly( 4 ); + }); + + wait_for { $f->is_ready }; + + is( scalar $f->get, "LALA", 'chained ->read_exactly' ); + + $loop->remove( $stream ); +} + +# watermarks +{ + my ( $rd, $wr ) = mkhandles; + + my $high_hit = 0; + my $low_hit = 0; + + my $stream = IO::Async::Stream->new( + read_handle => $rd, + on_read => sub { 0 }, # we'll work by Futures + read_high_watermark => 8, + read_low_watermark => 4, + on_read_high_watermark => sub { $high_hit++ }, + on_read_low_watermark => sub { $low_hit++ }, + ); + + $loop->add( $stream ); + + $wr->syswrite( "1234567890" ); + + wait_for { $high_hit }; + ok( 1, "Reading too much hits high watermark" ); + + is( $stream->read_exactly( 8 )->get, "12345678", 'Stream->read_exactly yields bytes' ); + + is( $low_hit, 1, 'Low watermark hit after ->read' ); +} + +# Errors +{ + my ( $rd, $wr ) = mkhandles; + $wr->syswrite( "X" ); # ensuring $rd is read-ready + + no warnings 'redefine'; + local *IO::Handle::sysread = sub { + $! = ECONNRESET; + return undef; + }; + + my $read_errno; + + my $stream = IO::Async::Stream->new( + read_handle => $rd, + on_read => sub {}, + on_read_error => sub { ( undef, $read_errno ) = @_ }, + ); + + $loop->add( $stream ); + + wait_for { defined $read_errno }; + + cmp_ok( $read_errno, "==", ECONNRESET, 'errno after failed read' ); + + my $f = $stream->read_atmost( 256 ); + + wait_for { $f->is_ready }; + cmp_ok( ( $f->failure )[-1], "==", ECONNRESET, 'failure from ->read_atmost after failed read' ); + + $loop->remove( $stream ); +} + +{ + binmode STDIN; # Avoid harmless warning in case -CS is in effect + my $stream = IO::Async::Stream->new_for_stdin; + is( $stream->read_handle, \*STDIN, 'Stream->new_for_stdin->read_handle is STDIN' ); +} + +done_testing; + +package TestStream; +use base qw( IO::Async::Stream ); + +sub on_read +{ + my $self = shift; + my ( $buffref, $eof ) = @_; + + return 0 unless $$buffref =~ s/^(.*\n)//; + + push @sub_lines, $1; + return 1; +} diff --git a/t/21stream-2write.t b/t/21stream-2write.t new file mode 100644 index 0000000..b49cea8 --- /dev/null +++ b/t/21stream-2write.t @@ -0,0 +1,479 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Refcount; + +use Errno qw( EAGAIN EWOULDBLOCK ECONNRESET ); + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Stream; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +sub mkhandles +{ + my ( $rd, $wr ) = IO::Async::OS->pipepair or die "Cannot pipe() - $!"; + # Need handles in nonblocking mode + $rd->blocking( 0 ); + $wr->blocking( 0 ); + + return ( $rd, $wr ); +} + +# useful test function +sub read_data +{ + my ( $s ) = @_; + + my $buffer; + my $ret = $s->sysread( $buffer, 8192 ); + + return $buffer if( defined $ret && $ret > 0 ); + die "Socket closed" if( defined $ret && $ret == 0 ); + return "" if $! == EAGAIN or $! == EWOULDBLOCK; + die "Cannot sysread() - $!"; +} + +{ + my ( $rd, $wr ) = mkhandles; + + my $empty; + + my $stream = IO::Async::Stream->new( + write_handle => $wr, + on_outgoing_empty => sub { $empty = 1 }, + ); + + ok( defined $stream, 'writing $stream defined' ); + isa_ok( $stream, "IO::Async::Stream", 'writing $stream isa IO::Async::Stream' ); + + is_oneref( $stream, 'writing $stream has refcount 1 initially' ); + + $loop->add( $stream ); + + is_refcount( $stream, 2, 'writing $stream has refcount 2 after adding to Loop' ); + + ok( !$stream->want_writeready, 'want_writeready before write' ); + $stream->write( "message\n" ); + + ok( $stream->want_writeready, 'want_writeready after write' ); + + wait_for { $empty }; + + ok( !$stream->want_writeready, 'want_writeready after wait' ); + is( $empty, 1, '$empty after writing buffer' ); + + is( read_data( $rd ), "message\n", 'data after writing buffer' ); + + my $written = 0; + my $flushed; + + my $f = $stream->write( "hello again\n", + on_write => sub { + is( $_[0], $stream, 'on_write $_[0] is $stream' ); + $written += $_[1]; + }, + on_flush => sub { + is( $_[0], $stream, 'on_flush $_[0] is $stream' ); + $flushed++ + }, + ); + + ok( !$f->is_ready, '->write future not yet ready' ); + + wait_for { $flushed }; + + ok( $f->is_ready, '->write future is ready after flush' ); + is( $written, 12, 'on_write given total write length after flush' ); + is( read_data( $rd ), "hello again\n", 'flushed data does get flushed' ); + + $flushed = 0; + $stream->write( "", on_flush => sub { $flushed++ } ); + wait_for { $flushed }; + + ok( 1, "write empty data with on_flush" ); + + $stream->configure( autoflush => 1 ); + $stream->write( "immediate\n" ); + + ok( !$stream->want_writeready, 'not want_writeready after autoflush write' ); + is( read_data( $rd ), "immediate\n", 'data after autoflush write' ); + + $stream->configure( autoflush => 0 ); + $stream->write( "partial " ); + $stream->configure( autoflush => 1 ); + $stream->write( "data\n" ); + + ok( !$stream->want_writeready, 'not want_writeready after split autoflush write' ); + is( read_data( $rd ), "partial data\n", 'data after split autoflush write' ); + + is_refcount( $stream, 2, 'writing $stream has refcount 2 before removing from Loop' ); + + $loop->remove( $stream ); + + is_oneref( $stream, 'writing $stream refcount 1 finally' ); +} + +# Abstract writing with writer function +{ + my ( $rd, $wr ) = mkhandles; + my $buffer; + + my $stream = IO::Async::Stream->new( + write_handle => $wr, + writer => sub { + my $self = shift; + $buffer .= substr( $_[1], 0, $_[2], "" ); + return $_[2]; + }, + ); + + $loop->add( $stream ); + + my $flushed; + $stream->write( "Some data for abstract buffer\n", on_flush => sub { $flushed++ } ); + + wait_for { $flushed }; + + is( $buffer, "Some data for abstract buffer\n", '$buffer after ->write to stream with abstract writer' ); + + $loop->remove( $stream ); +} + +# ->want_writeready_for_read +{ + my ( $rd, $wr ) = mkhandles; + + my $reader_called; + my $stream = IO::Async::Stream->new( + handle => $wr, + on_read => sub { return 0; }, # ignore reading + reader => sub { $reader_called++; $! = EAGAIN; return undef }, + ); + + $loop->add( $stream ); + + $loop->loop_once( 0.1 ); # haaaaack + + ok( !$reader_called, 'reader not yet called before ->want_writeready_for_read' ); + + $stream->want_writeready_for_read( 1 ); + + wait_for { $reader_called }; + + ok( $reader_called, 'reader now invoked with ->want_writeready_for_read' ); + + $loop->remove( $stream ); +} + +# on_writeable_{start,stop} +{ + my ( $rd, $wr ) = mkhandles; + my $buffer; + + my $writeable; + my $unwriteable; + my $emulate_writeable = 0; + my $stream = IO::Async::Stream->new( + write_handle => $wr, + writer => sub { + my $self = shift; + $! = EAGAIN, return undef unless $emulate_writeable; + + $buffer .= substr( $_[1], 0, $_[2], "" ); + return $_[2]; + }, + on_writeable_start => sub { $writeable++ }, + on_writeable_stop => sub { $unwriteable++ }, + ); + + $loop->add( $stream ); + + $stream->write( "Something" ); + + wait_for { $unwriteable }; + + $emulate_writeable = 1; + + wait_for { $writeable }; + + is( $buffer, "Something", '$buffer after emulated EAGAIN' ); + + $loop->remove( $stream ); +} + +{ + my ( $rd, $wr ) = mkhandles; + + my $stream = IO::Async::Stream->new( + write_handle => $wr, + write_len => 2, + ); + + $loop->add( $stream ); + + $stream->write( "partial" ); + + $loop->loop_once( 0.1 ); + + is( read_data( $rd ), "pa", 'data after writing buffer with write_len=2 without write_all'); + + $loop->loop_once( 0.1 ) for 1 .. 3; + + is( read_data( $rd ), "rtial", 'data finally after writing buffer with write_len=2 without write_all' ); + + $stream->configure( write_all => 1 ); + + $stream->write( "partial" ); + + $loop->loop_once( 0.1 ); + + is( read_data( $rd ), "partial", 'data after writing buffer with write_len=2 with write_all'); + + $loop->remove( $stream ); +} + +# EOF +SKIP: { + skip "This loop cannot detect hangup condition", 5 unless $loop->_CAN_ON_HANGUP; + + my ( $rd, $wr ) = mkhandles; + + local $SIG{PIPE} = "IGNORE"; + + my $eof = 0; + + my $stream = IO::Async::Stream->new( write_handle => $wr, + on_write_eof => sub { $eof++ }, + ); + + $loop->add( $stream ); + + my $write_future = $stream->write( "Junk" ); + + $rd->close; + + ok( !$stream->is_write_eof, '$stream->is_write_eof before wait' ); + is( $eof, 0, 'EOF indication before wait' ); + + wait_for { $eof }; + + ok( $stream->is_write_eof, '$stream->is_write_eof after wait' ); + is( $eof, 1, 'EOF indication after wait' ); + + ok( !defined $stream->loop, 'EOF stream no longer member of Loop' ); + + ok( $write_future->is_ready,'write future ready after stream closed' ); + ok( $write_future->is_failed,'write future failed after stream closed' ); +} + +# Close +{ + my ( $rd, $wr ) = mkhandles; + + my $closed = 0; + my $loop_during_closed; + + my $stream = IO::Async::Stream->new( write_handle => $wr, + on_closed => sub { + my ( $self ) = @_; + $closed = 1; + $loop_during_closed = $self->loop; + }, + ); + + is_oneref( $stream, 'closing $stream has refcount 1 initially' ); + + $stream->write( "hello" ); + + $loop->add( $stream ); + + is_refcount( $stream, 2, 'closing $stream has refcount 2 after adding to Loop' ); + + is( $closed, 0, 'closed before close' ); + + $stream->close_when_empty; + + is( $closed, 0, 'closed after close' ); + + wait_for { $closed }; + + is( $closed, 1, 'closed after wait' ); + is( $loop_during_closed, $loop, 'loop during closed' ); + + ok( !defined $stream->loop, 'Stream no longer member of Loop' ); + + is_oneref( $stream, 'closing $stream refcount 1 finally' ); +} + +# ->write( Future ) +{ + my ( $rd, $wr ) = mkhandles; + my $stream = IO::Async::Stream->new( + write_handle => $wr, + ); + $loop->add( $stream ); + + my $written = 0; + my $flushed; + $stream->write( + my $future = $loop->new_future, + on_write => sub { $written += $_[1] }, + on_flush => sub { $flushed++ }, + ); + + $loop->loop_once( 0.1 ); + is( read_data( $rd ), "", 'stream idle before Future completes' ); + + $future->done( "some data to write" ); + + wait_for { $flushed }; + + is( $written, 18, 'stream written by Future completion invokes on_write' ); + + is( read_data( $rd ), "some data to write", 'stream written by Future completion' ); + + $loop->remove( $stream ); +} + +# ->write( CODE ) +{ + my ( $rd, $wr ) = mkhandles; + my $stream = IO::Async::Stream->new( + write_handle => $wr, + ); + $loop->add( $stream ); + + my $done; + my $written = 0; + my $flushed; + + $stream->write( + sub { + is( $_[0], $stream, 'Writersub $_[0] is $stream' ); + return $done++ ? undef : "a lazy message\n"; + }, + on_write => sub { $written += $_[1] }, + on_flush => sub { $flushed++ }, + ); + + $flushed = 0; + wait_for { $flushed }; + + is( $written, 15, 'stream written by generator CODE invokes on_write' ); + + is( read_data( $rd ), "a lazy message\n", 'lazy data was written' ); + + my @chunks = ( "some ", "message chunks ", "here\n" ); + + $stream->write( + sub { + return shift @chunks; + }, + on_flush => sub { $flushed++ }, + ); + + $flushed = 0; + wait_for { $flushed }; + + is( read_data( $rd ), "some message chunks here\n", 'multiple lazy data was written' ); + + $loop->remove( $stream ); +} + +# ->write mixed returns +{ + my ( $rd, $wr ) = mkhandles; + my $stream = IO::Async::Stream->new( + write_handle => $wr, + ); + $loop->add( $stream ); + + my $flushed; + $stream->write( my $future = $loop->new_future, on_flush => sub { $flushed++ } ); + + my $once = 0; + $future->done( sub { + return $once++ ? undef : ( $future = $loop->new_future ); + }); + + wait_for { $once }; + + $future->done( "Eventual string" ); + + wait_for { $flushed }; + + is( read_data( $rd ), "Eventual string", 'multiple lazy data was written' ); + + $loop->remove( $stream ); +} + +{ + my ( $rd, $wr ) = mkhandles; + + my $stream = IO::Async::Stream->new; + + my $flushed; + + $stream->write( "Prequeued data", on_flush => sub { $flushed++ } ); + + $stream->configure( write_handle => $wr ); + + $loop->add( $stream ); + + wait_for { $flushed }; + + ok( 1, 'prequeued data gets flushed' ); + + is( read_data( $rd ), "Prequeued data", 'prequeued data gets written' ); + + $loop->remove( $stream ); +} + +# Errors +{ + my ( $rd, $wr ) = mkhandles; + + no warnings 'redefine'; + local *IO::Handle::syswrite = sub { + $! = ECONNRESET; + return undef; + }; + + my $write_errno; + + my $stream = IO::Async::Stream->new( + write_handle => $wr, + on_write_error => sub { ( undef, $write_errno ) = @_ }, + ); + + $loop->add( $stream ); + + my $write_future = $stream->write( "hello" ); + + wait_for { defined $write_errno }; + + cmp_ok( $write_errno, "==", ECONNRESET, 'errno after failed write' ); + + ok( $write_future->is_ready,'write future ready after failed write' ); + ok( $write_future->is_failed,'write future failed after failed write' ); + + $loop->remove( $stream ); +} + +{ + my $stream = IO::Async::Stream->new_for_stdout; + is( $stream->write_handle, \*STDOUT, 'Stream->new_for_stdout->write_handle is STDOUT' ); +} + +done_testing; diff --git a/t/21stream-3split.t b/t/21stream-3split.t new file mode 100644 index 0000000..1fd99e4 --- /dev/null +++ b/t/21stream-3split.t @@ -0,0 +1,187 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use IO::File; +use Errno qw( EAGAIN EWOULDBLOCK ); + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Stream; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; +my ( $S3, $S4 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + +# Need sockets in nonblocking mode +$_->blocking( 0 ) for $S1, $S2, $S3, $S4; + +# useful test function +sub read_data +{ + my ( $s ) = @_; + + my $buffer; + my $ret = $s->sysread( $buffer, 8192 ); + + return $buffer if( defined $ret && $ret > 0 ); + die "Socket closed" if( defined $ret && $ret == 0 ); + return "" if $! == EAGAIN or $! == EWOULDBLOCK; + die "Cannot sysread() - $!"; +} + +my @lines; + +my $stream = IO::Async::Stream->new( + read_handle => $S2, + write_handle => $S3, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, +); + +is_oneref( $stream, 'split read/write $stream has refcount 1 initially' ); + +undef @lines; + +$loop->add( $stream ); + +is_refcount( $stream, 2, 'split read/write $stream has refcount 2 after adding to Loop' ); + +$stream->write( "message\n" ); + +$loop->loop_once( 0.1 ); + +is( read_data( $S4 ), "message\n", '$S4 receives data from split stream' ); +is( read_data( $S1 ), "", '$S1 empty from split stream' ); + +$S1->syswrite( "reverse\n" ); + +$loop->loop_once( 0.1 ); + +is_deeply( \@lines, [ "reverse\n" ], '@lines on response to split stream' ); + +is_refcount( $stream, 2, 'split read/write $stream has refcount 2 before removing from Loop' ); + +$loop->remove( $stream ); + +is_oneref( $stream, 'split read/write $stream refcount 1 finally' ); + +undef $stream; + +my $buffer = ""; +my $closed; + +$stream = IO::Async::Stream->new( + # No handle yet + on_read => sub { + my ( $self, $buffref, $eof ) = @_; + $buffer .= $$buffref; + $$buffref = ""; + return 0; + }, + on_closed => sub { + my ( $self ) = @_; + $closed = 1; + }, +); + +is_oneref( $stream, 'latehandle $stream has refcount 1 initially' ); + +$loop->add( $stream ); + +is_refcount( $stream, 2, 'latehandle $stream has refcount 2 after adding to Loop' ); + +ok( exception { $stream->write( "some text" ) }, + '->write on stream with no IO handle fails' ); + +$stream->set_handle( $S1 ); + +is_refcount( $stream, 2, 'latehandle $stream has refcount 2 after setting a handle' ); + +$stream->write( "some text" ); + +$loop->loop_once( 0.1 ); + +my $buffer2; +$S2->sysread( $buffer2, 8192 ); + +is( $buffer2, "some text", 'stream-written text appears' ); + +$S2->syswrite( "more text" ); + +wait_for { length $buffer }; + +is( $buffer, "more text", 'stream-read text appears' ); + +$stream->close_when_empty; + +is( $closed, 1, 'closed after close' ); + +ok( !defined $stream->loop, 'Stream no longer member of Loop' ); + +is_oneref( $stream, 'latehandle $stream refcount 1 finally' ); + +# Now try re-opening the stream with a new handle, and check it continues to +# work + +$loop->add( $stream ); + +$stream->set_handle( $S3 ); + +$stream->write( "more text" ); + +$loop->loop_once( 0.1 ); + +undef $buffer2; +$S4->sysread( $buffer2, 8192 ); + +is( $buffer2, "more text", 'stream-written text appears after reopen' ); + +$loop->remove( $stream ); + +undef $stream; + +( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot socketpair - $!"; + +$stream = IO::Async::Stream->new( + handle => $S1, + on_read => sub { }, +); + +$stream->write( "hello" ); + +$loop->add( $stream ); + +is_refcount( $stream, 2, '$stream has two references' ); +undef $stream; # Only ref is now in the Loop + +$S2->close; + +# $S1 should now be both read- and write-ready. +ok( !exception { $loop->loop_once }, 'read+write-ready closed Stream doesn\'t die' ); + +undef $stream; + +binmode STDIN; # Avoid harmless warning in case -CS is in effect +$stream = IO::Async::Stream->new_for_stdio; +is( $stream->read_handle, \*STDIN, 'Stream->new_for_stdio->read_handle is STDIN' ); +is( $stream->write_handle, \*STDOUT, 'Stream->new_for_stdio->write_handle is STDOUT' ); + +done_testing; diff --git a/t/21stream-4encoding.t b/t/21stream-4encoding.t new file mode 100644 index 0000000..cae0cac --- /dev/null +++ b/t/21stream-4encoding.t @@ -0,0 +1,151 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use Errno qw( EAGAIN EWOULDBLOCK ); + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Stream; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +sub mkhandles +{ + my ( $rd, $wr ) = IO::Async::OS->pipepair or die "Cannot pipe() - $!"; + # Need handles in nonblocking mode + $rd->blocking( 0 ); + $wr->blocking( 0 ); + + return ( $rd, $wr ); +} + +# useful test function +sub read_data +{ + my ( $s ) = @_; + + my $buffer; + my $ret = $s->sysread( $buffer, 8192 ); + + return $buffer if( defined $ret && $ret > 0 ); + die "Socket closed" if( defined $ret && $ret == 0 ); + return "" if $! == EAGAIN or $! == EWOULDBLOCK; + die "Cannot sysread() - $!"; +} + +# To test correct multi-byte encoding handling, we'll use a UTF-8 character +# that requires multiple bytes. Furthermore we'll use one that doesn't appear +# in Latin-1 +# +# 'ĉ' [U+0109] - LATIN SMALL LETTER C WITH CIRCUMFLEX +# :0xc4 0x89 + +# Read encoding +{ + my ( $rd, $wr ) = mkhandles; + + my $read = ""; + my $stream = IO::Async::Stream->new( + read_handle => $rd, + encoding => "UTF-8", + on_read => sub { + $read = ${$_[1]}; + ${$_[1]} = ""; + return 0; + }, + ); + + $loop->add( $stream ); + + $wr->syswrite( "\xc4\x89" ); + + wait_for { length $read }; + + is( $read, "\x{109}", 'Unicode characters read by on_read' ); + + $wr->syswrite( "\xc4\x8a\xc4" ); + + $read = ""; + wait_for { length $read }; + + is( $read, "\x{10a}", 'Partial UTF-8 character not yet visible' ); + + $wr->syswrite( "\x8b" ); + + $read = ""; + wait_for { length $read }; + + is( $read, "\x{10b}", 'Partial UTF-8 character visible after completion' ); + + # An invalid sequence + $wr->syswrite( "\xc4!" ); + + $read = ""; + wait_for { length $read }; + + is( $read, "\x{fffd}!", 'Invalid UTF-8 byte yields U+FFFD' ); + + $loop->remove( $stream ); +} + +# Write encoding +{ + my ( $rd, $wr ) = mkhandles; + + my $stream = IO::Async::Stream->new( + write_handle => $wr, + encoding => "UTF-8", + ); + + $loop->add( $stream ); + + my $flushed; + $stream->write( "\x{109}", on_flush => sub { $flushed++ } ); + + wait_for { $flushed }; + + is( read_data( $rd ), "\xc4\x89", 'UTF-8 bytes written by ->write string' ); + + $stream->configure( write_len => 1 ); + + $stream->write( "\x{109}" ); + + my $byte; + + $loop->loop_once while !length( $byte = read_data( $rd ) ); + is( $byte, "\xc4", 'First UTF-8 byte written with write_len 1' ); + + $loop->loop_once while !length( $byte = read_data( $rd ) ); + is( $byte, "\x89", 'Remaining UTF-8 byte written with write_len 1' ); + + $flushed = 0; + $stream->write( Future->done( "\x{10a}" ), on_flush => sub { $flushed++ } ); + + wait_for { $flushed }; + + is( read_data( $rd ), "\xc4\x8a", 'UTF-8 bytes written by ->write Future' ); + + $flushed = 0; + my $once = 0; + $stream->write( sub { $once++ ? undef : "\x{10b}" }, on_flush => sub { $flushed++ } ); + + wait_for { $flushed }; + + is( read_data( $rd ), "\xc4\x8b", 'UTF-8 bytes written by ->write CODE' ); + + $loop->remove( $stream ); +} + +done_testing; diff --git a/t/22timer-absolute.t b/t/22timer-absolute.t new file mode 100644 index 0000000..6192bec --- /dev/null +++ b/t/22timer-absolute.t @@ -0,0 +1,143 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Refcount; +use t::TimeAbout; + +use Time::HiRes qw( time ); + +use IO::Async::Timer::Absolute; + +use IO::Async::Loop; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my $expired; + my @eargs; + + my $timer = IO::Async::Timer::Absolute->new( + time => time + 2 * AUT, + + on_expire => sub { @eargs = @_; $expired = 1 }, + ); + + ok( defined $timer, '$timer defined' ); + isa_ok( $timer, "IO::Async::Timer", '$timer isa IO::Async::Timer' ); + + is_oneref( $timer, '$timer has refcount 1 initially' ); + + $loop->add( $timer ); + + is_refcount( $timer, 2, '$timer has refcount 2 after adding to Loop' ); + + ok( $timer->is_running, 'Started Timer is running' ); + + time_about( sub { wait_for { $expired } }, 2, 'Timer works' ); + is_deeply( \@eargs, [ $timer ], 'on_expire args' ); + + ok( !$timer->is_running, 'Expired Timer is no longer running' ); + + undef @eargs; + + is_refcount( $timer, 2, '$timer has refcount 2 before removing from Loop' ); + + $loop->remove( $timer ); + + is_oneref( $timer, '$timer has refcount 1 after removing from Loop' ); +} + +{ + my $expired; + + my $timer = IO::Async::Timer::Absolute->new( + time => time + 2 * AUT, + on_expire => sub { $expired++ }, + ); + + $loop->add( $timer ); + $loop->remove( $timer ); + + $loop->loop_once( 3 * AUT ); + + ok( !$expired, "Removed Timer does not expire" ); +} + +{ + my $expired; + + my $timer = IO::Async::Timer::Absolute->new( + time => time + 5 * AUT, + on_expire => sub { $expired++ }, + ); + + $loop->add( $timer ); + + $timer->configure( time => time + 1 * AUT ); + + time_about( sub { wait_for { $expired } }, 1, 'Reconfigured timer works' ); + + $loop->remove( $timer ); +} + +{ + my $timer = IO::Async::Timer::Absolute->new( + time => time + 1 * AUT, + on_expire => sub { die "Test failed to replace expiry handler" }, + ); + + $loop->add( $timer ); + + my $new_expired; + $timer->configure( on_expire => sub { $new_expired = 1 } ); + + time_about( sub { wait_for { $new_expired } }, 1, 'Reconfigured timer on_expire works' ); + + $loop->remove( $timer ); +} + +## Subclass + +my $sub_expired; +{ + my $timer = TestTimer->new( + time => time + 2 * AUT, + ); + + ok( defined $timer, 'subclass $timer defined' ); + isa_ok( $timer, "IO::Async::Timer", 'subclass $timer isa IO::Async::Timer' ); + + is_oneref( $timer, 'subclass $timer has refcount 1 initially' ); + + $loop->add( $timer ); + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 after adding to Loop' ); + + ok( $timer->is_running, 'Started subclass Timer is running' ); + + time_about( sub { wait_for { $sub_expired } }, 2, 'subclass Timer works' ); + + ok( !$timer->is_running, 'Expired subclass Timer is no longer running' ); + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 before removing from Loop' ); + + $loop->remove( $timer ); + + is_oneref( $timer, 'subclass $timer has refcount 1 after removing from Loop' ); +} + +done_testing; + +package TestTimer; +use base qw( IO::Async::Timer::Absolute ); + +sub on_expire { $sub_expired = 1 } diff --git a/t/22timer-countdown.t b/t/22timer-countdown.t new file mode 100644 index 0000000..db8b49c --- /dev/null +++ b/t/22timer-countdown.t @@ -0,0 +1,257 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; +use t::TimeAbout; + +use Time::HiRes qw( time ); + +use IO::Async::Timer::Countdown; + +use IO::Async::Loop; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my $expired; + my @eargs; + + my $timer = IO::Async::Timer::Countdown->new( + delay => 2 * AUT, + + on_expire => sub { @eargs = @_; $expired = 1 }, + ); + + ok( defined $timer, '$timer defined' ); + isa_ok( $timer, "IO::Async::Timer", '$timer isa IO::Async::Timer' ); + + is_oneref( $timer, '$timer has refcount 1 initially' ); + + $loop->add( $timer ); + + is_refcount( $timer, 2, '$timer has refcount 2 after adding to Loop' ); + + ok( !$timer->is_running, 'New Timer is no yet running' ); + ok( !$timer->is_expired, 'New Timer is no yet expired' ); + + is( $timer->start, $timer, '$timer->start returns $timer' ); + + is_refcount( $timer, 2, '$timer has refcount 2 after starting' ); + + ok( $timer->is_running, 'Started Timer is running' ); + ok( !$timer->is_expired, 'Started Timer not yet expired' ); + + time_about( sub { wait_for { $expired } }, 2, 'Timer works' ); + is_deeply( \@eargs, [ $timer ], 'on_expire args' ); + + ok( !$timer->is_running, 'Expired Timer is no longer running' ); + ok( $timer->is_expired, 'Expired Timer now expired' ); + + undef @eargs; + + is_refcount( $timer, 2, '$timer has refcount 2 before removing from Loop' ); + + $loop->remove( $timer ); + + is_oneref( $timer, '$timer has refcount 1 after removing from Loop' ); + + undef $expired; + + is( $timer->start, $timer, '$timer->start out of a Loop returns $timer' ); + + $loop->add( $timer ); + + ok( $timer->is_running, 'Re-started Timer is running' ); + ok( !$timer->is_expired, 'Re-started Timer not yet expired' ); + + time_about( sub { wait_for { $expired } }, 2, 'Timer works a second time' ); + + ok( !$timer->is_running, '2nd-time expired Timer is no longer running' ); + ok( $timer->is_expired, '2nd-time expired Timer now expired' ); + + undef $expired; + $timer->start; + + $loop->loop_once( 1 * AUT ); + + $timer->stop; + + $timer->stop; + + ok( 1, "Timer can be stopped a second time" ); + + $loop->loop_once( 2 * AUT ); + + ok( !$expired, "Stopped timer doesn't expire" ); + + undef $expired; + $timer->start; + + $loop->loop_once( 1 * AUT ); + + my $now = time; + $timer->reset; + + $loop->loop_once( 1.5 * AUT ); + + ok( !$expired, "Reset Timer hasn't expired yet" ); + + wait_for { $expired }; + my $took = (time - $now) / AUT; + + cmp_ok( $took, '>', 1.5, "Timer has now expired took at least 1.5" ); + cmp_ok( $took, '<', 2.5, "Timer has now expired took no more than 2.5" ); + + $loop->remove( $timer ); + + undef @eargs; + + is_oneref( $timer, 'Timer has refcount 1 finally' ); +} + +{ + my $timer = IO::Async::Timer::Countdown->new( + delay => 2 * AUT, + on_expire => sub { }, + ); + + $loop->add( $timer ); + + $timer->start; + + $loop->remove( $timer ); + + $loop->loop_once( 3 * AUT ); + + ok( !$timer->is_expired, "Removed Timer does not expire" ); +} + +{ + my $timer = IO::Async::Timer::Countdown->new( + delay => 2 * AUT, + on_expire => sub { }, + ); + + $timer->start; + + $loop->add( $timer ); + + ok( $timer->is_running, 'Pre-started Timer is running after adding' ); + + time_about( sub { wait_for { $timer->is_expired } }, 2, 'Pre-started Timer works' ); + + $loop->remove( $timer ); +} + +{ + my $timer = IO::Async::Timer::Countdown->new( + delay => 2 * AUT, + on_expire => sub { }, + ); + + $timer->start; + $timer->stop; + + $loop->add( $timer ); + + $loop->loop_once( 3 * AUT ); + + ok( !$timer->is_expired, "start/stopped Timer doesn't expire" ); + + $loop->remove( $timer ); +} + +{ + my $timer = IO::Async::Timer::Countdown->new( + delay => 2 * AUT, + on_expire => sub { }, + ); + + $loop->add( $timer ); + + $timer->configure( delay => 1 * AUT ); + + $timer->start; + + time_about( sub { wait_for { $timer->is_expired } }, 1, 'Reconfigured timer delay works' ); + + my $expired; + $timer->configure( on_expire => sub { $expired = 1 } ); + + $timer->start; + + time_about( sub { wait_for { $expired } }, 1, 'Reconfigured timer on_expire works' ); + + $timer->start; + ok( exception { $timer->configure( delay => 5 ); }, + 'Configure a running timer fails' ); + + $loop->remove( $timer ); +} + +{ + my $timer = IO::Async::Timer::Countdown->new( + delay => 1 * AUT, + remove_on_expire => 1, + + on_expire => sub { }, + ); + + $loop->add( $timer ); + $timer->start; + + time_about( sub { wait_for { $timer->is_expired } }, 1, 'remove_on_expire Timer' ); + + is( $timer->loop, undef, 'remove_on_expire Timer removed from Loop after expire' ); +} + +## Subclass + +my $sub_expired; +{ + my $timer = TestTimer->new( + delay => 2 * AUT, + ); + + ok( defined $timer, 'subclass $timer defined' ); + isa_ok( $timer, "IO::Async::Timer", 'subclass $timer isa IO::Async::Timer' ); + + is_oneref( $timer, 'subclass $timer has refcount 1 initially' ); + + $loop->add( $timer ); + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 after adding to Loop' ); + + $timer->start; + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 after starting' ); + + ok( $timer->is_running, 'Started subclass Timer is running' ); + + time_about( sub { wait_for { $sub_expired } }, 2, 'subclass Timer works' ); + + ok( !$timer->is_running, 'Expired subclass Timer is no longer running' ); + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 before removing from Loop' ); + + $loop->remove( $timer ); + + is_oneref( $timer, 'subclass $timer has refcount 1 after removing from Loop' ); +} + +done_testing; + +package TestTimer; +use base qw( IO::Async::Timer::Countdown ); + +sub on_expire { $sub_expired = 1 } diff --git a/t/22timer-periodic.t b/t/22timer-periodic.t new file mode 100644 index 0000000..a2fc28e --- /dev/null +++ b/t/22timer-periodic.t @@ -0,0 +1,233 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; +use t::TimeAbout; + +use IO::Async::Timer::Periodic; + +use IO::Async::Loop; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my $tick = 0; + my @targs; + + my $timer = IO::Async::Timer::Periodic->new( + interval => 2 * AUT, + + on_tick => sub { @targs = @_; $tick++ }, + ); + + ok( defined $timer, '$timer defined' ); + isa_ok( $timer, "IO::Async::Timer", '$timer isa IO::Async::Timer' ); + + is_oneref( $timer, '$timer has refcount 1 initially' ); + + $loop->add( $timer ); + + is_refcount( $timer, 2, '$timer has refcount 2 after adding to Loop' ); + + is( $timer->start, $timer, '$timer->start returns $timer' ); + + is_refcount( $timer, 2, '$timer has refcount 2 after starting' ); + + ok( $timer->is_running, 'Started Timer is running' ); + + time_about( sub { wait_for { $tick == 1 } }, 2, 'Timer works' ); + is_deeply( \@targs, [ $timer ], 'on_tick args' ); + + ok( $timer->is_running, 'Timer is still running' ); + + time_about( sub { wait_for { $tick == 2 } }, 2, 'Timer works a second time' ); + + $loop->loop_once( 1 * AUT ); + + $timer->stop; + + $timer->stop; + + ok( 1, "Timer can be stopped a second time" ); + + $loop->loop_once( 2 * AUT ); + + ok( $tick == 2, "Stopped timer doesn't tick" ); + + undef @targs; + + is_refcount( $timer, 2, '$timer has refcount 2 before removing from Loop' ); + + $loop->remove( $timer ); + + is_oneref( $timer, '$timer has refcount 1 after removing from Loop' ); + + ok( !$timer->is_running, 'Removed timer not running' ); + + $loop->add( $timer ); + + $timer->configure( interval => 1 * AUT ); + + $timer->start; + + time_about( sub { wait_for { $tick == 3 } }, 1, 'Reconfigured timer interval works' ); + + $timer->stop; + + $timer->configure( interval => 2 * AUT, first_interval => 0 ); + + $timer->start; + is( $tick, 3, 'Zero first_interval start not invoked yet' ); + time_about( sub { wait_for { $tick == 4 } }, 0, 'Zero first_interval invokes callback async' ); + + time_about( sub { wait_for { $tick == 5 } }, 2, 'Normal interval used after first invocation' ); + + ok( exception { $timer->configure( interval => 5 ); }, + 'Configure a running timer fails' ); + + $loop->remove( $timer ); + + undef @targs; + + is_oneref( $timer, 'Timer has refcount 1 finally' ); +} + +# reschedule => "skip" +{ + my $tick = 0; + + my $timer = IO::Async::Timer::Periodic->new( + interval => 1 * AUT, + reschedule => "skip", + + on_tick => sub { $tick++ }, + ); + + $loop->add( $timer ); + $timer->start; + + time_about( sub { wait_for { $tick == 1 } }, 1, 'skip Timer works' ); + + ok( $timer->is_running, 'skip Timer is still running' ); + + time_about( sub { wait_for { $tick == 2 } }, 1, 'skip Timer ticks a second time' ); + + $loop->remove( $timer ); +} + +# reschedule => "drift" +{ + my $tick = 0; + + my $timer = IO::Async::Timer::Periodic->new( + interval => 1 * AUT, + reschedule => "drift", + + on_tick => sub { $tick++ }, + ); + + $loop->add( $timer ); + $timer->start; + + time_about( sub { wait_for { $tick == 1 } }, 1, 'drift Timer works' ); + + ok( $timer->is_running, 'drift Timer is still running' ); + + time_about( sub { wait_for { $tick == 2 } }, 1, 'drift Timer ticks a second time' ); + + $loop->remove( $timer ); +} + +# Self-stopping +{ + my $count = 0; + my $timer = IO::Async::Timer::Periodic->new( + interval => 0.1 * AUT, + + on_tick => sub { $count++; shift->stop if $count >= 5 }, + ); + + $loop->add( $timer ); + $timer->start; + + my $timedout; + my $id = $loop->watch_time( after => 1 * AUT, code => sub { $timedout++ } ); + + wait_for { $timedout }; + + is( $count, 5, 'Self-stopping timer can stop itself' ); + + $loop->remove( $timer ); + $loop->unwatch_time( $id ); +} + +# Exception in on_tick shouldn't prevent reschedule +{ + my $count = 0; + my $timer = IO::Async::Timer::Periodic->new( + interval => 0.1 * AUT, + + on_tick => sub { $count++; die "FAIL $count" }, + ); + + $loop->add( $timer ); + $timer->start; + + like( exception { wait_for { $count > 0 } }, + qr/FAIL 1/, 'on_tick death throws exception' ); + + like( exception { wait_for { $count > 1 } }, + qr/FAIL 2/, 'on_tick death rescheduled and runs a second time' ); + + $loop->remove( $timer ); +} + +## Subclass + +my $sub_tick = 0; + +{ + my $timer = TestTimer->new( + interval => 2 * AUT, + ); + + ok( defined $timer, 'subclass $timer defined' ); + isa_ok( $timer, "IO::Async::Timer", 'subclass $timer isa IO::Async::Timer' ); + + is_oneref( $timer, 'subclass $timer has refcount 1 initially' ); + + $loop->add( $timer ); + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 after adding to Loop' ); + + $timer->start; + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 after starting' ); + + ok( $timer->is_running, 'Started subclass Timer is running' ); + + time_about( sub { wait_for { $sub_tick == 1 } }, 2, 'subclass Timer works' ); + + is_refcount( $timer, 2, 'subclass $timer has refcount 2 before removing from Loop' ); + + $loop->remove( $timer ); + + is_oneref( $timer, 'subclass $timer has refcount 1 after removing from Loop' ); +} + +done_testing; + +package TestTimer; +use base qw( IO::Async::Timer::Periodic ); + +sub on_tick { $sub_tick++ } diff --git a/t/23signal.t b/t/23signal.t new file mode 100644 index 0000000..fc28642 --- /dev/null +++ b/t/23signal.t @@ -0,0 +1,148 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use POSIX qw( SIGTERM ); + +use IO::Async::Signal; + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "This OS does not have signals" unless IO::Async::OS->HAVE_SIGNALS; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my $caught = 0; + +my @rargs; + +my $signal = IO::Async::Signal->new( + name => 'TERM', + on_receipt => sub { @rargs = @_; $caught++ }, +); + +ok( defined $signal, '$signal defined' ); +isa_ok( $signal, "IO::Async::Signal", '$signal isa IO::Async::Signal' ); + +is_oneref( $signal, '$signal has refcount 1 initially' ); + +is( $signal->notifier_name, "TERM", '$signal->notifier_name' ); + +$loop->add( $signal ); + +is_refcount( $signal, 2, '$signal has refcount 2 after adding to Loop' ); + +$loop->loop_once( 0.1 ); # nothing happens + +is( $caught, 0, '$caught idling' ); + +kill SIGTERM, $$; + +wait_for { $caught }; + +is( $caught, 1, '$caught after raise' ); +is_deeply( \@rargs, [ $signal ], 'on_receipt args after raise' ); + +my $caught2 = 0; + +my $signal2 = IO::Async::Signal->new( + name => 'TERM', + on_receipt => sub { $caught2++ }, +); + +$loop->add( $signal2 ); + +undef $caught; + +kill SIGTERM, $$; + +wait_for { $caught }; + +is( $caught, 1, '$caught after raise' ); +is( $caught2, 1, '$caught2 after raise' ); + +$loop->remove( $signal2 ); + +undef $caught; undef $caught2; + +kill SIGTERM, $$; + +wait_for { $caught }; + +is( $caught, 1, '$caught after raise' ); +is( $caught2, undef, '$caught2 after raise' ); + +undef $caught; +my $new_caught; +$signal->configure( on_receipt => sub { $new_caught++ } ); + +kill SIGTERM, $$; + +wait_for { $new_caught }; + +is( $caught, undef, '$caught after raise after replace on_receipt' ); +is( $new_caught, 1, '$new_caught after raise after replace on_receipt' ); + +undef @rargs; + +is_refcount( $signal, 2, '$signal has refcount 2 before removing from Loop' ); + +$loop->remove( $signal ); + +is_oneref( $signal, '$signal has refcount 1 finally' ); + +undef $signal; + +## Subclass + +my $sub_caught = 0; + +$signal = TestSignal->new( + name => 'TERM', +); + +ok( defined $signal, 'subclass $signal defined' ); +isa_ok( $signal, "IO::Async::Signal", 'subclass $signal isa IO::Async::Signal' ); + +is_oneref( $signal, 'subclass $signal has refcount 1 initially' ); + +$loop->add( $signal ); + +is_refcount( $signal, 2, 'subclass $signal has refcount 2 after adding to Loop' ); + +$loop->loop_once( 0.1 ); # nothing happens + +is( $sub_caught, 0, '$sub_caught idling' ); + +kill SIGTERM, $$; + +wait_for { $sub_caught }; + +is( $sub_caught, 1, '$sub_caught after raise' ); + +ok( exception { + my $signal = IO::Async::Signal->new( + name => 'this signal name does not exist', + on_receipt => sub {}, + ); + $loop->add( $signal ); + }, + 'Bad signal name fails' +); + +done_testing; + +package TestSignal; +use base qw( IO::Async::Signal ); + +sub on_receipt { $sub_caught++ } diff --git a/t/24listener.t b/t/24listener.t new file mode 100644 index 0000000..5a296aa --- /dev/null +++ b/t/24listener.t @@ -0,0 +1,301 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Identity; +use Test::Refcount; + +use IO::Async::Loop; + +use IO::Socket::INET; + +use IO::Async::Listener; + +# Some odd locations like BSD jails might not like INADDR_ANY. We'll establish +# a baseline first to test against +my $INADDR_ANY = do { + my $anysock = IO::Socket::INET->new( LocalPort => 0, Listen => 1 ); + $anysock->sockaddr; +}; +my $INADDR_ANY_HOST = inet_ntoa( $INADDR_ANY ); +if( $INADDR_ANY ne INADDR_ANY ) { + diag( "Testing with INADDR_ANY=$INADDR_ANY_HOST; this may be because of odd networking" ); +} + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my $listensock; + +$listensock = IO::Socket::INET->new( + LocalAddr => "localhost", + Type => SOCK_STREAM, + Listen => 1, +) or die "Cannot socket() - $!"; + + +{ + my $newclient; + my $listener = IO::Async::Listener->new( + handle => $listensock, + on_accept => sub { ( undef, $newclient ) = @_ }, + ); + + ok( defined $listener, 'defined $listener' ); + isa_ok( $listener, "IO::Async::Listener", '$listener isa IO::Async::Listener' ); + isa_ok( $listener, "IO::Async::Notifier", '$listener isa IO::Async::Notifier' ); + + is_oneref( $listener, '$listener has refcount 1 initially' ); + + ok( $listener->is_listening, '$listener is_listening' ); + is_deeply( [ unpack_sockaddr_in $listener->sockname ], + [ unpack_sockaddr_in $listensock->sockname ], '$listener->sockname' ); + + is( $listener->family, AF_INET, '$listener->family' ); + is( $listener->socktype, SOCK_STREAM, '$listener->sockname' ); + + $loop->add( $listener ); + + is_refcount( $listener, 2, '$listener has refcount 2 after adding to Loop' ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + ok( defined $clientsock->peername, '$clientsock is connected' ); + + wait_for { defined $newclient }; + + is_deeply( [ unpack_sockaddr_in $newclient->peername ], + [ unpack_sockaddr_in $clientsock->sockname ], '$newclient peer is correct' ); + + is_refcount( $listener, 2, '$listener has refcount 2 before removing from Loop' ); + + $loop->remove( $listener ); + + is_oneref( $listener, '$listener has refcount 1 after removing from Loop' ); +} + +# on_accept handle constructors +{ + my $accepted; + my $listener = IO::Async::Listener->new( + handle => $listensock, + on_accept => sub { ( undef, $accepted ) = @_ }, + ); + + $loop->add( $listener ); + + require IO::Async::Stream; + + # handle_constructor + { + $listener->configure( handle_constructor => sub { + return IO::Async::Stream->new; + } ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + wait_for { defined $accepted }; + + isa_ok( $accepted, "IO::Async::Stream", '$accepted with handle_constructor' ); + undef $accepted; + } + + # handle_class + { + $listener->configure( handle_class => "IO::Async::Stream" ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + wait_for { defined $accepted }; + + isa_ok( $accepted, "IO::Async::Stream", '$accepted with handle_constructor' ); + undef $accepted; + } + + $loop->remove( $listener ); +} + +# on_stream +{ + my $newstream; + my $listener = IO::Async::Listener->new( + handle => $listensock, + on_stream => sub { ( undef, $newstream ) = @_ }, + ); + + $loop->add( $listener ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + wait_for { defined $newstream }; + + isa_ok( $newstream, "IO::Async::Stream", 'on_stream $newstream isa IO::Async::Stream' ); + is_deeply( [ unpack_sockaddr_in $newstream->read_handle->peername ], + [ unpack_sockaddr_in $clientsock->sockname ], '$newstream sock peer is correct' ); + + $loop->remove( $listener ); +} + +# on_socket +{ + my $newsocket; + my $listener = IO::Async::Listener->new( + handle => $listensock, + on_socket => sub { ( undef, $newsocket ) = @_ }, + ); + + $loop->add( $listener ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + wait_for { defined $newsocket }; + + isa_ok( $newsocket, "IO::Async::Socket", 'on_socket $newsocket isa IO::Async::Socket' ); + is_deeply( [ unpack_sockaddr_in $newsocket->read_handle->peername ], + [ unpack_sockaddr_in $clientsock->sockname ], '$newsocket sock peer is correct' ); + + $loop->remove( $listener ); +} + +# Subclass +{ + my $sub_newclient; + { + package TestListener; + use base qw( IO::Async::Listener ); + + sub on_accept { ( undef, $sub_newclient ) = @_ } + } + + my $listener = TestListener->new( + handle => $listensock, + ); + + ok( defined $listener, 'subclass defined $listener' ); + isa_ok( $listener, "IO::Async::Listener", 'subclass $listener isa IO::Async::Listener' ); + + is_oneref( $listener, 'subclass $listener has refcount 1 initially' ); + + $loop->add( $listener ); + + is_refcount( $listener, 2, 'subclass $listener has refcount 2 after adding to Loop' ); + + my $clientsock = IO::Socket::INET->new( LocalAddr => "127.0.0.1", Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + ok( defined $clientsock->peername, 'subclass $clientsock is connected' ); + + wait_for { defined $sub_newclient }; + + is_deeply( [ unpack_sockaddr_in $sub_newclient->peername ], + [ unpack_sockaddr_in $clientsock->sockname ], '$sub_newclient peer is correct' ); + + is_refcount( $listener, 2, 'subclass $listener has refcount 2 before removing from Loop' ); + + $loop->remove( $listener ); + + is_oneref( $listener, 'subclass $listener has refcount 1 after removing from Loop' ); +} + +# Subclass with handle_constructor +{ + { + package TestListener::WithConstructor; + use base qw( IO::Async::Listener ); + + sub handle_constructor { return IO::Async::Stream->new } + } + + my $accepted; + + my $listener = TestListener::WithConstructor->new( + handle => $listensock, + on_accept => sub { ( undef, $accepted ) = @_; }, + ); + + $loop->add( $listener ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + wait_for { defined $accepted }; + + isa_ok( $accepted, "IO::Async::Stream", '$accepted with handle_constructor method' ); + + $loop->remove( $listener ); +} + +{ + my $newclient; + my $listener = IO::Async::Listener->new( + on_accept => sub { ( undef, $newclient ) = @_ }, + ); + + ok( !$listener->is_listening, '$listener is_listening not yet' ); + + $loop->add( $listener ); + + my $listen_self; + + $listener->listen( + addr => { family => "inet", socktype => "stream", addr => pack_sockaddr_in( 0, $INADDR_ANY ) }, + on_listen => sub { $listen_self = shift }, + on_listen_error => sub { die "Test died early - $_[0] - $_[-1]\n"; }, + ); + + ok( $listener->is_listening, '$listener is_listening' ); + + my $sockname = $listener->sockname; + ok( defined $sockname, 'defined $sockname' ); + + my ( $port, $sinaddr ) = unpack_sockaddr_in( $sockname ); + + ok( $port > 0, 'socket listens on some defined port number' ); + is( inet_ntoa( $sinaddr ), $INADDR_ANY_HOST, 'socket listens on INADDR_ANY' ); + + is( $listener->family, AF_INET, '$listener->family' ); + is( $listener->socktype, SOCK_STREAM, '$listener->sockname' ); + + is( $listen_self, $listener, '$listen_self is $listener' ); + undef $listen_self; # for refcount + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( pack_sockaddr_in( $port, INADDR_LOOPBACK ) ) or die "Cannot connect() - $!"; + + ok( defined $clientsock->peername, '$clientsock is connected' ); + + wait_for { defined $newclient }; + + is_deeply( [ unpack_sockaddr_in $newclient->peername ], + [ unpack_sockaddr_in $clientsock->sockname ], '$newclient peer is correct' ); + + $loop->remove( $listener ); +} + +done_testing; diff --git a/t/25socket.t b/t/25socket.t new file mode 100644 index 0000000..8da852f --- /dev/null +++ b/t/25socket.t @@ -0,0 +1,325 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use Errno qw( EAGAIN EWOULDBLOCK ECONNRESET ); + +use Socket qw( unpack_sockaddr_in ); + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Socket; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# useful test function +sub recv_data +{ + my ( $s ) = @_; + + my $buffer; + my $ret = $s->recv( $buffer, 8192 ); + + return $buffer if defined $ret and length $buffer; + die "Socket closed" if defined $ret; + return "" if $! == EAGAIN or $! == EWOULDBLOCK; + die "Cannot recv - $!"; +} + +ok( !exception { IO::Async::Socket->new( write_handle => \*STDOUT ) }, 'Send-only Socket works' ); + +# Receiving +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair( "inet", "dgram" ) or die "Cannot socketpair - $!"; + my @S2addr = unpack_sockaddr_in $S2->sockname; + + # Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + my @received; + + my $socket = IO::Async::Socket->new( + handle => $S1, + on_recv => sub { + my $self = shift; + my ( $dgram, $sender ) = @_; + + push @received, [ $dgram, unpack_sockaddr_in $sender ]; + }, + ); + + ok( defined $socket, 'recving $socket defined' ); + isa_ok( $socket, "IO::Async::Socket", 'recving $socket isa IO::Async::Socket' ); + + is_oneref( $socket, 'recving $socket has refcount 1 initially' ); + + $loop->add( $socket ); + + is_refcount( $socket, 2, 'recving $socket has refcount 2 after adding to Loop' ); + + $S2->send( "message\n" ); + + is_deeply( \@received, [], '@received before wait' ); + + wait_for { scalar @received }; + + is_deeply( \@received, + [ [ "message\n", @S2addr ] ], + '@received after wait' ); + + undef @received; + my @new_received; + $socket->configure( + on_recv => sub { + my $self = shift; + my ( $dgram, $sender ) = @_; + push @new_received, [ $dgram, unpack_sockaddr_in $sender ]; + }, + ); + + $S2->send( "another message\n" ); + + wait_for { scalar @new_received }; + + is( scalar @received, 0, '@received still empty after on_recv replace' ); + is_deeply( \@new_received, + [ [ "another message\n", @S2addr ] ], + '@new_received after on_recv replace' ); + + is_refcount( $socket, 2, 'receiving $socket has refcount 2 before removing from Loop' ); + + $loop->remove( $socket ); + + is_oneref( $socket, 'receiving $socket refcount 1 finally' ); +} + +SKIP: { + # Don't bother with an OS constant for this as it's only used by this unit-test + skip "This OS cannot safely ->recv with truncation", 3 if $^O eq "MSWin32"; + + my ( $S1, $S2 ) = IO::Async::OS->socketpair( "inet", "dgram" ) or die "Cannot socketpair - $!"; + + # Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + my @frags; + my $socket = IO::Async::Socket->new( + handle => $S1, + recv_len => 4, + on_recv => sub { + my ( $self, $dgram ) = @_; + push @frags, $dgram; + }, + ); + + $loop->add( $socket ); + + $S2->send( "A nice long message" ); + $S2->send( "another one here" ); + $S2->send( "and again" ); + + wait_for { scalar @frags }; + + is_deeply( \@frags, [ "A ni" ], '@frags with recv_len=4 without recv_all' ); + + wait_for { @frags == 3 }; + + is_deeply( \@frags, [ "A ni", "anot", "and " ], '@frags finally with recv_len=4 without recv_all' ); + + undef @frags; + $socket->configure( recv_all => 1 ); + + $S2->send( "Long messages" ); + $S2->send( "Repeated" ); + $S2->send( "Once more" ); + + wait_for { scalar @frags }; + + is_deeply( \@frags, [ "Long", "Repe", "Once" ], '@frags with recv_len=4 with recv_all' ); + + $loop->remove( $socket ); +} + +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair( "inet", "dgram" ) or die "Cannot socketpair - $!"; + + my $no_on_recv_socket; + ok( !exception { $no_on_recv_socket = IO::Async::Socket->new( handle => $S1 ) }, + 'Allowed to construct a Socket without an on_recv handler' ); + ok( exception { $loop->add( $no_on_recv_socket ) }, + 'Not allowed to add an on_recv-less Socket to a Loop' ); + } + +# Subclass + +my @sub_received; + +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair( "inet", "dgram" ) or die "Cannot socketpair - $!"; + my @S2addr = unpack_sockaddr_in $S2->sockname; + + # Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + my $socket = TestSocket->new( + handle => $S1, + ); + + ok( defined $socket, 'receiving subclass $socket defined' ); + isa_ok( $socket, "IO::Async::Socket", 'receiving $socket isa IO::Async::Socket' ); + + is_oneref( $socket, 'subclass $socket has refcount 1 initially' ); + + $loop->add( $socket ); + + is_refcount( $socket, 2, 'subclass $socket has refcount 2 after adding to Loop' ); + + $S2->send( "message\n" ); + + is_deeply( \@sub_received, [], '@sub_received before wait' ); + + wait_for { scalar @sub_received }; + + is_deeply( \@sub_received, + [ [ "message\n", @S2addr ] ], + '@sub_received after wait' ); + + $loop->remove( $socket ); +} + +# Sending +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair( "inet", "dgram" ) or die "Cannot socketpair - $!"; + + # Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + my $empty; + + my $socket = IO::Async::Socket->new( + write_handle => $S1, + on_outgoing_empty => sub { $empty = 1 }, + ); + + ok( defined $socket, 'sending $socket defined' ); + isa_ok( $socket, "IO::Async::Socket", 'sending $socket isa IO::Async::Socket' ); + + is_oneref( $socket, 'sending $socket has refcount 1 intially' ); + + $loop->add( $socket ); + + is_refcount( $socket, 2, 'sending $socket has refcount 2 after adding to Loop' ); + + ok( !$socket->want_writeready, 'want_writeready before send' ); + $socket->send( "message\n" ); + + ok( $socket->want_writeready, 'want_writeready after send' ); + + wait_for { $empty }; + + ok( !$socket->want_writeready, 'want_writeready after wait' ); + is( $empty, 1, '$empty after writing buffer' ); + + is( recv_data( $S2 ), "message\n", 'data after writing buffer' ); + + $socket->configure( autoflush => 1 ); + $socket->send( "immediate\n" ); + + ok( !$socket->want_writeready, 'not want_writeready after autoflush send' ); + is( recv_data( $S2 ), "immediate\n", 'data after autoflush send' ); + + $socket->configure( autoflush => 0 ); + $socket->send( "First\n" ); + $socket->configure( autoflush => 1 ); + $socket->send( "Second\n" ); + + ok( !$socket->want_writeready, 'not want_writeready after split autoflush send' ); + is( recv_data( $S2 ), "First\n", 'data[0] after split autoflush send' ); + is( recv_data( $S2 ), "Second\n", 'data[1] after split autoflush send' ); + + is_refcount( $socket, 2, 'sending $socket has refcount 2 before removing from Loop' ); + + $loop->remove( $socket ); + + is_oneref( $socket, 'sending $socket has refcount 1 finally' ); +} + +# Socket errors +{ + my ( $ES1, $ES2 ) = IO::Async::OS->socketpair or die "Cannot socketpair - $!"; + $ES2->syswrite( "X" ); # ensuring $ES1 is read- and write-ready + # cheating and hackery + bless $ES1, "ErrorSocket"; + + $ErrorSocket::errno = ECONNRESET; + + my $recv_errno; + my $send_errno; + + my $socket = IO::Async::Socket->new( + read_handle => $ES1, + on_recv => sub {}, + on_recv_error => sub { ( undef, $recv_errno ) = @_ }, + ); + + $loop->add( $socket ); + + wait_for { defined $recv_errno }; + + cmp_ok( $recv_errno, "==", ECONNRESET, 'errno after failed recv' ); + + $loop->remove( $socket ); + + $socket = IO::Async::Socket->new( + write_handle => $ES1, + on_send_error => sub { ( undef, $send_errno ) = @_ }, + ); + + $loop->add( $socket ); + + $socket->send( "hello" ); + + wait_for { defined $send_errno }; + + cmp_ok( $send_errno, "==", ECONNRESET, 'errno after failed send' ); + + $loop->remove( $socket ); +} + +done_testing; + +package TestSocket; +use base qw( IO::Async::Socket ); +use Socket qw( unpack_sockaddr_in ); + +sub on_recv +{ + my $self = shift; + my ( $dgram, $sender ) = @_; + + push @sub_received, [ $dgram, unpack_sockaddr_in $sender ]; +} + +package ErrorSocket; + +use base qw( IO::Socket ); +our $errno; + +sub recv { $! = $errno; undef; } +sub send { $! = $errno; undef; } +sub close { } diff --git a/t/26pid.t b/t/26pid.t new file mode 100644 index 0000000..cdd770b --- /dev/null +++ b/t/26pid.t @@ -0,0 +1,89 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Refcount; + +use POSIX qw( SIGTERM ); + +use IO::Async::PID; + +use IO::Async::Loop; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my $kid = fork; + defined $kid or die "Cannot fork() - $!"; + + if( $kid == 0 ) { + # child + exit( 3 ); + # this exists as a zombie for now, but we'll deal with this later + } + + my $exitcode; + my $pid = IO::Async::PID->new( + pid => $kid, + on_exit => sub { ( undef, $exitcode ) = @_; } + ); + + ok( defined $pid, '$pid defined' ); + isa_ok( $pid, "IO::Async::PID", '$pid isa IO::Async::PID' ); + + is_oneref( $pid, '$pid has refcount 1 initially' ); + + is( $pid->pid, $kid, '$pid->pid' ); + + is( $pid->notifier_name, "$kid", '$pid->notifier_name' ); + + $loop->add( $pid ); + + is_refcount( $pid, 2, '$pid has refcount 2 after adding to Loop' ); + + # reap zombie + wait_for { defined $exitcode }; + + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after process exit' ); + is( ($exitcode >> 8), 3, 'WEXITSTATUS($exitcode) after process exit' ); +} + +SKIP: { + skip "This OS has no signals", 1 unless IO::Async::OS->HAVE_SIGNALS; + + # We require that SIGTERM perform its default action; i.e. terminate the + # process. Ensure this definitely happens, in case the test harness has it + # ignored or handled elsewhere. + local $SIG{TERM} = "DEFAULT"; + + my $kid = fork; + defined $kid or die "Cannot fork() - $!"; + + if( $kid == 0 ) { + sleep( 10 ); + # Just in case the parent died already and didn't kill us + exit( 0 ); + } + + my $exitcode; + my $pid = IO::Async::PID->new( + pid => $kid, + on_exit => sub { ( undef, $exitcode ) = @_; } + ); + + $loop->add( $pid ); + + $pid->kill( SIGTERM ); + + wait_for { defined $exitcode }; + + is( ($exitcode & 0x7f), SIGTERM, 'WTERMSIG($exitcode) after SIGTERM' ); +} + +done_testing; diff --git a/t/27file.t b/t/27file.t new file mode 100644 index 0000000..6c79ff7 --- /dev/null +++ b/t/27file.t @@ -0,0 +1,113 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Refcount; + +use Fcntl qw( SEEK_SET SEEK_END ); +use File::Temp qw( tempfile ); + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::File; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +sub mkhandles +{ + my ( $rd, $filename ) = tempfile( "tmpfile.XXXXXX", UNLINK => 1 ); + open my $wr, ">", $filename or die "Cannot reopen file for writing - $!"; + + $wr->autoflush( 1 ); + + return ( $rd, $wr, $filename ); +} + +{ + my ( $rd, $wr ) = mkhandles; + + my $size_change; + my ( $new_size, $old_size ); + my ( $new_stat, $old_stat ); + my $file = IO::Async::File->new( + interval => 0.1 * AUT, + handle => $rd, + on_size_changed => sub { + ( undef, $new_size, $old_size ) = @_; + $size_change++; + }, + on_stat_changed => sub { + ( undef, $new_stat, $old_stat ) = @_; + }, + ); + + ok( defined $file, '$file defined' ); + isa_ok( $file, "IO::Async::File", '$file isa IO::Async::File' ); + + is_oneref( $file, '$file has refcount 1 initially' ); + + is( $file->handle, $rd, '$file->handle is $rd' ); + + $loop->add( $file ); + + is_refcount( $file, 2, '$file has refcount 2 after adding to Loop' ); + + $wr->syswrite( "message\n" ); + + wait_for { $size_change }; + + is( $old_size, 0, '$old_size' ); + is( $new_size, 8, '$new_size' ); + + isa_ok( $old_stat, "File::stat", '$old_stat isa File::stat' ); + isa_ok( $new_stat, "File::stat", '$new_stat isa File::stat' ); + + $loop->remove( $file ); +} + +# Follow by name +SKIP: { + skip "OS is unable to rename open files", 3 unless IO::Async::OS->HAVE_RENAME_OPEN_FILES; + + my ( undef, $wr, $filename ) = mkhandles; + + my $devino_changed; + my ( $old_stat, $new_stat ); + my $file = IO::Async::File->new( + interval => 0.1 * AUT, + filename => $filename, + on_devino_changed => sub { + ( undef, $new_stat, $old_stat ) = @_; + $devino_changed++; + }, + ); + + ok( $file->handle, '$file has a ->handle' ); + + $loop->add( $file ); + + close $wr; + rename( $filename, "$filename.old" ) or die "Cannot rename $filename - $!"; + END { defined $filename and -f $filename and unlink $filename } + END { defined $filename and -f "$filename.old" and unlink "$filename.old" } + open $wr, ">", $filename or die "Cannot reopen $filename for writing - $!"; + + wait_for { $devino_changed }; + + is( $new_stat->dev, (stat $wr)[0], '$new_stat->dev for renamed file' ); + is( $new_stat->ino, (stat $wr)[1], '$new_stat->ino for renamed file' ); + + $loop->remove( $file ); +} + +done_testing; diff --git a/t/28filestream.t b/t/28filestream.t new file mode 100644 index 0000000..f51887e --- /dev/null +++ b/t/28filestream.t @@ -0,0 +1,323 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; + +use Fcntl qw( SEEK_SET SEEK_END ); +use File::Temp qw( tempfile ); + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::FileStream; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +sub mkhandles +{ + my ( $rd, $filename ) = tempfile( "tmpfile.XXXXXX", UNLINK => 1 ); + open my $wr, ">", $filename or die "Cannot reopen file for writing - $!"; + + $wr->autoflush( 1 ); + + return ( $rd, $wr, $filename ); +} + +{ + my ( $rd, $wr ) = mkhandles; + + my @lines; + my $initial_size; + + my $filestream = IO::Async::FileStream->new( + interval => 0.1 * AUT, + read_handle => $rd, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + on_initial => sub { ( undef, $initial_size ) = @_ }, + ); + + ok( defined $filestream, '$filestream defined' ); + isa_ok( $filestream, "IO::Async::FileStream", '$filestream isa IO::Async::FileStream' ); + + is_oneref( $filestream, 'reading $filestream has refcount 1 initially' ); + + $loop->add( $filestream ); + + is_refcount( $filestream, 2, '$filestream has refcount 2 after adding to Loop' ); + + is( $initial_size, 0, '$initial_size is 0' ); + + $wr->syswrite( "message\n" ); + + is_deeply( \@lines, [], '@lines before wait' ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "message\n" ], '@lines after wait' ); + + $loop->remove( $filestream ); +} + +# on_initial +{ + my ( $rd, $wr ) = mkhandles; + + $wr->syswrite( "Some initial content\n" ); + + my @lines; + my $initial_size; + + my $filestream = IO::Async::FileStream->new( + interval => 0.1 * AUT, + read_handle => $rd, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + on_initial => sub { ( undef, $initial_size ) = @_ }, + ); + + $loop->add( $filestream ); + + is( $initial_size, 21, '$initial_size is 21' ); + + $wr->syswrite( "More content\n" ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "Some initial content\n", "More content\n" ], 'All content is visible' ); + + $loop->remove( $filestream ); +} + +# seek_to_last +{ + my ( $rd, $wr ) = mkhandles; + + $wr->syswrite( "Some skipped content\nWith a partial line" ); + + my @lines; + + my $filestream = IO::Async::FileStream->new( + interval => 0.1 * AUT, + read_handle => $rd, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + return 0 unless( $$buffref =~ s/^(.*\n)// ); + + push @lines, $1; + return 1; + }, + on_initial => sub { + my $self = shift; + # Give it a tiny block size, forcing it to have to seek harder to find the \n + ok( $self->seek_to_last( "\n", blocksize => 8 ), 'FileStream successfully seeks to last \n' ); + }, + ); + + $loop->add( $filestream ); + + $wr->syswrite( " finished here\n" ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "With a partial line finished here\n" ], 'Partial line completely returned' ); + + $loop->remove( $filestream ); +} + +# on_initial can skip content +{ + my ( $rd, $wr ) = mkhandles; + + $wr->syswrite( "Some skipped content\n" ); + + my @lines; + + my $filestream = IO::Async::FileStream->new( + interval => 0.1 * AUT, + read_handle => $rd, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + return 0 unless( $$buffref =~ s/^(.*\n)// ); + + push @lines, $1; + return 1; + }, + on_initial => sub { my $self = shift; $self->seek( 0, SEEK_END ); }, + ); + + $loop->add( $filestream ); + + $wr->syswrite( "Additional content\n" ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "Additional content\n" ], 'Initial content is skipped' ); + + $loop->remove( $filestream ); +} + +# Truncation +{ + my ( $rd, $wr ) = mkhandles; + + my @lines; + my $truncated; + + my $filestream = IO::Async::FileStream->new( + interval => 0.1 * AUT, + read_handle => $rd, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + return 0 unless( $$buffref =~ s/^(.*\n)// ); + + push @lines, $1; + return 1; + }, + on_truncated => sub { $truncated++ }, + ); + + $loop->add( $filestream ); + + $wr->syswrite( "Some original lines\nin the file\n" ); + + wait_for { scalar @lines }; + + $wr->truncate( 0 ); + sysseek( $wr, 0, SEEK_SET ); + $wr->syswrite( "And another\n" ); + + wait_for { @lines == 3 }; + + is( $truncated, 1, 'File content truncation detected' ); + is_deeply( \@lines, + [ "Some original lines\n", "in the file\n", "And another\n" ], + 'All three lines read' ); + + $loop->remove( $filestream ); +} + +# Follow by name +SKIP: { + skip "OS is unable to rename open files", 7 unless IO::Async::OS->HAVE_RENAME_OPEN_FILES; + + my ( undef, $wr, $filename ) = mkhandles; + + my @lines; + + my $filestream = IO::Async::FileStream->new( + interval => 0.1 * AUT, + filename => $filename, + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + ok( defined $filestream, '$filestream defined for filenaem' ); + isa_ok( $filestream, "IO::Async::FileStream", '$filestream isa IO::Async::FileStream' ); + + is_oneref( $filestream, 'reading $filestream has refcount 1 initially' ); + + $loop->add( $filestream ); + + is_refcount( $filestream, 2, '$filestream has refcount 2 after adding to Loop' ); + + $wr->syswrite( "message\n" ); + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "message\n" ], '@lines after wait' ); + shift @lines; + + $wr->syswrite( "last line of old file\n" ); + close $wr; + rename( $filename, "$filename.old" ) or die "Cannot rename $filename - $!"; + END { defined $filename and -f $filename and unlink $filename } + END { defined $filename and -f "$filename.old" and unlink "$filename.old" } + open $wr, ">", $filename or die "Cannot reopen $filename for writing - $!"; + $wr->syswrite( "first line of new file\n" ); + + wait_for { scalar @lines }; + is_deeply( $lines[0], "last line of old file\n", '@lines sees last line of old file' ); + wait_for { scalar @lines >= 2 }; + is_deeply( $lines[1], "first line of new file\n", '@lines sees first line of new file' ); + + $loop->remove( $filestream ); +} + +# Subclass +my @sub_lines; + +{ + my ( $rd, $wr ) = mkhandles; + + my $filestream = TestStream->new( + interval => 0.1 * AUT, + read_handle => $rd, + ); + + ok( defined $filestream, 'subclass $filestream defined' ); + isa_ok( $filestream, "IO::Async::FileStream", '$filestream isa IO::Async::FileStream' ); + + is_oneref( $filestream, 'subclass $filestream has refcount 1 initially' ); + + $loop->add( $filestream ); + + is_refcount( $filestream, 2, 'subclass $filestream has refcount 2 after adding to Loop' ); + + $wr->syswrite( "message\n" ); + + is_deeply( \@sub_lines, [], '@sub_lines before wait' ); + + wait_for { scalar @sub_lines }; + + is_deeply( \@sub_lines, [ "message\n" ], '@sub_lines after wait' ); + + $loop->remove( $filestream ); +} + +done_testing; + +package TestStream; +use base qw( IO::Async::FileStream ); + +sub on_read +{ + my $self = shift; + my ( $buffref ) = @_; + + return 0 unless $$buffref =~ s/^(.*\n)//; + + push @sub_lines, $1; + return 1; +} diff --git a/t/30loop-fork.t b/t/30loop-fork.t new file mode 100644 index 0000000..927b6c9 --- /dev/null +++ b/t/30loop-fork.t @@ -0,0 +1,81 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; + +use POSIX qw( SIGINT ); + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my $exitcode; + $loop->fork( + code => sub { return 5; }, + on_exit => sub { ( undef, $exitcode ) = @_ }, + ); + + wait_for { defined $exitcode }; + + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after child exit' ); + is( ($exitcode >> 8), 5, 'WEXITSTATUS($exitcode) after child exit' ); +} + +{ + my $exitcode; + $loop->fork( + code => sub { die "error"; }, + on_exit => sub { ( undef, $exitcode ) = @_ }, + ); + + wait_for { defined $exitcode }; + + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after child die' ); + is( ($exitcode >> 8), 255, 'WEXITSTATUS($exitcode) after child die' ); +} + +SKIP: { + skip "This OS does not have signals", 1 unless IO::Async::OS->HAVE_SIGNALS; + + local $SIG{INT} = sub { exit( 22 ) }; + + my $exitcode; + $loop->fork( + code => sub { kill SIGINT, $$ }, + on_exit => sub { ( undef, $exitcode ) = @_ }, + ); + + wait_for { defined $exitcode }; + + is( ($exitcode & 0x7f), SIGINT, 'WTERMSIG($exitcode) after child SIGINT' ); +} + +SKIP: { + skip "This OS does not have signals", 2 unless IO::Async::OS->HAVE_SIGNALS; + + local $SIG{INT} = sub { exit( 22 ) }; + + my $exitcode; + $loop->fork( + code => sub { kill SIGINT, $$ }, + on_exit => sub { ( undef, $exitcode ) = @_ }, + keep_signals => 1, + ); + + wait_for { defined $exitcode }; + + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after child SIGINT with keep_signals' ); + is( ($exitcode >> 8), 22, 'WEXITSTATUS($exitcode) after child SIGINT with keep_signals' ); +} + +done_testing; diff --git a/t/31loop-spawnchild.t b/t/31loop-spawnchild.t new file mode 100644 index 0000000..1cac1d9 --- /dev/null +++ b/t/31loop-spawnchild.t @@ -0,0 +1,168 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; + +use POSIX qw( ENOENT EBADF ); + +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +use IO::Async::Loop; + +# Need to look this up, so we don't hardcode the message in the test script +# This might cause locale issues +use constant ENOENT_MESSAGE => do { local $! = ENOENT; "$!" }; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +ok( exception { $loop->spawn_child( badoption => 1 ); }, 'Bad option to spawn fails' ); + +ok( exception { $loop->spawn_child( code => sub { 1 }, command => "hello" ); }, + 'Both code and command options to spawn fails' ); + +ok( exception { $loop->spawn_child( on_exit => sub { 1 } ); }, 'Bad option to spawn fails' ); + +{ + my ( $exited_pid, $exitcode, $dollarbang, $dollarat ); + my $spawned_pid = $loop->spawn_child( + code => sub { return 42; }, + on_exit => sub { ( $exited_pid, $exitcode, $dollarbang, $dollarat ) = @_; } + ); + + wait_for { defined $exited_pid }; + + is( $exited_pid, $spawned_pid, '$exited_pid == $spawned_pid after spawn CODE' ); + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after spawn CODE' ); + is( ($exitcode >> 8), 42, 'WEXITSTATUS($exitcode) after spawn CODE' ); + # dollarbang isn't interesting here + is( $dollarat, '', '$dollarat after spawn CODE' ); +} + +my $ENDEXIT = 10; +END { exit $ENDEXIT if defined $ENDEXIT; } + +{ + my ( $exited_pid, $exitcode, $dollarbang, $dollarat ); + my $spawned_pid = $loop->spawn_child( + code => sub { return 0; }, + on_exit => sub { ( $exited_pid, $exitcode, $dollarbang, $dollarat ) = @_; } + ); + + wait_for { defined $exited_pid }; + + is( $exited_pid, $spawned_pid, '$exited_pid == $spawned_pid after spawn CODE with END' ); + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after spawn CODE with END' ); + # If this comes out as 10 then the END block ran and we fail. + is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after spawn CODE with END' ); + # dollarbang isn't interesting here + is( $dollarat, '', '$dollarat after spawn CODE with END' ); +} + +{ + my ( $exited_pid, $exitcode, $dollarbang, $dollarat ); + my $spawned_pid = $loop->spawn_child( + code => sub { die "An exception here\n"; }, + on_exit => sub { ( $exited_pid, $exitcode, $dollarbang, $dollarat ) = @_; } + ); + + wait_for { defined $exited_pid }; + + is( $exited_pid, $spawned_pid, '$exited_pid == $spawned_pid after spawn CODE with die with END' ); + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after spawn CODE with die with END' ); + is( ($exitcode >> 8), 255, 'WEXITSTATUS($exitcode) after spawn CODE with die with END' ); + # dollarbang isn't interesting here + is( $dollarat, "An exception here\n", '$dollarat after spawn CODE with die with END' ); +} + +undef $ENDEXIT; + +# We need a command that just exits immediately with 0 +my $true; +foreach (qw( /bin/true /usr/bin/true )) { + $true = $_, last if -x $_; +} + +# Didn't find a likely-looking candidate. We'll fake one using perl itself +$true = "$^X -e 1" if !defined $true; + +{ + my ( $exited_pid, $exitcode, $dollarbang, $dollarat ); + my $spawned_pid = $loop->spawn_child( + command => $true, + on_exit => sub { ( $exited_pid, $exitcode, $dollarbang, $dollarat ) = @_; } + ); + + wait_for { defined $exited_pid }; + + is( $exited_pid, $spawned_pid, '$exited_pid == $spawned_pid after spawn '.$true ); + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after spawn '.$true ); + is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after spawn '.$true ); + is( $dollarbang+0, 0, '$dollarbang after spawn '.$true ); + is( $dollarat, '', '$dollarat after spawn '.$true ); +} + +# Just be paranoid in case anyone actually has this +my $donotexist = "/bin/donotexist"; +$donotexist .= "X" while -e $donotexist; + +{ + my ( $exited_pid, $exitcode, $dollarbang, $dollarat ); + my $spawned_pid = $loop->spawn_child( + command => $donotexist, + on_exit => sub { ( $exited_pid, $exitcode, $dollarbang, $dollarat ) = @_; } + ); + + wait_for { defined $exited_pid }; + + is( $exited_pid, $spawned_pid, '$exited_pid == $spawned_pid after spawn donotexist' ); + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after spawn donotexist' ); + is( ($exitcode >> 8), 255, 'WEXITSTATUS($exitcode) after spawn donotexist' ); + is( $dollarbang+0, ENOENT, '$dollarbang numerically after spawn donotexist' ); + is( "$dollarbang", ENOENT_MESSAGE, '$dollarbang string after spawn donotexist' ); + is( $dollarat, '', '$dollarat after spawn donotexist' ); +} + +{ + my ( $exited_pid, $exitcode, $dollarbang, $dollarat ); + my $spawned_pid = $loop->spawn_child( + command => [ $^X, "-e", "exit 14" ], + on_exit => sub { ( $exited_pid, $exitcode, $dollarbang, $dollarat ) = @_; } + ); + + wait_for { defined $exited_pid }; + + is( $exited_pid, $spawned_pid, '$exited_pid == $spawned_pid after spawn ARRAY' ); + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after spawn ARRAY' ); + is( ($exitcode >> 8), 14, 'WEXITSTATUS($exitcode) after spawn ARRAY' ); + is( $dollarbang+0, 0, '$dollarbang after spawn ARRAY' ); + is( $dollarat, '', '$dollarat after spawn ARRAY' ); +} + +{ + my( $pipe_r, $pipe_w ) = IO::Async::OS->pipepair or die "Cannot pipepair - $!"; + + my ( $exited_pid, $exitcode, $dollarbang, $dollarat ); + my $spawned_pid = $loop->spawn_child( + code => sub { return $pipe_w->syswrite( "test" ); }, + on_exit => sub { ( $exited_pid, $exitcode, $dollarbang, $dollarat ) = @_; } + ); + + wait_for { defined $exited_pid }; + + is( $exited_pid, $spawned_pid, '$exited_pid == $spawned_pid after pipe close test' ); + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after pipe close test' ); + is( ($exitcode >> 8), 255, 'WEXITSTATUS($exitcode) after pipe close test' ); + is( $dollarbang+0, EBADF, '$dollarbang numerically after pipe close test' ); + is( $dollarat, '', '$dollarat after pipe close test' ); +} + +done_testing; diff --git a/t/32loop-spawnchild-setup.t b/t/32loop-spawnchild-setup.t new file mode 100644 index 0000000..7ecdf85 --- /dev/null +++ b/t/32loop-spawnchild-setup.t @@ -0,0 +1,439 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; + +use File::Temp qw( tmpnam ); +use POSIX qw( ENOENT EBADF getcwd ); + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +ok( exception { $loop->spawn_child( code => sub { 1 }, setup => "hello" ); }, + 'Bad setup type fails' ); + +ok( exception { $loop->spawn_child( code => sub { 1 }, setup => [ 'somerandomthing' => 1 ] ); }, + 'Setup with bad key fails' ); + +# These tests are all very similar looking, with slightly different start and +# code values. Easiest to wrap them up in a common testing wrapper. + +sub TEST +{ + my ( $name, %attr ) = @_; + + my $exitcode; + my $dollarbang; + my $dollarat; + + my ( undef, $callerfile, $callerline ) = caller; + + $loop->spawn_child( + code => $attr{code}, + exists $attr{setup} ? ( setup => $attr{setup} ) : (), + on_exit => sub { ( undef, $exitcode, $dollarbang, $dollarat ) = @_; }, + ); + + wait_for { defined $exitcode }; + + if( exists $attr{exitstatus} ) { + ok( ($exitcode & 0x7f) == 0, "WIFEXITED(\$exitcode) after $name" ); + is( ($exitcode >> 8), $attr{exitstatus}, "WEXITSTATUS(\$exitcode) after $name" ); + } + + if( exists $attr{dollarbang} ) { + is( $dollarbang+0, $attr{dollarbang}, "\$dollarbang numerically after $name" ); + } + + if( exists $attr{dollarat} ) { + is( $dollarat, $attr{dollarat}, "\$dollarat after $name" ); + } +} + +# A useful utility function like blocking read with a timeout +sub read_timeout +{ + my ( $fh, undef, $len, $timeout ) = @_; + + my $rvec = ''; + vec( $rvec, fileno $fh, 1 ) = 1; + + select( $rvec, undef, undef, $timeout ); + + return undef if !vec( $rvec, fileno $fh, 1 ); + + return $fh->read( $_[1], $len ); +} + +my $buffer; +my $ret; + +{ + my( $pipe_r, $pipe_w ) = IO::Async::OS->pipepair or die "Cannot pipepair - $!"; + + TEST "pipe dup to fd1", + setup => [ fd1 => [ 'dup', $pipe_w ] ], + code => sub { print "test"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after pipe dup to fd1' ); + is( $buffer, 'test', '$buffer after pipe dup to fd1' ); + + my $pipe_w_fileno = fileno $pipe_w; + + TEST "pipe dup to fd1 closes pipe", + setup => [ fd1 => [ 'dup', $pipe_w ] ], + code => sub { + my $f = IO::Handle->new_from_fd( $pipe_w_fileno, "w" ); + defined $f and return 1; + $! == EBADF or return 1; + return 0; + }, + + exitstatus => 0, + dollarat => ''; + + TEST "pipe dup to stdout shortcut", + setup => [ stdout => $pipe_w ], + code => sub { print "test"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after pipe dup to stdout shortcut' ); + is( $buffer, 'test', '$buffer after pipe dup to stdout shortcut' ); + + TEST "pipe dup to \\*STDOUT IO reference", + setup => [ \*STDOUT => $pipe_w ], + code => sub { print "test2"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 5, 0.1 ); + + is( $ret, 5, '$pipe_r->read after pipe dup to \\*STDOUT IO reference' ); + is( $buffer, 'test2', '$buffer after pipe dup to \\*STDOUT IO reference' ); + + TEST "pipe keep open", + setup => [ "fd$pipe_w_fileno" => [ 'keep' ] ], + code => sub { $pipe_w->autoflush(1); $pipe_w->print( "test" ) }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after keep pipe open' ); + is( $buffer, 'test', '$buffer after keep pipe open' ); + + TEST "pipe keep shortcut", + setup => [ "fd$pipe_w_fileno" => 'keep' ], + code => sub { $pipe_w->autoflush(1); $pipe_w->print( "test" ) }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after keep pipe open' ); + is( $buffer, 'test', '$buffer after keep pipe open' ); + + + TEST "pipe dup to stdout", + setup => [ stdout => [ 'dup', $pipe_w ] ], + code => sub { print "test"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after pipe dup to stdout' ); + is( $buffer, 'test', '$buffer after pipe dup to stdout' ); + + TEST "pipe dup to fd2", + setup => [ fd2 => [ 'dup', $pipe_w ] ], + code => sub { print STDERR "test"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after pipe dup to fd2' ); + is( $buffer, 'test', '$buffer after pipe dup to fd2' ); + + TEST "pipe dup to stderr", + setup => [ stderr => [ 'dup', $pipe_w ] ], + code => sub { print STDERR "test"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after pipe dup to stderr' ); + is( $buffer, 'test', '$buffer after pipe dup to stderr' ); + + TEST "pipe dup to other FD", + setup => [ fd4 => [ 'dup', $pipe_w ] ], + code => sub { + close STDOUT; + open( STDOUT, ">&=4" ) or die "Cannot open fd4 as stdout - $!"; + print "test"; + }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after pipe dup to other FD' ); + is( $buffer, 'test', '$buffer after pipe dup to other FD' ); + + TEST "pipe dup to its own FD", + setup => [ "fd$pipe_w_fileno" => $pipe_w ], + code => sub { + close STDOUT; + open( STDOUT, ">&=$pipe_w_fileno" ) or die "Cannot open fd$pipe_w_fileno as stdout - $!"; + print "test"; + }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after pipe dup to its own FD' ); + is( $buffer, 'test', '$buffer after pipe dup to its own FD' ); + + TEST "other FD close", + code => sub { return $pipe_w->syswrite( "test" ); }, + + exitstatus => 255, + dollarbang => EBADF, + dollarat => ''; + + # Try to force a writepipe clash by asking to dup the pipe to lots of FDs + TEST "writepipe clash", + code => sub { print "test"; }, + setup => [ map { +"fd$_" => $pipe_w } ( 1 .. 19 ) ], + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 4, 0.1 ); + + is( $ret, 4, '$pipe_r->read after writepipe clash' ); + is( $buffer, 'test', '$buffer after writepipe clash' ); + + my( $pipe2_r, $pipe2_w ) = IO::Async::OS->pipepair or die "Cannot pipepair - $!"; + $pipe2_r->blocking( 0 ); + + TEST "pipe dup to stdout and stderr", + setup => [ stdout => $pipe_w, stderr => $pipe2_w ], + code => sub { print "output"; print STDERR "error"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 6, 0.1 ); + + is( $ret, 6, '$pipe_r->read after pipe dup to stdout and stderr' ); + is( $buffer, 'output', '$buffer after pipe dup to stdout and stderr' ); + + undef $buffer; + $ret = read_timeout( $pipe2_r, $buffer, 5, 0.1 ); + + is( $ret, 5, '$pipe2_r->read after pipe dup to stdout and stderr' ); + is( $buffer, 'error', '$buffer after pipe dup to stdout and stderr' ); + + TEST "pipe dup to stdout and stderr same pipe", + setup => [ stdout => $pipe_w, stderr => $pipe_w ], + code => sub { print "output"; print STDERR "error"; }, + + exitstatus => 1, + dollarat => ''; + + undef $buffer; + $ret = read_timeout( $pipe_r, $buffer, 11, 0.1 ); + + is( $ret, 11, '$pipe_r->read after pipe dup to stdout and stderr same pipe' ); + is( $buffer, 'outputerror', '$buffer after pipe dup to stdout and stderr same pipe' ); +} + +{ + my ( $child_r, $my_w, $my_r, $child_w ) = IO::Async::OS->pipequad or die "Cannot pipequad - $!"; + + $my_w->syswrite( "hello\n" ); + + TEST "pipe quad to fd0/fd1", + setup => [ stdin => $child_r, + stdout => $child_w, ], + code => sub { print uc scalar <STDIN>; return 0 }, + + exitstatus => 0, + dollarat => ''; + + my $buffer; + $ret = read_timeout( $my_r, $buffer, 6, 0.1 ); + + is( $ret, 6, '$my_r->read after pipe quad to fd0/fd1' ); + is( $buffer, "HELLO\n", '$buffer after pipe quad to fd0/fd1' ); +} + +{ + # Try to swap two filehandles and cause a dup2() collision + my @fhA = IO::Async::OS->pipepair or die "Cannot pipepair - $!"; + my @fhB = IO::Async::OS->pipepair or die "Cannot pipepair - $!"; + + my $filenoA = $fhA[1]->fileno; + my $filenoB = $fhB[1]->fileno; + + TEST "fd swap", + setup => [ + "fd$filenoA" => $fhB[1], + "fd$filenoB" => $fhA[1], + ], + code => sub { + $fhA[1]->print( "FHA" ); $fhA[1]->autoflush(1); + $fhB[1]->print( "FHB" ); $fhB[1]->autoflush(1); + return 0; + }, + + exitstatus => 0; + + my $buffer; + + read_timeout( $fhA[0], $buffer, 3, 0.1 ); + is( $buffer, "FHB", '$buffer [A] after dup2() swap' ); + + read_timeout( $fhB[0], $buffer, 3, 0.1 ); + is( $buffer, "FHA", '$buffer [B] after dup2() swap' ); +} + +TEST "stdout close", + setup => [ stdout => [ 'close' ] ], + code => sub { print "test"; }, + + exitstatus => 255, + dollarbang => EBADF, + dollarat => ''; + +TEST "stdout close shortcut", + setup => [ stdout => 'close' ], + code => sub { print "test"; }, + + exitstatus => 255, + dollarbang => EBADF, + dollarat => ''; + +{ + my $name = tmpnam; + END { unlink $name if defined $name and -f $name } + + TEST "stdout open", + setup => [ stdout => [ 'open', '>', $name ] ], + code => sub { print "test"; }, + + exitstatus => 1, + dollarat => ''; + + ok( -f $name, 'tmpnam file exists after stdout open' ); + + open( my $tmpfh, "<", $name ) or die "Cannot open '$name' for reading - $!"; + + undef $buffer; + $ret = read_timeout( $tmpfh, $buffer, 4, 0.1 ); + + is( $ret, 4, '$tmpfh->read after stdout open' ); + is( $buffer, 'test', '$buffer after stdout open' ); + + TEST "stdout open append", + setup => [ stdout => [ 'open', '>>', $name ] ], + code => sub { print "value"; }, + + exitstatus => 1, + dollarat => ''; + + seek( $tmpfh, 0, 0 ); + + undef $buffer; + $ret = read_timeout( $tmpfh, $buffer, 9, 0.1 ); + + is( $ret, 9, '$tmpfh->read after stdout open append' ); + is( $buffer, 'testvalue', '$buffer after stdout open append' ); +} + +$ENV{TESTKEY} = "parent value"; + +TEST "environment is preserved", + setup => [], + code => sub { return $ENV{TESTKEY} eq "parent value" ? 0 : 1 }, + + exitstatus => 0, + dollarat => ''; + +TEST "environment is overwritten", + setup => [ env => { TESTKEY => "child value" } ], + code => sub { return $ENV{TESTKEY} eq "child value" ? 0 : 1 }, + + exitstatus => 0, + dollarat => ''; + +SKIP: { + # Some of the CPAN smoke testers might run test scripts under modified nice + # anyway. We'd better get our starting value to check for difference, not + # absolute + my $prio_now = getpriority(0,0); + + # If it's already quite high, we don't want to hit the limit and be + # clamped. Just skip the tests if it's too high before we start. + skip "getpriority is already above 15, so I won't try renicing upwards", 3 if $prio_now > 15; + + TEST "nice works", + setup => [ nice => 3 ], + code => sub { return getpriority(0,0) == $prio_now + 3 ? 0 : 1 }, + + exitstatus => 0, + dollarat => ''; +} + +TEST "chdir works", + setup => [ chdir => "/" ], + code => sub { return getcwd eq "/" ? 0 : 1 }, + + exitstatus => 0, + dollarat => ''; + +done_testing; diff --git a/t/33process.t b/t/33process.t new file mode 100644 index 0000000..72c07b1 --- /dev/null +++ b/t/33process.t @@ -0,0 +1,245 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Refcount; + +use POSIX qw( ENOENT SIGTERM SIGUSR1 ); +use constant ENOENT_MESSAGE => do { local $! = ENOENT; "$!" }; + +use IO::Async::Process; + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my ( $invocant, $exitcode ); + + my $process = IO::Async::Process->new( + code => sub { return 0 }, + on_finish => sub { ( $invocant, $exitcode ) = @_; }, + ); + + is_oneref( $process, '$process has refcount 1 before $loop->add' ); + + is( $process->notifier_name, "nopid", '$process->notifier_name before $loop->add' ); + + ok( !$process->is_running, '$process is not yet running' ); + ok( !defined $process->pid, '$process has no PID yet' ); + + $loop->add( $process ); + + is_refcount( $process, 2, '$process has refcount 2 after $loop->add' ); + + my $pid = $process->pid; + + ok( $process->is_running, '$process is running' ); + ok( defined $pid, '$process now has a PID' ); + + is( $process->notifier_name, "$pid", '$process->notifier_name after $loop->add' ); + + wait_for { defined $exitcode }; + + is( $invocant, $process, '$_[0] in on_finish is $process' ); + undef $invocant; # refcount + + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after sub { 0 }' ); + is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after sub { 0 }' ); + + ok( !$process->is_running, '$process no longer running' ); + ok( defined $process->pid, '$process still has PID after exit' ); + + is( $process->notifier_name, "[$pid]", '$process->notifier_name after exit' ); + + ok( $process->is_exited, '$process->is_exited after sub { 0 }' ); + is( $process->exitstatus, 0, '$process->exitstatus after sub { 0 }' ); + + ok( !defined $process->loop, '$process no longer in Loop' ); + + is_oneref( $process, '$process has refcount 1 before EOS' ); +} + +{ + my $process = IO::Async::Process->new( + code => sub { return 3 }, + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after sub { 3 }' ); + is( $process->exitstatus, 3, '$process->exitstatus after sub { 3 }' ); +} + +{ + my ( $invocant, $exception, $exitcode ); + + my $process = IO::Async::Process->new( + code => sub { die "An exception\n" }, + on_finish => sub { die "Test failed early\n" }, + on_exception => sub { ( $invocant, $exception, undef, $exitcode ) = @_ }, + ); + + is_oneref( $process, '$process has refcount 1 before $loop->add' ); + + $loop->add( $process ); + + is_refcount( $process, 2, '$process has refcount 2 after $loop->add' ); + + wait_for { defined $exitcode }; + + is( $invocant, $process, '$_[0] in on_exception is $process' ); + undef $invocant; # refcount + + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after sub { die }' ); + is( ($exitcode >> 8), 255, 'WEXITSTATUS($exitcode) after sub { die }' ); + is( $exception, "An exception\n", '$exception after sub { die }' ); + + ok( $process->is_exited, '$process->is_exited after sub { die }' ); + is( $process->exitstatus, 255, '$process->exitstatus after sub { die }' ); + is( $process->exception, "An exception\n", '$process->exception after sub { die }' ); + + is_oneref( $process, '$process has refcount 1 before EOS' ); +} + +{ + my $exitcode; + + my $process = IO::Async::Process->new( + code => sub { die "An exception\n" }, + on_finish => sub { ( undef, $exitcode ) = @_ }, + ); + + $loop->add( $process ); + + wait_for { defined $exitcode }; + + ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after sub { die } on_finish' ); + is( ($exitcode >> 8), 255, 'WEXITSTATUS($exitcode) after sub { die } on_finish' ); + + ok( $process->is_exited, '$process->is_exited after sub { die } on_finish' ); + is( $process->exitstatus, 255, '$process->exitstatus after sub { die } on_finish' ); + is( $process->exception, "An exception\n", '$process->exception after sub { die } on_finish' ); +} + +{ + my $process = IO::Async::Process->new( + command => [ $^X, "-e", '1' ], + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl -e 1' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl -e 1' ); +} + +{ + my $process = IO::Async::Process->new( + command => [ $^X, "-e", 'exit 5' ], + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl -e exit 5' ); + is( $process->exitstatus, 5, '$process->exitstatus after perl -e exit 5' ); +} + +{ + # Just be paranoid in case anyone actually has this + my $donotexist = "/bin/donotexist"; + $donotexist .= "X" while -e $donotexist; + + my ( $exception, $errno ); + + my $process = IO::Async::Process->new( + command => $donotexist, + on_finish => sub { die "Test failed early\n" }, + on_exception => sub { ( undef, $exception, $errno ) = @_ }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + is( $errno+0, ENOENT, '$errno number after donotexist' ); + is( "$errno", ENOENT_MESSAGE, '$errno string after donotexist' ); + + ok( $process->is_exited, '$process->is_exited after donotexist' ); + is( $process->exitstatus, 255, '$process->exitstatus after donotexist' ); + is( $process->errno, ENOENT, '$process->errno number after donotexist' ); + is( $process->errstr, ENOENT_MESSAGE, '$process->errno string after donotexist' ); + is( $process->exception, "", '$process->exception after donotexist' ); +} + +{ + $ENV{TEST_KEY} = "foo"; + + my $process = IO::Async::Process->new( + code => sub { $ENV{TEST_KEY} eq "bar" ? 0 : 1 }, + setup => [ + env => { TEST_KEY => "bar" }, + ], + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after %ENV test' ); + is( $process->exitstatus, 0, '$process->exitstatus after %ENV test' ); +} + +SKIP: { + skip "This OS does not have signals", 2 unless IO::Async::OS->HAVE_SIGNALS; + + my $child_ready; + $loop->watch_signal( USR1 => sub { $child_ready++ } ); + + my $parentpid = $$; + my $process = IO::Async::Process->new( + code => sub { + my $exitcode = 10; + eval { + local $SIG{TERM} = sub { $exitcode = 20; die }; + kill SIGUSR1 => $parentpid; + sleep 60; # block on signal + }; + return $exitcode; + }, + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { $child_ready }; + + $process->kill( SIGTERM ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after ->kill' ); + is( $process->exitstatus, 20, '$process->exitstatus after ->kill' ); + + $loop->unwatch_signal( USR1 => ); +} + +done_testing; diff --git a/t/34process-handles.t b/t/34process-handles.t new file mode 100644 index 0000000..f199ef7 --- /dev/null +++ b/t/34process-handles.t @@ -0,0 +1,429 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; + +use IO::Async::Process; + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +use Socket qw( PF_INET sockaddr_family ); + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my $process = IO::Async::Process->new( + code => sub { print "hello\n"; return 0 }, + stdout => { via => "pipe_read" }, + on_finish => sub { }, + ); + + isa_ok( $process->stdout, "IO::Async::Stream", '$process->stdout' ); + + is( $process->stdout->notifier_name, "stdout", '$process->stdout->notifier_name' ); + + my @stdout_lines; + + $process->stdout->configure( + on_read => sub { + my ( undef, $buffref ) = @_; + push @stdout_lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + $loop->add( $process ); + + ok( defined $process->stdout->read_handle, '$process->stdout has read_handle for sub { print }' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after sub { print }' ); + is( $process->exitstatus, 0, '$process->exitstatus after sub { print }' ); + + is_deeply( \@stdout_lines, [ "hello\n" ], '@stdout_lines after sub { print }' ); +} + +{ + my @stdout_lines; + + my $process = IO::Async::Process->new( + code => sub { print "hello\n"; return 0 }, + stdout => { + on_read => sub { + my ( undef, $buffref ) = @_; + push @stdout_lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + }, + on_finish => sub { }, + ); + + isa_ok( $process->stdout, "IO::Async::Stream", '$process->stdout' ); + + $loop->add( $process ); + + ok( defined $process->stdout->read_handle, '$process->stdout has read_handle for sub { print } inline' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after sub { print } inline' ); + is( $process->exitstatus, 0, '$process->exitstatus after sub { print } inline' ); + + is_deeply( \@stdout_lines, [ "hello\n" ], '@stdout_lines after sub { print } inline' ); +} + +{ + my $stdout; + + my $process = IO::Async::Process->new( + code => sub { print "hello\n"; return 0 }, + stdout => { into => \$stdout }, + on_finish => sub { }, + ); + + isa_ok( $process->stdout, "IO::Async::Stream", '$process->stdout' ); + + $loop->add( $process ); + + ok( defined $process->stdout->read_handle, '$process->stdout has read_handle for sub { print } into' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after sub { print } into' ); + is( $process->exitstatus, 0, '$process->exitstatus after sub { print } into' ); + + is( $stdout, "hello\n", '$stdout after sub { print } into' ) +} + +{ + my $stdout; + + my $process = IO::Async::Process->new( + command => [ $^X, "-e", 'print "hello\n"' ], + stdout => { into => \$stdout }, + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDOUT' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDOUT' ); + + is( $stdout, "hello\n", '$stdout after perl STDOUT' ); +} + +{ + my $stdout; + my $stderr; + + my $process = IO::Async::Process->new( + command => [ $^X, "-e", 'print STDOUT "output\n"; print STDERR "error\n";' ], + stdout => { into => \$stdout }, + stderr => { into => \$stderr }, + on_finish => sub { }, + ); + + isa_ok( $process->stderr, "IO::Async::Stream", '$process->stderr' ); + + is( $process->stderr->notifier_name, "stderr", '$process->stderr->notifier_name' ); + + $loop->add( $process ); + + ok( defined $process->stderr->read_handle, '$process->stderr has read_handle' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDOUT/STDERR' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDOUT/STDERR' ); + + is( $stdout, "output\n", '$stdout after perl STDOUT/STDERR' ); + is( $stderr, "error\n", '$stderr after perl STDOUT/STDERR' ); +} + +{ + my $stdout; + + my $process = IO::Async::Process->new( + command => [ $^X, "-pe", '$_ = uc' ], + stdin => { via => "pipe_write" }, + stdout => { into => \$stdout }, + on_finish => sub { }, + ); + + isa_ok( $process->stdin, "IO::Async::Stream", '$process->stdin' ); + + is( $process->stdin->notifier_name, "stdin", '$process->stdin->notifier_name' ); + + $process->stdin->write( "some data\n", on_flush => sub { $_[0]->close } ); + + $loop->add( $process ); + + ok( defined $process->stdin->write_handle, '$process->stdin has write_handle for perl STDIN->STDOUT' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIN->STDOUT' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIN->STDOUT' ); + + is( $stdout, "SOME DATA\n", '$stdout after perl STDIN->STDOUT' ); +} + +{ + my $process = IO::Async::Process->new( + command => [ $^X, "-e", 'exit 4' ], + stdin => { via => "pipe_write" }, + on_finish => sub { }, + ); + + isa_ok( $process->stdin, "IO::Async::Stream", '$process->stdin' ); + + $loop->add( $process ); + + ok( defined $process->stdin->write_handle, '$process->stdin has write_handle for perl STDIN no-wait close' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIN no-wait close' ); + is( $process->exitstatus, 4, '$process->exitstatus after perl STDIN no-wait close' ); +} + +{ + my $stdout; + + my $process = IO::Async::Process->new( + command => [ $^X, "-pe", '$_ = uc' ], + stdin => { from => "some data\n" }, + stdout => { into => \$stdout }, + on_finish => sub { }, + ); + + isa_ok( $process->stdin, "IO::Async::Stream", '$process->stdin' ); + + $loop->add( $process ); + + ok( defined $process->stdin->write_handle, '$process->stdin has write_handle for perl STDIN->STDOUT from' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIN->STDOUT from' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIN->STDOUT from' ); + + is( $stdout, "SOME DATA\n", '$stdout after perl STDIN->STDOUT from' ); +} + +{ + my $stdout; + + my $process = IO::Async::Process->new( + command => [ $^X, "-pe", '$_ = "line"' ], + stdin => { from => "" }, + stdout => { into => \$stdout }, + on_finish => sub { }, + ); + + isa_ok( $process->stdin, "IO::Async::Stream", '$process->stdin' ); + + $loop->add( $process ); + + ok( defined $process->stdin->write_handle, '$process->stdin has write_handle for perl STDIN->STDOUT from empty string' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIN->STDOUT from empty string' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIN->STDOUT from empty string' ); + + is( $stdout, "", '$stdout after perl STDIN->STDOUT from empty string' ); +} + +{ + my $stdout; + + my $process = IO::Async::Process->new( + command => [ $^X, "-pe", '$_ = uc' ], + fd0 => { from => "some data\n" }, + fd1 => { into => \$stdout }, + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIN->STDOUT using fd[n]' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIN->STDOUT using fd[n]' ); + + is( $stdout, "SOME DATA\n", '$stdout after perl STDIN->STDOUT using fd[n]' ); +} + +{ + my $output; + + my $process = IO::Async::Process->new( + command => [ $^X, "-pe", '$_ = uc' ], + stdio => { via => "pipe_rdwr" }, + on_finish => sub { }, + ); + + isa_ok( $process->stdio, "IO::Async::Stream", '$process->stdio' ); + + is( $process->stdio->notifier_name, "stdio", '$process->stdio->notifier_name' ); + + my @output_lines; + + $process->stdio->write( "some data\n", on_flush => sub { $_[0]->close_write } ); + $process->stdio->configure( + on_read => sub { + my ( undef, $buffref ) = @_; + push @output_lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + $loop->add( $process ); + + ok( defined $process->stdio->read_handle, '$process->stdio has read_handle for perl STDIO' ); + ok( defined $process->stdio->write_handle, '$process->stdio has write_handle for perl STDIO' ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIO' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIO' ); + + is_deeply( \@output_lines, [ "SOME DATA\n" ], '@output_lines after perl STDIO' ); +} + +{ + my $output; + + my $process = IO::Async::Process->new( + command => [ $^X, "-pe", '$_ = uc' ], + stdio => { + from => "some data\n", + into => \$output, + }, + on_finish => sub { }, + ); + + $loop->add( $process ); + + wait_for { !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIN->STDOUT using stdio' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIN->STDOUT using stdio' ); + + is( $output, "SOME DATA\n", '$stdout after perl STDIN->STDOUT using stdio' ); +} + +{ + my $process = IO::Async::Process->new( + code => sub { + defined( recv STDIN, my $pkt, 8192, 0 ) or die "Cannot recv - $!"; + send STDOUT, $pkt, 0 or die "Cannot send - $!"; + return 0; + }, + stdio => { via => "socketpair" }, + on_finish => sub { }, + ); + + isa_ok( $process->stdio, "IO::Async::Stream", '$process->stdio isa Stream' ); + + $process->stdio->write( "A packet to be echoed" ); + + my $output_packet = ""; + $process->stdio->configure( + on_read => sub { + my ( undef, $buffref ) = @_; + $output_packet .= $$buffref; + $$buffref = ""; + return 0; + }, + ); + + $loop->add( $process ); + + isa_ok( $process->stdio->read_handle, "IO::Socket", '$process->stdio handle isa IO::Socket' ); + + wait_for { defined $output_packet and !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIO via socketpair' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIO via socketpair' ); + + is_deeply( $output_packet, "A packet to be echoed", '$output_packet after perl STDIO via socketpair' ); +} + +{ + my $process = IO::Async::Process->new( + code => sub { return 0 }, + stdio => { via => "socketpair", family => "inet" }, + on_finish => sub { }, + ); + + isa_ok( $process->stdio, "IO::Async::Stream", '$process->stdio isa Stream' ); + + $process->stdio->configure( on_read => sub { } ); + + $loop->add( $process ); + + isa_ok( $process->stdio->read_handle, "IO::Socket", '$process->stdio handle isa IO::Socket' ); + is( sockaddr_family( $process->stdio->read_handle->sockname ), PF_INET, '$process->stdio handle sockdomain is PF_INET' ); + + wait_for { !$process->is_running }; +} + +{ + my $process = IO::Async::Process->new( + code => sub { + for( 1, 2 ) { + defined( recv STDIN, my $pkt, 8192, 0 ) or die "Cannot recv - $!"; + send STDOUT, $pkt, 0 or die "Cannot send - $!"; + } + return 0; + }, + stdio => { via => "socketpair", socktype => "dgram", family => "inet" }, + on_finish => sub { }, + ); + + isa_ok( $process->stdio, "IO::Async::Socket", '$process->stdio isa Socket' ); + + my @output_packets; + $process->stdio->configure( + on_recv => sub { + my ( $self, $packet ) = @_; + push @output_packets, $packet; + + $self->close if @output_packets == 2; + + return 0; + }, + ); + + $loop->add( $process ); + + isa_ok( $process->stdio->read_handle, "IO::Socket", '$process->stdio handle isa IO::Socket' ); + ok( defined sockaddr_family( $process->stdio->read_handle->sockname ), '$process->stdio handle sockdomain is defined' ); + + $process->stdio->send( $_ ) for "First packet", "Second packet"; + + wait_for { @output_packets == 2 and !$process->is_running }; + + ok( $process->is_exited, '$process->is_exited after perl STDIO via dgram socketpair' ); + is( $process->exitstatus, 0, '$process->exitstatus after perl STDIO via dgram socketpair' ); + + is_deeply( \@output_packets, + [ "First packet", "Second packet" ], + '@output_packets after perl STDIO via dgram socketpair' ); +} + +done_testing; diff --git a/t/35loop-openchild.t b/t/35loop-openchild.t new file mode 100644 index 0000000..608952d --- /dev/null +++ b/t/35loop-openchild.t @@ -0,0 +1,65 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my $exitcode; + +$loop->open_child( + code => sub { 0 }, + on_finish => sub { ( undef, $exitcode ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after sub { 0 }' ); +is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after sub { 0 }' ); + +$loop->open_child( + command => [ $^X, "-e", 'exit 5' ], + on_finish => sub { ( undef, $exitcode ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after perl -e exit 5' ); +is( ($exitcode >> 8), 5, 'WEXITSTATUS($exitcode) after perl -e exit 5' ); + +ok( exception { $loop->open_child( + command => [ $^X, "-e", 1 ] + ) }, + 'Missing on_finish fails' +); + +ok( exception { $loop->open_child( + command => [ $^X, "-e", 1 ], + on_finish => "hello" + ) }, + 'on_finish not CODE ref fails' +); + +ok( exception { $loop->open_child( + command => [ $^X, "-e", 1 ], + on_finish => sub {}, + on_exit => sub {}, + ) }, + 'on_exit parameter fails' +); + +done_testing; diff --git a/t/36loop-runchild.t b/t/36loop-runchild.t new file mode 100644 index 0000000..50801d1 --- /dev/null +++ b/t/36loop-runchild.t @@ -0,0 +1,158 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my ( $exitcode, $child_out, $child_err ); + +$loop->run_child( + code => sub { 0 }, + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after sub { 0 }' ); +is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after sub { 0 }' ); +is( $child_out, "", '$child_out after sub { 0 }' ); +is( $child_err, "", '$child_err after sub { 0 }' ); + +$loop->run_child( + code => sub { 3 }, + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after sub { 3 }' ); +is( ($exitcode >> 8), 3, 'WEXITSTATUS($exitcode) after sub { 3 }' ); +is( $child_out, "", '$child_out after sub { 3 }' ); +is( $child_err, "", '$child_err after sub { 3 }' ); + +$loop->run_child( + command => [ $^X, "-e", '1' ], + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after perl -e 1' ); +is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after perl -e 1' ); +is( $child_out, "", '$child_out after perl -e 1' ); +is( $child_err, "", '$child_err after perl -e 1' ); + +$loop->run_child( + command => [ $^X, "-e", 'exit 5' ], + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after perl -e exit 5' ); +is( ($exitcode >> 8), 5, 'WEXITSTATUS($exitcode) after perl -e exit 5' ); +is( $child_out, "", '$child_out after perl -e exit 5' ); +is( $child_err, "", '$child_err after perl -e exit 5' ); + +$loop->run_child( + code => sub { print "hello\n"; 0 }, + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after sub { print }' ); +is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after sub { print }' ); +is( $child_out, "hello\n", '$child_out after sub { print }' ); +is( $child_err, "", '$child_err after sub { print }' ); + +$loop->run_child( + command => [ $^X, "-e", 'print "goodbye\n"' ], + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after perl STDOUT' ); +is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after perl STDOUT' ); +is( $child_out, "goodbye\n", '$child_out after perl STDOUT' ); +is( $child_err, "", '$child_err after perl STDOUT' ); + +$loop->run_child( + command => [ $^X, "-e", 'print STDOUT "output\n"; print STDERR "error\n";' ], + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after perl STDOUT/STDERR' ); +is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after perl STDOUT/STDERR' ); +is( $child_out, "output\n", '$child_out after perl STDOUT/STDERR' ); +is( $child_err, "error\n", '$child_err after perl STDOUT/STDERR' ); + +# perl -pe 1 behaves like cat; copies STDIN to STDOUT + +$loop->run_child( + command => [ $^X, "-pe", '1' ], + stdin => "some data\n", + on_finish => sub { ( undef, $exitcode, $child_out, $child_err ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +ok( ($exitcode & 0x7f) == 0, 'WIFEXITED($exitcode) after perl STDIN->STDOUT' ); +is( ($exitcode >> 8), 0, 'WEXITSTATUS($exitcode) after perl STDIN->STDOUT' ); +is( $child_out, "some data\n", '$child_out after perl STDIN->STDOUT' ); +is( $child_err, "", '$child_err after perl STDIN->STDOUT' ); + +ok( exception { $loop->run_child( + command => [ $^X, "-e", 1 ] + ) }, + 'Missing on_finish fails' +); + +ok( exception { $loop->run_child( + command => [ $^X, "-e", 1 ], + on_finish => "hello" + ) }, + 'on_finish not CODE ref fails' +); + +ok( exception { $loop->run_child( + command => [ $^X, "-e", 1 ], + on_finish => sub {}, + on_exit => sub {}, + ) }, + 'on_exit parameter fails' +); + +ok( exception { $loop->run_child( + command => [ $^X, "-e", 1 ], + on_finish => sub {}, + some_key_you_fail => 1 + ) }, + 'unrecognised key fails' +); + +done_testing; diff --git a/t/37loop-child-root.t b/t/37loop-child-root.t new file mode 100644 index 0000000..237b338 --- /dev/null +++ b/t/37loop-child-root.t @@ -0,0 +1,89 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "POSIX fork() is not available" unless IO::Async::OS->HAVE_POSIX_FORK; + +use POSIX qw( WEXITSTATUS ); + +# These tests check the parts of Loop->spawn_child that need to be root to +# work. Since we're unlikely to be root, skip the lot if we're not. + +unless( $< == 0 ) { + plan skip_all => "not root"; +} + +is( $>, 0, 'am root'); + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my ( $exitcode, $dollarbang, $dollarat ); + +$loop->spawn_child( + code => sub { return $> }, + setup => [ setuid => 10 ], + on_exit => sub { ( undef, $exitcode, $dollarbang, $dollarat ) = @_ }, +); + +wait_for { defined $exitcode }; + +is( WEXITSTATUS($exitcode), 10, 'setuid' ); + +$loop->spawn_child( + code => sub { return $) }, + setup => [ setgid => 10 ], + on_exit => sub { ( undef, $exitcode, $dollarbang, $dollarat ) = @_ }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +is( WEXITSTATUS($exitcode), 10, 'setgid' ); + +$loop->spawn_child( + code => sub { return $) =~ m/ 5 / }, + setup => [ setgroups => [ 4, 5, 6 ] ], + on_exit => sub { ( undef, $exitcode, $dollarbang, $dollarat ) = @_ }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +is( WEXITSTATUS($exitcode), 1, 'setgroups' ); + +my $child_out; + +$loop->run_child( + code => sub { + print "EUID: $>\n"; + my ( $gid, @groups ) = split( m/ /, $) ); + print "EGID: $gid\n"; + print "Groups: " . join( " ", sort { $a <=> $b } @groups ) . "\n"; + return 0; + }, + setup => [ + setgid => 10, + setgroups => [ 4, 5, 6, 10 ], + setuid => 20, + ], + on_finish => sub { ( undef, $exitcode, $child_out ) = @_; }, +); + +undef $exitcode; +wait_for { defined $exitcode }; + +is( $child_out, + "EUID: 20\nEGID: 10\nGroups: 4 5 6 10\n", + 'combined setuid/gid/groups' ); + +done_testing; diff --git a/t/38loop-thread.t b/t/38loop-thread.t new file mode 100644 index 0000000..f2d2389 --- /dev/null +++ b/t/38loop-thread.t @@ -0,0 +1,59 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; + +use IO::Async::Loop; +use IO::Async::OS; + +plan skip_all => "Threads are not available" unless IO::Async::OS->HAVE_THREADS; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# thread in scalar context +{ + my @result; + $loop->create_thread( + code => sub { return "A result" }, + on_joined => sub { @result = @_ }, + ); + + wait_for { @result }; + + is_deeply( \@result, [ return => "A result" ], 'result to on_joined for returning thread' ); +} + +# thread in list context +{ + my @result; + $loop->create_thread( + code => sub { return "A result", "of many", "values" }, + context => "list", + on_joined => sub { @result = @_ }, + ); + + wait_for { @result }; + + is_deeply( \@result, [ return => "A result", "of many", "values" ], 'result to on_joined for returning thread in list context' ); +} + +# thread that dies +{ + my @result; + $loop->create_thread( + code => sub { die "Ooops I fail\n" }, + on_joined => sub { @result = @_ }, + ); + + wait_for { @result }; + + is_deeply( \@result, [ died => "Ooops I fail\n" ], 'result to on_joined for a died thread' ); +} + +done_testing; diff --git a/t/40channel.t b/t/40channel.t new file mode 100644 index 0000000..930129b --- /dev/null +++ b/t/40channel.t @@ -0,0 +1,263 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Identity; + +use IO::Async::Channel; + +use IO::Async::OS; + +use IO::Async::Loop; +use Storable qw( freeze ); + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# sync->sync - mostly doesn't involve IO::Async +{ + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $channel_rd = IO::Async::Channel->new; + $channel_rd->setup_sync_mode( $pipe_rd ); + + my $channel_wr = IO::Async::Channel->new; + $channel_wr->setup_sync_mode( $pipe_wr ); + + $channel_wr->send( [ structure => "here" ] ); + + is_deeply( $channel_rd->recv, [ structure => "here" ], 'Sync mode channels can send/recv structures' ); + + $channel_wr->send_frozen( freeze [ prefrozen => "data" ] ); + + is_deeply( $channel_rd->recv, [ prefrozen => "data" ], 'Sync mode channels can send_frozen' ); + + $channel_wr->close; + + is( $channel_rd->recv, undef, 'Sync mode can be closed' ); +} + +# async->sync +{ + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $channel_rd = IO::Async::Channel->new; + $channel_rd->setup_sync_mode( $pipe_rd ); + + my $channel_wr = IO::Async::Channel->new; + $channel_wr->setup_async_mode( write_handle => $pipe_wr ); + + $loop->add( $channel_wr ); + + $channel_wr->send( [ data => "by async" ] ); + + # Cheat for semi-sync + my $flushed; + $channel_wr->{stream}->write( "", on_flush => sub { $flushed++ } ); + wait_for { $flushed }; + + is_deeply( $channel_rd->recv, [ data => "by async" ], 'Async mode channel can send' ); + + $channel_wr->close; + + is( $channel_rd->recv, undef, 'Sync mode can be closed' ); +} + +# sync->async configured on_recv +{ + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my @recv_queue; + my $recv_eof; + + my $channel_rd = IO::Async::Channel->new; + $channel_rd->setup_async_mode( read_handle => $pipe_rd ); + + $loop->add( $channel_rd ); + + $channel_rd->configure( + on_recv => sub { + identical( $_[0], $channel_rd, 'Channel passed to on_recv' ); + push @recv_queue, $_[1]; + }, + on_eof => sub { + $recv_eof++; + }, + ); + + my $channel_wr = IO::Async::Channel->new; + $channel_wr->setup_sync_mode( $pipe_wr ); + + $channel_wr->send( [ data => "by sync" ] ); + + wait_for { @recv_queue }; + + is_deeply( shift @recv_queue, [ data => "by sync" ], 'Async mode channel can on_recv' ); + + $channel_wr->close; + + wait_for { $recv_eof }; + is( $recv_eof, 1, 'Async mode channel can on_eof' ); +} + +# sync->async oneshot ->recv with future +{ + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $channel_rd = IO::Async::Channel->new; + $channel_rd->setup_async_mode( read_handle => $pipe_rd ); + + $loop->add( $channel_rd ); + + my $channel_wr = IO::Async::Channel->new; + $channel_wr->setup_sync_mode( $pipe_wr ); + + $channel_wr->send( [ data => "by sync" ] ); + + my $recv_f = $channel_rd->recv; + + wait_for { $recv_f->is_ready }; + + is_deeply( scalar $recv_f->get, [ data => "by sync" ], 'Async mode future can receive data' ); + + $channel_wr->close; + + my $eof_f = $channel_rd->recv; + + wait_for { $eof_f->is_ready }; + + is( ( $eof_f->failure )[1], "eof", 'Async mode future can receive EOF' ); +} + +# sync->async oneshot ->recv with callbacks +{ + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $channel_rd = IO::Async::Channel->new; + $channel_rd->setup_async_mode( read_handle => $pipe_rd ); + + $loop->add( $channel_rd ); + + my $channel_wr = IO::Async::Channel->new; + $channel_wr->setup_sync_mode( $pipe_wr ); + + $channel_wr->send( [ data => "by sync" ] ); + + my $recved; + $channel_rd->recv( + on_recv => sub { + identical( $_[0], $channel_rd, 'Channel passed to ->recv on_recv' ); + $recved = $_[1]; + }, + on_eof => sub { die "Test failed early" }, + ); + + wait_for { $recved }; + + is_deeply( $recved, [ data => "by sync" ], 'Async mode channel can ->recv on_recv' ); + + $channel_wr->close; + + my $recv_eof; + $channel_rd->recv( + on_recv => sub { die "Channel recv'ed when not expecting" }, + on_eof => sub { $recv_eof++ }, + ); + + wait_for { $recv_eof }; + is( $recv_eof, 1, 'Async mode channel can ->recv on_eof' ); +} + +# sync->async write once then close +{ + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $channel_rd = IO::Async::Channel->new; + $channel_rd->setup_async_mode( read_handle => $pipe_rd ); + + $loop->add( $channel_rd ); + + my $channel_wr = IO::Async::Channel->new; + $channel_wr->setup_sync_mode( $pipe_wr ); + + $channel_wr->send( [ "One value here" ] ); + $channel_wr->close; + undef $channel_wr; + + my $recved; + $channel_rd->recv( + on_recv => sub { + $recved = $_[1]; + }, + on_eof => sub { die "Test failed early" }, + ); + + wait_for { $recved }; + + is( $recved->[0], "One value here", 'Async mode channel can ->recv buffer at EOF' ); + + $loop->remove( $channel_rd ); +} + +# Async ->recv cancellation +{ + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $channel_rd = IO::Async::Channel->new; + $channel_rd->setup_async_mode( read_handle => $pipe_rd ); + + $loop->add( $channel_rd ); + + my $channel_wr = IO::Async::Channel->new; + $channel_wr->setup_sync_mode( $pipe_wr ); + + $channel_wr->send( [ "first" ] ); + $channel_wr->send( [ "second" ] ); + + my $r1_f = $channel_rd->recv; + my $r2_f = $channel_rd->recv; + + $r1_f->cancel; + + wait_for { $r2_f->is_ready }; + + is_deeply( scalar $r2_f->get, [ "second" ], 'Async recv result after cancellation' ); + + $loop->remove( $channel_rd ); +} + +# Sereal encoder +SKIP: { + skip "Sereal is not available", 1 unless eval { require Sereal::Encoder; require Sereal::Decoder; }; + + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $channel_rd = IO::Async::Channel->new( + codec => "Sereal" + ); + $channel_rd->setup_async_mode( read_handle => $pipe_rd ); + + $loop->add( $channel_rd ); + + my $channel_wr = IO::Async::Channel->new( + codec => "Sereal", + ); + $channel_wr->setup_sync_mode( $pipe_wr ); + + $channel_wr->send( [ data => "by sync" ] ); + + my $recv_f = $channel_rd->recv; + + wait_for { $recv_f->is_ready }; + + is_deeply( scalar $recv_f->get, [ data => "by sync" ], 'Channel can use Sereal as codec' ); + + $loop->remove( $channel_rd ); +} + +done_testing; diff --git a/t/41routine.t b/t/41routine.t new file mode 100644 index 0000000..945a1ec --- /dev/null +++ b/t/41routine.t @@ -0,0 +1,322 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Identity; +use Test::Refcount; + +use IO::Async::Routine; + +use IO::Async::Channel; +use IO::Async::Loop; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +sub test_with_model +{ + my ( $model ) = @_; + + { + my $calls = IO::Async::Channel->new; + my $returns = IO::Async::Channel->new; + + my $routine = IO::Async::Routine->new( + model => $model, + channels_in => [ $calls ], + channels_out => [ $returns ], + code => sub { + while( my $args = $calls->recv ) { + last if ref $args eq "SCALAR"; + + my $ret = 0; + $ret += $_ for @$args; + $returns->send( \$ret ); + } + }, + on_finish => sub {}, + ); + + isa_ok( $routine, "IO::Async::Routine", "\$routine for $model model" ); + is_oneref( $routine, "\$routine has refcount 1 initially for $model model" ); + + $loop->add( $routine ); + + is_refcount( $routine, 2, "\$routine has refcount 2 after \$loop->add for $model model" ); + + is( $routine->model, $model, "\$routine->model for $model model" ); + + $calls->send( [ 1, 2, 3 ] ); + + my $f = $returns->recv; + + wait_for { $f->is_ready }; + + my $result = $f->get; + is( ${$result}, 6, "Result for $model model" ); + + is_refcount( $routine, 2, '$routine has refcount 2 before $loop->remove' ); + + $loop->remove( $routine ); + + is_oneref( $routine, '$routine has refcount 1 before EOF' ); + } + + { + my $returned; + my $return_routine = IO::Async::Routine->new( + model => $model, + code => sub { return 23 }, + on_return => sub { $returned = $_[1]; }, + ); + + $loop->add( $return_routine ); + + wait_for { defined $returned }; + + is( $returned, 23, "on_return for $model model" ); + + my $died; + my $die_routine = IO::Async::Routine->new( + model => $model, + code => sub { die "ARGH!\n" }, + on_die => sub { $died = $_[1]; }, + ); + + $loop->add( $die_routine ); + + wait_for { defined $died }; + + is( $died, "ARGH!\n", "on_die for $model model" ); + } + + { + my $channel = IO::Async::Channel->new; + + my $finished; + my $routine = IO::Async::Routine->new( + model => $model, + channels_in => [ $channel ], + code => sub { while( $channel->recv ) { 1 } }, + on_finish => sub { $finished++ }, + ); + + $loop->add( $routine ); + + $channel->close; + + wait_for { $finished }; + pass( "Recv on closed channel for $model model" ); + } + + { + my $channel = IO::Async::Channel->new; + + my $routine = IO::Async::Routine->new( + model => $model, + channels_out => [ $channel ], + code => sub { + $SIG{INT} = sub { $channel->send( \"SIGINT" ); die "SIGINT" }; + $channel->send( \"READY" ); + + # Busy-wait so thread kill still works + my $until = time() + 5; + 1 while time() < $until; + }, + ); + + $loop->add( $routine ); + + my $f; + $f = $channel->recv; + + wait_for { $f->is_ready }; + + is( ${ $f->get }, "READY", 'Routine is ready for SIGINT' ); + + $routine->kill( "INT" ); + + $f = $channel->recv; + + wait_for { $f->is_ready }; + + is( ${ $f->get }, "SIGINT", 'Routine caught SIGINT' ); + } +} + +foreach my $model (qw( fork thread )) { + SKIP: { + skip "This Perl does not support threads", 9 + if $model eq "thread" and not IO::Async::OS->HAVE_THREADS; + skip "This Perl does not support fork()", 9 + if $model eq "fork" and not IO::Async::OS->HAVE_POSIX_FORK; + + test_with_model( $model ); + } +} + +# multiple channels in and out +{ + my $in1 = IO::Async::Channel->new; + my $in2 = IO::Async::Channel->new; + my $out1 = IO::Async::Channel->new; + my $out2 = IO::Async::Channel->new; + + my $routine = IO::Async::Routine->new( + channels_in => [ $in1, $in2 ], + channels_out => [ $out1, $out2 ], + code => sub { + while( my $op = $in1->recv ) { + $op = $$op; # deref + $out1->send( \"Ready $op" ); + my @args = @{ $in2->recv }; + my $result = $op eq "+" ? $args[0] + $args[1] + : "ERROR"; + $out2->send( \$result ); + } + }, + on_finish => sub { }, + ); + + isa_ok( $routine, "IO::Async::Routine", '$routine' ); + + $loop->add( $routine ); + + $in1->send( \"+" ); + + my $status_f = $out1->recv; + + wait_for { $status_f->is_ready }; + is( ${ $status_f->get }, "Ready +", '$status_f result midway through Routine' ); + + $in2->send( [ 10, 20 ] ); + + my $result_f = $out2->recv; + + wait_for { $result_f->is_ready }; + + is( ${ $result_f->get }, 30, '$result_f result at end of Routine' ); + + $loop->remove( $routine ); +} + +# sharing a Channel between Routines +{ + my $channel = IO::Async::Channel->new; + + my $src_finished; + my $src_routine = IO::Async::Routine->new( + channels_out => [ $channel ], + code => sub { + $channel->send( [ some => "data" ] ); + return 0; + }, + on_finish => sub { $src_finished++ }, + on_die => sub { die "source routine failed - $_[1]" }, + ); + + $loop->add( $src_routine ); + + my $sink_result; + my $sink_routine = IO::Async::Routine->new( + channels_in => [ $channel ], + code => sub { + my @data = @{ $channel->recv }; + return ( $data[0] eq "some" and $data[1] eq "data" ) ? 0 : 1; + }, + on_return => sub { $sink_result = $_[1] }, + on_die => sub { die "sink routine failed - $_[1]" }, + ); + + $loop->add( $sink_routine ); + + wait_for { $src_finished and defined $sink_result }; + + is( $sink_result, 0, 'synchronous src->sink can share a channel' ); +} + +# Test that 'setup' works +SKIP: { + skip "This Perl does not support fork()", 1 + if not IO::Async::OS->HAVE_POSIX_FORK; + + my $channel = IO::Async::Channel->new; + + my $routine = IO::Async::Routine->new( + model => "fork", + setup => [ + env => { FOO => "Here is a random string" }, + ], + + channels_out => [ $channel ], + code => sub { + $channel->send( [ $ENV{FOO} ] ); + $channel->close; + return 0; + }, + on_finish => sub {}, + ); + + $loop->add( $routine ); + + my $f = $channel->recv; + + wait_for { $f->is_ready }; + + my $result = $f->get; + is( $result->[0], "Here is a random string", '$result from Routine with modified ENV' ); + + $loop->remove( $routine ); +} + +# Test that STDOUT/STDERR are unaffected +SKIP: { + skip "This Perl does not support fork()", 1 + if not IO::Async::OS->HAVE_POSIX_FORK; + + my ( $pipe_rd, $pipe_wr ) = IO::Async::OS->pipepair; + + my $routine; + { + open my $stdoutsave, ">&", \*STDOUT; + POSIX::dup2( $pipe_wr->fileno, STDOUT->fileno ); + + open my $stderrsave, ">&", \*STDERR; + POSIX::dup2( $pipe_wr->fileno, STDERR->fileno ); + + $routine = IO::Async::Routine->new( + model => "fork", + code => sub { + STDOUT->autoflush(1); + print STDOUT "A line to STDOUT\n"; + print STDERR "A line to STDERR\n"; + return 0; + } + ); + + $loop->add( $routine ); + + POSIX::dup2( $stdoutsave->fileno, STDOUT->fileno ); + POSIX::dup2( $stderrsave->fileno, STDERR->fileno ); + } + + my $buffer = ""; + $loop->watch_io( + handle => $pipe_rd, + on_read_ready => sub { sysread $pipe_rd, $buffer, 8192, length $buffer or die "Cannot read - $!" }, + ); + + wait_for { $buffer =~ m/\n.*\n/ }; + + is( $buffer, "A line to STDOUT\nA line to STDERR\n", 'Write-to-STD{OUT+ERR} wrote to pipe' ); + + $loop->unwatch_io( handle => $pipe_rd, on_read_ready => 1 ); + $loop->remove( $routine ); +} + +done_testing; diff --git a/t/42function.t b/t/42function.t new file mode 100644 index 0000000..f4b1c4d --- /dev/null +++ b/t/42function.t @@ -0,0 +1,569 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Fatal; +use Test::Refcount; +use constant HAVE_TEST_MEMORYGROWTH => eval { require Test::MemoryGrowth; }; + +use File::Temp qw( tempdir ); +use Time::HiRes qw( sleep ); + +use IO::Async::Function; + +use IO::Async::OS; + +use IO::Async::Loop; + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# by future +{ + my $function = IO::Async::Function->new( + min_workers => 1, + max_workers => 1, + code => sub { return $_[0] + $_[1] }, + ); + + ok( defined $function, '$function defined' ); + isa_ok( $function, "IO::Async::Function", '$function isa IO::Async::Function' ); + + is_oneref( $function, '$function has refcount 1' ); + + $loop->add( $function ); + + is_refcount( $function, 2, '$function has refcount 2 after $loop->add' ); + + is( $function->workers, 1, '$function has 1 worker' ); + is( $function->workers_busy, 0, '$function has 0 workers busy' ); + is( $function->workers_idle, 1, '$function has 1 workers idle' ); + + my $future = $function->call( + args => [ 10, 20 ], + ); + + isa_ok( $future, "Future", '$future' ); + + is_refcount( $function, 2, '$function has refcount 2 after ->call' ); + + is( $function->workers_busy, 1, '$function has 1 worker busy after ->call' ); + is( $function->workers_idle, 0, '$function has 0 worker idle after ->call' ); + + wait_for { $future->is_ready }; + + my ( $result ) = $future->get; + + is( $result, 30, '$result after call returns by future' ); + + is( $function->workers_busy, 0, '$function has 0 workers busy after call returns' ); + is( $function->workers_idle, 1, '$function has 1 workers idle after call returns' ); + + $loop->remove( $function ); +} + +# by callback +{ + my $function = IO::Async::Function->new( + min_workers => 1, + max_workers => 1, + code => sub { return $_[0] + $_[1] }, + ); + + $loop->add( $function ); + + my $result; + + $function->call( + args => [ 10, 20 ], + on_return => sub { $result = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + + wait_for { defined $result }; + + is( $result, 30, '$result after call returns by callback' ); + + $loop->remove( $function ); +} + +# Test queueing +{ + my $function = IO::Async::Function->new( + min_workers => 1, + max_workers => 1, + code => sub { return $_[0] + $_[1] }, + ); + + $loop->add( $function ); + + my @result; + + my $f1 = $function->call( + args => [ 1, 2 ], + on_return => sub { push @result, shift }, + on_error => sub { die "Test failed early - @_" }, + ); + my $f2 = $function->call( + args => [ 3, 4 ], + on_return => sub { push @result, shift }, + on_error => sub { die "Test failed early - @_" }, + ); + + is( $function->workers, 1, '$function->workers is still 1 after 2 calls' ); + + isa_ok( $f1, "Future", '$f1' ); + isa_ok( $f2, "Future", '$f2' ); + + wait_for { @result == 2 }; + + is_deeply( \@result, [ 3, 7 ], '@result after both calls return' ); + + is( $function->workers, 1, '$function->workers is still 1 after 2 calls return' ); + + $loop->remove( $function ); +} + +# References +{ + my $function = IO::Async::Function->new( + code => sub { return ref( $_[0] ), \$_[1] }, + ); + + $loop->add( $function ); + + my @result; + + $function->call( + args => [ \'a', 'b' ], + on_return => sub { @result = @_ }, + on_error => sub { die "Test failed early - @_" }, + ); + + wait_for { scalar @result }; + + is_deeply( \@result, [ 'SCALAR', \'b' ], 'Call and result preserves references' ); + + $loop->remove( $function ); +} + +# Exception throwing +{ + my $line = __LINE__ + 2; + my $function = IO::Async::Function->new( + code => sub { die shift }, + ); + + $loop->add( $function ); + + my $err; + + my $f = $function->call( + args => [ "exception name" ], + on_return => sub { }, + on_error => sub { $err = shift }, + ); + + wait_for { defined $err }; + + like( $err, qr/^exception name at \Q$0\E line \d+\.$/, '$err after exception' ); + + is_deeply( [ $f->failure ], + [ "exception name at $0 line $line.", error => ], + '$f->failure after exception' ); + + $loop->remove( $function ); +} + +# max_workers +{ + my $count = 0; + + my $function = IO::Async::Function->new( + max_workers => 1, + code => sub { $count++; die "$count\n" }, + exit_on_die => 0, + ); + + $loop->add( $function ); + + my @errs; + $function->call( + args => [], + on_return => sub { }, + on_error => sub { push @errs, shift }, + ); + $function->call( + args => [], + on_return => sub { }, + on_error => sub { push @errs, shift }, + ); + + undef @errs; + wait_for { scalar @errs == 2 }; + + is_deeply( \@errs, [ "1", "2" ], 'Closed variables preserved when exit_on_die => 0' ); + + $loop->remove( $function ); +} + +# exit_on_die +{ + my $count = 0; + + my $function = IO::Async::Function->new( + max_workers => 1, + code => sub { $count++; die "$count\n" }, + exit_on_die => 1, + ); + + $loop->add( $function ); + + my @errs; + $function->call( + args => [], + on_return => sub { }, + on_error => sub { push @errs, shift }, + ); + $function->call( + args => [], + on_return => sub { }, + on_error => sub { push @errs, shift }, + ); + + undef @errs; + wait_for { scalar @errs == 2 }; + + is_deeply( \@errs, [ "1", "1" ], 'Closed variables preserved when exit_on_die => 1' ); + + $loop->remove( $function ); +} + +# restart after exit +SKIP: { + skip "This Perl does not support fork()", 4 + if not IO::Async::OS->HAVE_POSIX_FORK; + + my $function = IO::Async::Function->new( + model => "fork", + min_workers => 0, + max_workers => 1, + code => sub { $_[0] ? exit shift : return 0 }, + ); + + $loop->add( $function ); + + my $err; + + $function->call( + args => [ 16 ], + on_return => sub { $err = "" }, + on_error => sub { $err = [ @_ ] }, + ); + + wait_for { defined $err }; + + # Not sure what reason we might get - need to check both + ok( $err->[0] eq "closed" || $err->[0] eq "exit", '$err->[0] after child death' ) + or diag( 'Expected "closed" or "exit", found ' . $err->[0] ); + + is( scalar $function->workers, 0, '$function->workers is now 0' ); + + $function->call( + args => [ 0 ], + on_return => sub { $err = "return" }, + on_error => sub { $err = [ @_ ] }, + ); + + is( scalar $function->workers, 1, '$function->workers is now 1 again' ); + + undef $err; + wait_for { defined $err }; + + is( $err, "return", '$err is "return" after child nondeath' ); + + $loop->remove( $function ); +} + +## Now test that parallel runs really are parallel +{ + # touch $dir/$n in each worker, touch $dir/done to finish it + sub touch + { + my ( $file ) = @_; + + open( my $fh, ">", $file ) or die "Cannot write $file - $!"; + close( $fh ); + } + + my $function = IO::Async::Function->new( + min_workers => 3, + code => sub { + my ( $dir, $n ) = @_; + my $file = "$dir/$n"; + + touch( $file ); + + # Wait for synchronisation + sleep 0.1 while ! -e "$dir/done"; + + unlink( $file ); + + return $n; + }, + ); + + $loop->add( $function ); + + is( scalar $function->workers, 3, '$function->workers is 3' ); + + my $dir = tempdir( CLEANUP => 1 ); + + my %ret; + + foreach my $id ( 1, 2, 3 ) { + $function->call( + args => [ $dir, $id ], + on_return => sub { $ret{$id} = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + } + + wait_for { -e "$dir/1" and -e "$dir/2" and -e "$dir/3" }; + + ok( 1, 'synchronise files created' ); + + # Synchronize deleting them; + touch( "$dir/done" ); + + undef %ret; + wait_for { keys %ret == 3 }; + + unlink( "$dir/done" ); + + is_deeply( \%ret, { 1 => 1, 2 => 2, 3 => 3 }, 'ret keys after parallel run' ); + + is( scalar $function->workers, 3, '$function->workers is still 3' ); + + $loop->remove( $function ); +} + +# Test for idle timeout +{ + my $function = IO::Async::Function->new( + min_workers => 0, + max_workers => 1, + idle_timeout => 2 * AUT, + code => sub { return $_[0] }, + ); + + $loop->add( $function ); + + my $result; + + $function->call( + args => [ 1 ], + on_result => sub { $result = $_[0] }, + ); + + wait_for { defined $result }; + + is( $function->workers, 1, '$function has 1 worker after call' ); + + my $waited; + $loop->watch_time( after => 1 * AUT, code => sub { $waited++ } ); + + wait_for { $waited }; + + is( $function->workers, 1, '$function still has 1 worker after short delay' ); + + undef $result; + $function->call( + args => [ 1 ], + on_result => sub { $result = $_[0] }, + ); + + wait_for { defined $result }; + + undef $waited; + $loop->watch_time( after => 3 * AUT, code => sub { $waited++ } ); + + wait_for { $waited }; + + is( $function->workers, 0, '$function has 0 workers after longer delay' ); + + $loop->remove( $function ); +} + +# Restart +{ + my $value = 1; + + my $function = IO::Async::Function->new( + code => sub { return $value }, + ); + + $loop->add( $function ); + + my $result; + $function->call( + args => [], + on_return => sub { $result = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + + wait_for { defined $result }; + + is( $result, 1, '$result before restart' ); + + $value = 2; + $function->restart; + + undef $result; + $function->call( + args => [], + on_return => sub { $result = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + + wait_for { defined $result }; + + is( $result, 2, '$result after restart' ); + + undef $result; + $function->call( + args => [], + on_return => sub { $result = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + + $function->restart; + + wait_for { defined $result }; + + is( $result, 2, 'call before restart still returns result' ); + + $loop->remove( $function ); +} + +# max_worker_calls +{ + my $counter; + my $function = IO::Async::Function->new( + max_workers => 1, + max_worker_calls => 2, + code => sub { return ++$counter; } + ); + + $loop->add( $function ); + + my $result; + $function->call( + args => [], + on_return => sub { $result = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + wait_for { defined $result }; + is( $result, 1, '$result from first call' ); + + undef $result; + $function->call( + args => [], + on_return => sub { $result = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + wait_for { defined $result }; + is( $result, 2, '$result from second call' ); + + undef $result; + $function->call( + args => [], + on_return => sub { $result = shift }, + on_error => sub { die "Test failed early - @_" }, + ); + wait_for { defined $result }; + is( $result, 1, '$result from third call' ); + + $loop->remove( $function ); +} + +# Cancellation of sent calls +{ + my $function = IO::Async::Function->new( + max_workers => 1, + code => sub { + return 123; + }, + ); + + $loop->add( $function ); + + my $f1 = $function->call( args => [] ); + $f1->cancel; + + my $f2 = $function->call( args => [] ); + + wait_for { $f2->is_ready }; + + is( scalar $f2->get, 123, 'Result of function call after cancelled call' ); + + $loop->remove( $function ); +} + +# Cancellation of pending calls +{ + my $function = IO::Async::Function->new( + max_workers => 1, + code => do { my $state; sub { + my $oldstate = $state; + $state = shift; + return $oldstate; + } }, + ); + + $loop->add( $function ); + + # Queue 3 calls but immediately cancel the middle one + my ( $f1, $f2, $f3 ) = map { + $function->call( args => [ $_ ] ) + } 1 .. 3; + + $f2->cancel; + + wait_for { $f1->is_ready and $f3->is_ready }; + + is( scalar $f1->get, undef, '$f1 result is undef' ); + is( scalar $f3->get, 1, '$f3 result is 1' ); + + $loop->remove( $function ); +} + +# Leak test (RT99552) +if( HAVE_TEST_MEMORYGROWTH ) { + diag( "Performing memory leak test" ); + + my $function = IO::Async::Function->new( + max_workers => 8, + code => sub {}, + ); + + $loop->add( $function ); + + Test::MemoryGrowth::no_growth( sub { + $function->restart; + $function->call( args => [] )->get; + }, calls => 100, + 'IO::Async::Function calls do not leak memory' ); + + $loop->remove( $function ); + undef $function; +} + +done_testing; diff --git a/t/50resolver.t b/t/50resolver.t new file mode 100644 index 0000000..c73bc0f --- /dev/null +++ b/t/50resolver.t @@ -0,0 +1,389 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; + +use Socket 1.93 qw( + AF_INET SOCK_STREAM INADDR_LOOPBACK AI_PASSIVE + pack_sockaddr_in getaddrinfo getnameinfo +); + +use IO::Async::Loop; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my $resolver = $loop->resolver; +isa_ok( $resolver, "IO::Async::Resolver", '$loop->resolver' ); + +SKIP: { + my @pwuid; + defined eval { @pwuid = getpwuid( $< ) } or + skip "No getpwuid()", 5; + + { + my $future = $resolver->resolve( + type => 'getpwuid', + data => [ $< ], + ); + + isa_ok( $future, "Future", '$future' ); + + wait_for { $future->is_ready }; + + my @result = $future->get; + + is_deeply( \@result, \@pwuid, 'getpwuid from future' ); + } + + { + my $result; + + $resolver->resolve( + type => 'getpwuid', + data => [ $< ], + on_resolved => sub { $result = [ @_ ] }, + on_error => sub { die "Test died early" }, + ); + + wait_for { $result }; + + is_deeply( $result, \@pwuid, 'getpwuid' ); + } + + { + my $result; + + $loop->resolve( + type => 'getpwuid', + data => [ $< ], + on_resolved => sub { $result = [ @_ ] }, + on_error => sub { die "Test died early" }, + ); + + wait_for { $result }; + + is_deeply( $result, \@pwuid, 'getpwuid via $loop->resolve' ); + } + + SKIP: { + my $user_name = $pwuid[0]; + skip "getpwnam - No user name", 1 unless defined $user_name; + + my @pwnam = getpwnam( $user_name ); + + my $result; + + $resolver->resolve( + type => 'getpwnam', + data => [ $user_name ], + on_resolved => sub { $result = [ @_ ] }, + on_error => sub { die "Test died early" }, + ); + + wait_for { $result }; + + is_deeply( $result, \@pwnam, 'getpwnam' ); + } +} + +my @proto = getprotobyname( "tcp" ); + +{ + my $result; + + $resolver->resolve( + type => 'getprotobyname', + data => [ "tcp" ], + on_resolved => sub { $result = [ @_ ] }, + on_error => sub { die "Test died early" }, + ); + + wait_for { $result }; + + is_deeply( $result, \@proto, 'getprotobyname' ); +} + +SKIP: { + my $proto_number = $proto[2]; + skip "getprotobynumber - No protocol number", 1 unless defined $proto_number; + + my @proto = getprotobynumber( $proto_number ); + + my $result; + + $resolver->resolve( + type => 'getprotobynumber', + data => [ $proto_number ], + on_resolved => sub { $result = [ @_ ] }, + on_error => sub { die "Test died early" }, + ); + + wait_for { $result }; + + is_deeply( $result, \@proto, 'getprotobynumber' ); +} + +# Some systems seem to mangle the order of results between PF_INET and +# PF_INET6 depending on who asks. We'll hint AF_INET + SOCK_STREAM to minimise +# the risk of a spurious test failure because of ordering issues + +my ( $localhost_err, @localhost_addrs ) = getaddrinfo( "localhost", "www", { family => AF_INET, socktype => SOCK_STREAM } ); + +{ + my $result; + + $resolver->resolve( + type => 'getaddrinfo_array', + data => [ "localhost", "www", "inet", "stream" ], + on_resolved => sub { $result = [ 'resolved', @_ ] }, + on_error => sub { $result = [ 'error', @_ ] }, + ); + + wait_for { $result }; + + if( $localhost_err ) { + is( $result->[0], "error", 'getaddrinfo_array - error' ); + is_deeply( $result->[1], "$localhost_err\n", 'getaddrinfo_array - error message' ); + } + else { + is( $result->[0], "resolved", 'getaddrinfo_array - resolved' ); + + my @got = @{$result}[1..$#$result]; + my @expect = map { [ @{$_}{qw( family socktype protocol addr canonname )} ] } @localhost_addrs; + + is_deeply( \@got, \@expect, 'getaddrinfo_array - resolved addresses' ); + } +} + +{ + my $result; + + $resolver->resolve( + type => 'getaddrinfo_hash', + data => [ host => "localhost", service => "www", family => "inet", socktype => "stream" ], + on_resolved => sub { $result = [ 'resolved', @_ ] }, + on_error => sub { $result = [ 'error', @_ ] }, + ); + + wait_for { $result }; + + if( $localhost_err ) { + is( $result->[0], "error", 'getaddrinfo_hash - error' ); + is_deeply( $result->[1], "$localhost_err\n", 'getaddrinfo_hash - error message' ); + } + else { + is( $result->[0], "resolved", 'getaddrinfo_hash - resolved' ); + + my @got = @{$result}[1..$#$result]; + + is_deeply( \@got, \@localhost_addrs, 'getaddrinfo_hash - resolved addresses' ); + } +} + +{ + my $result; + + $resolver->getaddrinfo( + host => "localhost", + service => "www", + family => "inet", + socktype => "stream", + on_resolved => sub { $result = [ 'resolved', @_ ] }, + on_error => sub { $result = [ 'error', @_ ] }, + ); + + wait_for { $result }; + + if( $localhost_err ) { + is( $result->[0], "error", '$resolver->getaddrinfo - error' ); + is_deeply( $result->[1], "$localhost_err\n", '$resolver->getaddrinfo - error message' ); + } + else { + is( $result->[0], "resolved", '$resolver->getaddrinfo - resolved' ); + + my @got = @{$result}[1..$#$result]; + + is_deeply( \@got, \@localhost_addrs, '$resolver->getaddrinfo - resolved addresses' ); + } +} + +{ + my $future = $resolver->getaddrinfo( + host => "localhost", + service => "www", + family => "inet", + socktype => "stream", + ); + + isa_ok( $future, "Future", '$future for $resolver->getaddrinfo' ); + + wait_for { $future->is_ready }; + + if( $localhost_err ) { + is( scalar $future->failure, "$localhost_err\n", '$resolver->getaddrinfo - error message' ); + } + else { + my @got = $future->get; + + is_deeply( \@got, \@localhost_addrs, '$resolver->getaddrinfo - resolved addresses' ); + } +} + +{ + my ( $lo_err, @lo_addrs ) = getaddrinfo( "127.0.0.1", "80", { socktype => SOCK_STREAM } ); + + my $result; + + $resolver->getaddrinfo( + host => "127.0.0.1", + service => "80", + socktype => SOCK_STREAM, + on_resolved => sub { $result = [ 'resolved', @_ ] }, + on_error => sub { $result = [ 'error', @_ ] }, + ); + + is( $result->[0], 'resolved', '$resolver->getaddrinfo on numeric host/service is synchronous' ); + + my @got = @{$result}[1..$#$result]; + + is_deeply( \@got, \@lo_addrs, '$resolver->getaddrinfo resolved addresses synchronously' ); +} + +{ + my ( $passive_err, @passive_addrs ) = getaddrinfo( "", "3000", { socktype => SOCK_STREAM, family => AF_INET, flags => AI_PASSIVE } ); + + my $result; + + $resolver->getaddrinfo( + family => "inet", + service => "3000", + socktype => "stream", + passive => 1, + on_resolved => sub { $result = [ 'resolved', @_ ] }, + on_error => sub { $result = [ 'error', @_ ] }, + ); + + if( $passive_err ) { + is( $result->[0], "error", '$resolver->getaddrinfo passive - error synchronously' ); + is_deeply( $result->[1], "$passive_err\n", '$resolver->getaddrinfo passive - error message' ); + } + else { + is( $result->[0], "resolved", '$resolver->getaddrinfo passive - resolved synchronously' ); + + my @got = @{$result}[1..$#$result]; + + is_deeply( \@got, \@passive_addrs, '$resolver->getaddrinfo passive - resolved addresses' ); + } +} + +{ + my ( $lo_err, @lo_addrs ) = getaddrinfo( "127.0.0.1", "80", { socktype => SOCK_STREAM } ); + + my $future = $resolver->getaddrinfo( + host => "127.0.0.1", + service => "80", + socktype => SOCK_STREAM, + ); + + isa_ok( $future, "Future", '$future for $resolver->getaddrinfo numerical' ); + + wait_for { $future->is_ready }; + + my @got = $future->get; + + is_deeply( \@got, \@lo_addrs, '$resolver->getaddrinfo resolved addresses synchronously' ); +} + +my $testaddr = pack_sockaddr_in( 80, INADDR_LOOPBACK ); +my ( $testerr, $testhost, $testserv ) = getnameinfo( $testaddr ); + +{ + my $result; + + $resolver->getnameinfo( + addr => $testaddr, + on_resolved => sub { $result = [ 'resolved', @_ ] }, + on_error => sub { $result = [ 'error', @_ ] }, + ); + + wait_for { $result }; + + if( $testerr ) { + is( $result->[0], "error", '$resolver->getnameinfo - error' ); + is_deeply( $result->[1], "$testerr\n", '$resolver->getnameinfo - error message' ); + } + else { + is( $result->[0], "resolved", '$resolver->getnameinfo - resolved' ); + is_deeply( [ @{$result}[1..2] ], [ $testhost, $testserv ], '$resolver->getnameinfo - resolved names' ); + } +} + +{ + my $future = $resolver->getnameinfo( + addr => $testaddr, + ); + + wait_for { $future->is_ready }; + + if( $testerr ) { + is( scalar $future->failure, "$testerr\n", '$resolver->getnameinfo - error message from future' ); + } + else { + my @got = $future->get; + + is_deeply( \@got, [ $testhost, $testserv ], '$resolver->getnameinfo - resolved names from future' ); + } +} + +{ + my $result; + + $resolver->getnameinfo( + addr => $testaddr, + numeric => 1, + on_resolved => sub { $result = [ 'resolved', @_ ] }, + on_error => sub { $result = [ 'error', @_ ] }, + ); + + is_deeply( $result, [ resolved => "127.0.0.1", 80 ], '$resolver->getnameinfo with numeric is synchronous' ); +} + +{ + my $future = $resolver->getnameinfo( + addr => $testaddr, + numeric => 1, + ); + + is_deeply( [ $future->get ], [ "127.0.0.1", 80 ], '$resolver->getnameinfo with numeric is synchronous for future' ); +} + +# $loop->set_resolver +{ + my $callcount = 0; + { + package MockResolver; + use base qw( IO::Async::Notifier ); + + sub new { bless {}, shift } + + sub resolve { + $callcount++; return Future->done(); + } + sub getaddrinfo {} + sub getnameinfo {} + } + + $loop->set_resolver( MockResolver->new ); + + $loop->resolve( type => "getpwuid", data => [ 0 ] )->get; + + is( $callcount, 1, '$callcount 1 after ->resolve' ); +} + +done_testing; diff --git a/t/51loop-connect.t b/t/51loop-connect.t new file mode 100644 index 0000000..c217397 --- /dev/null +++ b/t/51loop-connect.t @@ -0,0 +1,333 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Identity; + +use IO::Socket::INET; +use POSIX qw( ENOENT ); +use Socket qw( AF_UNIX inet_ntoa ); + +use IO::Async::Loop; + +use IO::Async::Stream; +use IO::Async::Socket; + +# Some odd locations like BSD jails might not like INADDR_LOOPBACK. We'll +# establish a baseline first to test against +my $INADDR_LOOPBACK = do { + my $localsock = IO::Socket::INET->new( LocalAddr => "localhost", Listen => 1 ); + $localsock->sockaddr; +}; +my $INADDR_LOOPBACK_HOST = inet_ntoa( $INADDR_LOOPBACK ); +if( $INADDR_LOOPBACK ne INADDR_LOOPBACK ) { + diag( "Testing with INADDR_LOOPBACK=$INADDR_LOOPBACK_HOST; this may be because of odd networking" ); +} + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# Try connect(2)ing to a socket we've just created +my $listensock = IO::Socket::INET->new( + Type => SOCK_STREAM, + LocalAddr => 'localhost', + LocalPort => 0, + Listen => 1 +) or die "Cannot create listensock - $!"; + +my $addr = $listensock->sockname; + +{ + my $future = $loop->connect( + addr => { family => "inet", socktype => "stream", addr => $addr }, + ); + + isa_ok( $future, "Future", '$future' ); + + wait_for { $future->is_ready }; + + my ( $sock ) = $future->get; + + can_ok( $sock, qw( peerhost peerport ) ); + is_deeply( [ unpack_sockaddr_in $sock->peername ], + [ unpack_sockaddr_in $addr ], 'by addr: $sock->getpeername is $addr from future' ); + + $listensock->accept; # Throw it away +} + +# handle +{ + my $future = $loop->connect( + handle => my $given_stream = IO::Async::Stream->new, + addr => { family => "inet", socktype => "stream", addr => $addr }, + ); + + isa_ok( $future, "Future", '$future for ->connect( handle )' ); + + wait_for { $future->is_ready }; + + my $stream = $future->get; + identical( $stream, $given_stream, '$future->get returns given Stream' ); + ok( my $sock = $stream->read_handle, '$stream has a read handle' ); + is_deeply( [ unpack_sockaddr_in $sock->peername ], + [ unpack_sockaddr_in $addr ], 'Returned $stream->read_handle->getpeername is $addr' ); + + $listensock->accept; # Throw it away +} + +# legacy callbacks +{ + my $sock; + + $loop->connect( + addr => { family => "inet", socktype => "stream", addr => $addr }, + on_connected => sub { $sock = shift; }, + on_connect_error => sub { die "Test died early - connect error $_[0]() - $_[-1]\n"; }, + ); + + wait_for { $sock }; + + # Not sure if it'll be an IO::Socket::INET or ::IP, but either way it should support these + can_ok( $sock, qw( peerhost peerport ) ); + is_deeply( [ unpack_sockaddr_in $sock->peername ], + [ unpack_sockaddr_in $addr ], 'by addr: $sock->getpeername is $addr' ); + + $listensock->accept; # Throw it away +} + +# Now try by name +{ + my $future = $loop->connect( + host => $listensock->sockhost, + service => $listensock->sockport, + socktype => $listensock->socktype, + ); + + isa_ok( $future, "Future", '$future' ); + + wait_for { $future->is_ready }; + + my ( $sock ) = $future->get; + + can_ok( $sock, qw( peerhost peerport ) ); + is_deeply( [ unpack_sockaddr_in $sock->peername ], + [ unpack_sockaddr_in $addr ], 'by host/service: $sock->getpeername is $addr from future' ); + + is( $sock->sockhost, $INADDR_LOOPBACK_HOST, '$sock->sockhost is INADDR_LOOPBACK_HOST from future' ); + + $listensock->accept; # Throw it away +} + +# legacy callbacks +{ + my $sock; + + $loop->connect( + host => $listensock->sockhost, + service => $listensock->sockport, + socktype => $listensock->socktype, + on_connected => sub { $sock = shift; }, + on_resolve_error => sub { die "Test died early - resolve error - $_[-1]\n"; }, + on_connect_error => sub { die "Test died early - connect error $_[0]() - $_[-1]\n"; }, + ); + + wait_for { $sock }; + + can_ok( $sock, qw( peerhost peerport ) ); + is_deeply( [ unpack_sockaddr_in $sock->peername ], + [ unpack_sockaddr_in $addr ], 'by host/service: $sock->getpeername is $addr' ); + + is( $sock->sockhost, $INADDR_LOOPBACK_HOST, '$sock->sockhost is INADDR_LOOPBACK_HOST' ); + + $listensock->accept; # Throw it away +} + +SKIP: { + # Some OSes can't bind(2) locally to other addresses on 127./8 + skip "Cannot bind to 127.0.0.2", 1 unless eval { IO::Socket::INET->new( + LocalHost => "127.0.0.2", LocalPort => 0 + ) }; + + # Some can bind(2) but then cannot connect() to 127.0.0.1 from it + chomp($@), skip "Cannot connect to 127.0.0.1 from 127.0.0.2 - $@", 1 unless eval { + my $s = IO::Socket::INET->new( + LocalHost => "127.0.0.2", LocalPort => 0, + PeerHost => $listensock->sockhost, PeerPort => $listensock->sockport, + ) or die $@; + $listensock->accept; # Throw it away + $s->sockhost eq "127.0.0.2" or die "sockhost is not 127.0.0.2\n"; }; + + my $sock; + + $loop->connect( + local_host => "127.0.0.2", + host => $listensock->sockhost, + service => $listensock->sockport, + socktype => $listensock->socktype, + on_connected => sub { $sock = shift; }, + on_resolve_error => sub { die "Test died early - resolve error - $_[-1]\n"; }, + on_connect_error => sub { die "Test died early - connect error $_[0]() - $_[-1]\n"; }, + ); + + wait_for { $sock }; + + is( $sock->sockhost, "127.0.0.2", '$sock->sockhost is 127.0.0.2' ); + + $listensock->accept; # Throw it away + undef $sock; # This too +} + +# Now try on_stream event +{ + my $stream; + + $loop->connect( + host => $listensock->sockhost, + service => $listensock->sockport, + socktype => $listensock->socktype, + on_stream => sub { $stream = shift; }, + on_resolve_error => sub { die "Test died early - resolve error - $_[-1]\n"; }, + on_connect_error => sub { die "Test died early - connect error $_[0]() - $_[-1]\n"; }, + ); + + wait_for { $stream }; + + isa_ok( $stream, "IO::Async::Stream", 'on_stream $stream isa IO::Async::Stream' ); + my $sock = $stream->read_handle; + is_deeply( [ unpack_sockaddr_in $sock->peername ], + [ unpack_sockaddr_in $addr ], 'on_stream $sock->getpeername is $addr' ); + + $listensock->accept; # Throw it away +} + +my $udpsock = IO::Socket::INET->new( LocalAddr => 'localhost', Protocol => 'udp' ) or + die "Cannot create udpsock - $!"; + +{ + my $future = $loop->connect( + handle => my $given_socket = IO::Async::Socket->new, + addr => { family => "inet", socktype => "dgram", addr => $udpsock->sockname }, + ); + + isa_ok( $future, "Future", '$future for ->connect( handle socket )' ); + + wait_for { $future->is_ready }; + + my $socket = $future->get; + identical( $socket, $given_socket, '$future->get returns given Socket' ); + is_deeply( [ unpack_sockaddr_in $socket->read_handle->peername ], + [ unpack_sockaddr_in $udpsock->sockname ], 'Returned $socket->read_handle->getpeername is $addr' ); +} + +# legacy callbacks +{ + my $sock; + + $loop->connect( + addr => { family => "inet", socktype => "dgram", addr => $udpsock->sockname }, + on_socket => sub { $sock = shift; }, + on_connect_error => sub { die "Test died early - connect error $_[0]() - $_[-1]\n"; }, + ); + + wait_for { $sock }; + + isa_ok( $sock, "IO::Async::Socket", 'on_socket $sock isa IO::Async::Socket' ); + is_deeply( [ unpack_sockaddr_in $sock->read_handle->peername ], + [ unpack_sockaddr_in $udpsock->sockname ], 'on_socket $sock->read_handle->getpeername is $addr' ); +} + +SKIP: { + # Now try an address we know to be invalid - a UNIX socket that doesn't exist + + socket( my $dummy, AF_UNIX, SOCK_STREAM, 0 ) or + skip "Cannot create AF_UNIX sockets - $!", 2; + + my $error; + + my $failop; + my $failerr; + + $loop->connect( + addr => { family => "unix", socktype => "stream", path => "/some/path/we/know/breaks" }, + on_connected => sub { die "Test died early - connect succeeded\n"; }, + on_fail => sub { $failop = shift @_; $failerr = pop @_; }, + on_connect_error => sub { $error = 1 }, + ); + + wait_for { $error }; + + is( $failop, "connect", '$failop is connect' ); + is( $failerr+0, ENOENT, '$failerr is ENOENT' ); +} + +SKIP: { + socket( my $dummy, AF_UNIX, SOCK_STREAM, 0 ) or + skip "Cannot create AF_UNIX sockets - $!", 2; + + my $failop; + my $failerr; + + my $future = $loop->connect( + addr => { family => "unix", socktype => "stream", path => "/some/path/we/know/breaks" }, + on_fail => sub { $failop = shift @_; $failerr = pop @_; }, + ); + + wait_for { $future->is_ready }; + + is( $failop, "connect", '$failop is connect' ); + is( $failerr+0, ENOENT, '$failerr is ENOENT' ); + + ok( $future->is_failed, '$future failed' ); + is( ( $future->failure )[2], "connect", '$future fail op is connect' ); + is( ( $future->failure )[3]+0, ENOENT, '$future fail err is ENOENT' ); +} + +# UNIX sockets always connect(2) synchronously, meaning if they fail, the error +# is available immediately. The above has therefore not properly tested +# asynchronous connect(2) failures. INET sockets should do this. + +# First off we need a local socket that isn't listening - at lease one of the +# first 100 is likely not to be + +my $port; +my $failure; + +foreach ( 1 .. 100 ) { + IO::Socket::INET->new( PeerHost => "127.0.0.1", PeerPort => $_ ) and next; + + $failure = "$!"; + $port = $_; + + last; +} + +SKIP: { + skip "Cannot find an un-connect(2)able socket on 127.0.0.1", 2 unless defined $port; + + my $failop; + my $failerr; + + my @error; + + $loop->connect( + addr => { family => "inet", socktype => "stream", port => $port, ip => "127.0.0.1" }, + on_connected => sub { die "Test died early - connect succeeded\n"; }, + on_fail => sub { $failop = shift @_; $failerr = pop @_; }, + on_connect_error => sub { @error = @_; }, + ); + + wait_for { @error }; + + is( $failop, "connect", '$failop is connect' ); + is( "$failerr", $failure, "\$failerr is '$failure'" ); + + is( $error[0], "connect", '$error[0] is connect' ); + is( "$error[1]", $failure, "\$error[1] is '$failure'" ); +} + +done_testing; diff --git a/t/52loop-listen.t b/t/52loop-listen.t new file mode 100644 index 0000000..5453630 --- /dev/null +++ b/t/52loop-listen.t @@ -0,0 +1,183 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Identity; + +use IO::Socket::INET; + +use Socket qw( inet_ntoa unpack_sockaddr_in ); + +use IO::Async::Loop; + +# Some odd locations like BSD jails might not like INADDR_LOOPBACK. We'll +# establish a baseline first to test against +my $INADDR_LOOPBACK = do { + my $localsock = IO::Socket::INET->new( LocalAddr => "localhost", Listen => 1 ); + $localsock->sockaddr; +}; +my $INADDR_LOOPBACK_HOST = inet_ntoa( $INADDR_LOOPBACK ); +if( $INADDR_LOOPBACK ne INADDR_LOOPBACK ) { + diag( "Testing with INADDR_LOOPBACK=$INADDR_LOOPBACK_HOST; this may be because of odd networking" ); +} + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my $listensock = IO::Socket::INET->new( + LocalAddr => "localhost", + Type => SOCK_STREAM, + Listen => 1, + ) or die "Cannot socket() - $!"; + + my $newclient; + + my $f = $loop->listen( + handle => $listensock, + on_accept => sub { $newclient = $_[0]; }, + ); + + ok( $f->is_ready, '$loop->listen on handle ready synchronously' ); + + my $notifier = $f->get; + isa_ok( $notifier, "IO::Async::Notifier", 'synchronous on_notifier given a Notifier' ); + + identical( $notifier->loop, $loop, 'synchronous $notifier->loop is $loop' ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listensock->sockname ) or die "Cannot connect() - $!"; + + ok( defined $clientsock->peername, '$clientsock is connected' ); + + wait_for { defined $newclient }; + + is_deeply( [ unpack_sockaddr_in $newclient->peername ], + [ unpack_sockaddr_in $clientsock->sockname ], '$newclient peer is correct' ); +} + +{ + my $listensock; + my $newclient; + + my $f = $loop->listen( + family => "inet", + socktype => "stream", + service => "", # Ask the kernel to allocate a port for us + host => "localhost", + + on_listen => sub { $listensock = $_[0]; }, + + on_accept => sub { $newclient = $_[0]; }, + ); + + my $notifier = $f->get; + + ok( defined $listensock->fileno, '$listensock has a fileno' ); + # Not sure if it'll be an IO::Socket::INET or ::IP, but either way it should support these + can_ok( $listensock, qw( peerhost peerport ) ); + + isa_ok( $notifier, "IO::Async::Notifier", 'asynchronous on_notifier given a Notifier' ); + + identical( $notifier->loop, $loop, 'asynchronous $notifier->loop is $loop' ); + + my $listenaddr = $listensock->sockname; + + ok( defined $listenaddr, '$listensock has address' ); + + my ( $listenport, $listen_inaddr ) = unpack_sockaddr_in( $listenaddr ); + + is( inet_ntoa( $listen_inaddr ), $INADDR_LOOPBACK_HOST, '$listenaddr is INADDR_LOOPBACK' ); + + my $clientsock = IO::Socket::INET->new( Type => SOCK_STREAM ) + or die "Cannot socket() - $!"; + + $clientsock->connect( $listenaddr ) or die "Cannot connect() - $!"; + + is( (unpack_sockaddr_in( $clientsock->peername ))[0], $listenport, '$clientsock on the correct port' ); + + wait_for { defined $newclient }; + + can_ok( $newclient, qw( peerhost peerport ) ); + + is_deeply( [ unpack_sockaddr_in $newclient->peername ], + [ unpack_sockaddr_in $clientsock->sockname ], '$newclient peer is correct' ); +} + +# Now we want to test failure. It's hard to know in a test script what will +# definitely fail, but it's likely we're either running as non-root, or the +# machine has at least one of an SSH or a webserver running. In this case, +# it's likely we'll fail to bind TCP port 22 or 80. + +my $badport; +my $failure; +foreach my $port ( 22, 80 ) { + IO::Socket::INET->new( + Type => SOCK_STREAM, + LocalHost => "localhost", + LocalPort => $port, + ReuseAddr => 1, + Listen => 1, + ) and next; + + $badport = $port; + $failure = $!; + last; +} + +SKIP: { + skip "No bind()-failing ports found", 6 unless defined $badport; + + my $failop; + my $failerr; + + my @error; + + # We need to capture the Listener object before failure, so we can assert + # it gets removed from the Loop again afterwards + my $listener; + no warnings 'redefine'; + my $add = IO::Async::Loop->can( "add" ); + local *IO::Async::Loop::add = sub { + $listener = $_[1]; + $add->( @_ ); + }; + + $loop->listen( + family => "inet", + socktype => "stream", + host => "localhost", + service => $badport, + + on_resolve_error => sub { die "Test died early - resolve error $_[0]\n"; }, + + on_listen => sub { die "Test died early - listen on port $badport actually succeeded\n"; }, + + on_accept => sub { "DUMMY" }, # really hope this doesn't happen ;) + + on_fail => sub { $failop = shift; $failerr = pop; }, + on_listen_error => sub { @error = @_; }, + ); + + ok( defined $listener, 'Managed to capture listener being added to Loop' ); + + wait_for { @error }; + + is( $failop, "bind", '$failop is bind' ); + is( "$failerr", $failure, "\$failerr is '$failure'" ); + + is( $error[0], "bind", '$error[0] is bind' ); + is( "$error[1]", $failure, "\$error[1] is '$failure'" ); + + ok( defined $listener, '$listener defined after bind failure' ); + ok( !$listener->loop, '$listener not in loop after bind failure' ); +} + +done_testing; diff --git a/t/53loop-extend.t b/t/53loop-extend.t new file mode 100644 index 0000000..82e7088 --- /dev/null +++ b/t/53loop-extend.t @@ -0,0 +1,103 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Identity; + +use IO::Async::Loop; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# connect +{ + my %connectargs; + my $connect_future; + sub IO::Async::Loop::FOO_connect + { + my $self = shift; + %connectargs = @_; + + identical( $self, $loop, 'FOO_connect invocant is $loop' ); + + return $connect_future = $loop->new_future; + } + + my $sock; + my $f = $loop->connect( + extensions => [qw( FOO )], + some_param => "here", + on_connected => sub { $sock = shift }, + ); + + is( ref delete $connectargs{on_connected}, "CODE", 'FOO_connect received on_connected continuation' ); + is_deeply( \%connectargs, + { some_param => "here" }, + 'FOO_connect received some_param and no others' ); + + identical( $f, $connect_future, 'FOO_connect returns Future object' ); + + $loop->connect( + extensions => [qw( FOO BAR )], + param1 => "one", + param2 => "two", + on_connected => sub { $sock = shift }, + ); + + delete $connectargs{on_connected}; + is_deeply( \%connectargs, + { extensions => [qw( BAR )], + param1 => "one", + param2 => "two" }, + 'FOO_connect still receives other extensions' ); +} + +# listen +{ + my %listenargs; + my $listen_future; + sub IO::Async::Loop::FOO_listen + { + my $self = shift; + %listenargs = @_; + + identical( $self, $loop, 'FOO_listen invocant is $loop' ); + + return $listen_future = $loop->new_future; + } + + my $sock; + my $f = $loop->listen( + extensions => [qw( FOO )], + some_param => "here", + on_accept => sub { $sock = shift }, + ); + + isa_ok( delete $listenargs{listener}, "IO::Async::Listener", '$listenargs{listener}' ); + is_deeply( \%listenargs, + { some_param => "here" }, + 'FOO_listen received some_param and no others' ); + + identical( $f, $listen_future, 'FOO_listen returns Future object' ); + + $loop->listen( + extensions => [qw( FOO BAR )], + param1 => "one", + param2 => "two", + on_accept => sub { $sock = shift }, + ); + + delete $listenargs{listener}; + is_deeply( \%listenargs, + { extensions => [qw( BAR )], + param1 => "one", + param2 => "two" }, + 'FOO_listen still receives other extensions' ); +} + +done_testing; diff --git a/t/60protocol.t b/t/60protocol.t new file mode 100644 index 0000000..493931e --- /dev/null +++ b/t/60protocol.t @@ -0,0 +1,146 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Identity; +use Test::Refcount; + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Handle; +use IO::Async::Protocol; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + +# Need sockets in nonblocking mode +$S1->blocking( 0 ); +$S2->blocking( 0 ); + +my $handle = IO::Async::Handle->new( + handle => $S1, + on_read_ready => sub {}, + on_write_ready => sub {}, +); + +my @setup_args; +my @teardown_args; +my $readready; +my $writeready; + +my $proto = TestProtocol->new; + +ok( defined $proto, '$proto defined' ); +isa_ok( $proto, "IO::Async::Protocol", '$proto isa IO::Async::Protocol' ); + +is_oneref( $proto, '$proto has refcount 1 initially' ); + +$proto->configure( transport => $handle ); + +identical( $proto->transport, $handle, '$proto->transport' ); + +is( scalar @setup_args, 1, '@setup_args after configure transport' ); +identical( $setup_args[0], $handle, '$setup_args[0] after configure transport'); + +undef @setup_args; + +is_oneref( $proto, '$proto has refcount 1 after configure transport' ); +# lexical $handle, $proto->{transport}, $proto->{children} == 3 +is_refcount( $handle, 3, '$handle has refcount 3 after proto configure transport' ); + +$loop->add( $proto ); + +is_refcount( $proto, 2, '$proto has refcount 2 after adding to Loop' ); +is_refcount( $handle, 4, '$handle has refcount 4 after adding proto to Loop' ); + +$S2->syswrite( "hello\n" ); + +wait_for { $readready }; + +is( $readready, 1, '$readready after wait' ); + +# Just to shut poll/select/etc... up +$S1->sysread( my $dummy, 8192 ); + +my $newhandle = IO::Async::Handle->new( + handle => $S1, + on_read_ready => sub {}, + on_write_ready => sub {}, +); + +$proto->configure( transport => $newhandle ); + +identical( $proto->transport, $newhandle, '$proto->transport after reconfigure' ); + +is( scalar @teardown_args, 1, '@teardown_args after reconfigure transport' ); +identical( $teardown_args[0], $handle, '$teardown_args[0] after reconfigure transport'); + +is( scalar @setup_args, 1, '@setup_args after reconfigure transport' ); +identical( $setup_args[0], $newhandle, '$setup_args[0] after reconfigure transport'); + +undef @teardown_args; +undef @setup_args; + +is_oneref( $handle, '$handle has refcount 1 after reconfigure' ); + +my $closed = 0; +$proto->configure( + on_closed => sub { $closed++ }, +); + +$proto->transport->close; + +wait_for { $closed }; + +is( $closed, 1, '$closed after stream close' ); + +is( $proto->transport, undef, '$proto->transport is undef after close' ); + +is_refcount( $proto, 2, '$proto has refcount 2 before removal from Loop' ); + +$loop->remove( $proto ); + +is_oneref( $proto, '$proto has refcount 1 before EOF' ); + +done_testing; + +package TestProtocol; +use base qw( IO::Async::Protocol ); + +sub setup_transport +{ + my $self = shift; + @setup_args = @_; + + my ( $transport ) = @_; + + $self->SUPER::setup_transport( $transport ); + + $transport->configure( + on_read_ready => sub { $readready = 1 }, + on_write_ready => sub { $writeready = 1 }, + ); +} + +sub teardown_transport +{ + my $self = shift; + @teardown_args = @_; + + my ( $transport ) = @_; + $transport->configure( + on_read_ready => sub {}, + on_write_ready => sub {}, + ); + + $self->SUPER::teardown_transport( $transport ); +} diff --git a/t/61protocol-stream.t b/t/61protocol-stream.t new file mode 100644 index 0000000..da5832a --- /dev/null +++ b/t/61protocol-stream.t @@ -0,0 +1,245 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Refcount; + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Stream; +use IO::Async::Protocol::Stream; + +use IO::Socket::INET; +use Socket qw( SOCK_STREAM ); + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + +# Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + my @lines; + + my $streamproto = IO::Async::Protocol::Stream->new( + transport => IO::Async::Stream->new( handle => $S1 ), + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + ok( defined $streamproto, '$streamproto defined' ); + isa_ok( $streamproto, "IO::Async::Protocol::Stream", '$streamproto isa IO::Async::Protocol::Stream' ); + + is_oneref( $streamproto, '$streamproto has refcount 1 initially' ); + + $loop->add( $streamproto ); + + is_refcount( $streamproto, 2, '$streamproto has refcount 2 after adding to Loop' ); + + $S2->syswrite( "message\n" ); + + is_deeply( \@lines, [], '@lines before wait' ); + + wait_for { scalar @lines }; + + is_deeply( \@lines, [ "message\n" ], '@lines after wait' ); + + undef @lines; + my @new_lines; + $streamproto->configure( + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @new_lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + }, + ); + + $S2->syswrite( "new\nlines\n" ); + + wait_for { scalar @new_lines }; + + is( scalar @lines, 0, '@lines still empty after on_read replace' ); + is_deeply( \@new_lines, [ "new\n", "lines\n" ], '@new_lines after on_read replace' ); + + $streamproto->write( "response\n" ); + + my $response = ""; + wait_for_stream { $response =~ m/\n/ } $S2 => $response; + + is( $response, "response\n", 'response written by protocol' ); + + my $done; + my $flushed; + + $streamproto->write( + sub { + is( $_[0], $streamproto, 'writersub $_[0] is $streamproto' ); + return $done++ ? undef : "a lazy message\n"; + }, + on_flush => sub { + is( $_[0], $streamproto, 'on_flush $_[0] is $streamproto' ); + $flushed = 1; + }, + ); + + wait_for { $flushed }; + + $response = ""; + wait_for_stream { $response =~ m/\n/ } $S2 => $response; + + is( $response, "a lazy message\n", 'response written by protocol writersub' ); + + my $closed = 0; + $streamproto->configure( + on_closed => sub { $closed++ }, + ); + + $S2->close; + + wait_for { $closed }; + + is( $closed, 1, '$closed after stream close' ); + + is_refcount( $streamproto, 2, '$streamproto has refcount 2 before removing from Loop' ); + + $loop->remove( $streamproto ); + + is_oneref( $streamproto, '$streamproto refcount 1 finally' ); +} + +my @sub_lines; + +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + + # Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + my $streamproto = TestProtocol::Stream->new( + transport => IO::Async::Stream->new( handle => $S1 ), + ); + + ok( defined $streamproto, 'subclass $streamproto defined' ); + isa_ok( $streamproto, "IO::Async::Protocol::Stream", '$streamproto isa IO::Async::Protocol::Stream' ); + + is_oneref( $streamproto, 'subclass $streamproto has refcount 1 initially' ); + + $loop->add( $streamproto ); + + is_refcount( $streamproto, 2, 'subclass $streamproto has refcount 2 after adding to Loop' ); + + $S2->syswrite( "message\n" ); + + is_deeply( \@sub_lines, [], '@sub_lines before wait' ); + + wait_for { scalar @sub_lines }; + + is_deeply( \@sub_lines, [ "message\n" ], '@sub_lines after wait' ); + + $loop->remove( $streamproto ); +} + +{ + my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + + # Need sockets in nonblocking mode + $S1->blocking( 0 ); + $S2->blocking( 0 ); + + my $serversock = IO::Socket::INET->new( + Type => SOCK_STREAM, + LocalHost => "localhost", + LocalPort => 0, + Listen => 1, + ) or die "Cannot create server socket - $!"; + + my @lines; + my $streamproto = IO::Async::Protocol::Stream->new( + on_read => sub { + my $self = shift; + my ( $buffref, $eof ) = @_; + push @lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; + } + ); + + $loop->add( $streamproto ); + + my $connected = 0; + + $streamproto->connect( + host => $serversock->sockhost, + service => $serversock->sockport, + family => $serversock->sockdomain, + + on_connected => sub { $connected++ }, + + on_connect_error => sub { die "Test failed early - $_[-1]" }, + on_resolve_error => sub { die "Test failed early - $_[-1]" }, + ); + + wait_for { $connected }; + + my $clientsock = $serversock->accept; + + is( $streamproto->transport->read_handle->peerport, + $serversock->sockport, + 'Protocol is connected to server socket port' ); + + $clientsock->syswrite( "A message\n" ); + + undef @lines; + + wait_for { @lines }; + + is( $lines[0], "A message\n", 'Protocol transport works' ); +} + +{ + my $read_eof; + my $write_eof; + my $streamproto = IO::Async::Protocol::Stream->new( + on_read_eof => sub { $read_eof++ }, + on_write_eof => sub { $write_eof++ }, + ); + + $streamproto->configure( transport => my $stream = IO::Async::Stream->new ); + + $stream->invoke_event( on_read_eof => ); + is( $read_eof, 1, '$read_eof after on_read_eof' ); + + $stream->invoke_event( on_write_eof => ); + is( $write_eof, 1, '$write_eof after on_write_eof' ); +} + +done_testing; + +package TestProtocol::Stream; +use base qw( IO::Async::Protocol::Stream ); + +sub on_read +{ + my $self = shift; + my ( $buffref, $eof ) = @_; + + push @sub_lines, $1 while $$buffref =~ s/^(.*\n)//; + return 0; +} diff --git a/t/62protocol-linestream.t b/t/62protocol-linestream.t new file mode 100644 index 0000000..2acab97 --- /dev/null +++ b/t/62protocol-linestream.t @@ -0,0 +1,118 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; +use Test::Refcount; + +use IO::Async::Loop; + +use IO::Async::OS; + +use IO::Async::Protocol::LineStream; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +my ( $S1, $S2 ) = IO::Async::OS->socketpair or die "Cannot create socket pair - $!"; + +# Need sockets in nonblocking mode +$S1->blocking( 0 ); +$S2->blocking( 0 ); + +my @lines; + +my $linestreamproto = IO::Async::Protocol::LineStream->new( + handle => $S1, + on_read_line => sub { + my $self = shift; + + push @lines, $_[0]; + }, +); + +ok( defined $linestreamproto, '$linestreamproto defined' ); +isa_ok( $linestreamproto, "IO::Async::Protocol::LineStream", '$linestreamproto isa IO::Async::Protocol::LineStream' ); + +is_oneref( $linestreamproto, '$linestreamproto has refcount 1 initially' ); + +$loop->add( $linestreamproto ); + +is_refcount( $linestreamproto, 2, '$linestreamproto has refcount 2 after adding to Loop' ); + +$S2->syswrite( "message\r\n" ); + +is_deeply( \@lines, [], '@lines before wait' ); + +wait_for { scalar @lines }; + +is_deeply( \@lines, [ "message" ], '@lines after wait' ); + +undef @lines; +my @new_lines; +$linestreamproto->configure( + on_read_line => sub { + my $self = shift; + + push @new_lines, $_[0]; + }, +); + +$S2->syswrite( "new\r\nlines\r\n" ); + +wait_for { scalar @new_lines }; + +is( scalar @lines, 0, '@lines still empty after on_read replace' ); +is_deeply( \@new_lines, [ "new", "lines" ], '@new_lines after on_read replace' ); + +$linestreamproto->write_line( "response" ); + +my $response = ""; +wait_for_stream { $response =~ m/\r\n/ } $S2 => $response; + +is( $response, "response\r\n", 'response written by protocol' ); + +my @sub_lines; + +$linestreamproto = TestProtocol::Stream->new( + handle => $S1, +); + +ok( defined $linestreamproto, 'subclass $linestreamproto defined' ); +isa_ok( $linestreamproto, "IO::Async::Protocol::LineStream", '$linestreamproto isa IO::Async::Protocol::LineStream' ); + +is_oneref( $linestreamproto, 'subclass $linestreamproto has refcount 1 initially' ); + +$loop->add( $linestreamproto ); + +is_refcount( $linestreamproto, 2, 'subclass $linestreamproto has refcount 2 after adding to Loop' ); + +$S2->syswrite( "message\r\n" ); + +is_deeply( \@sub_lines, [], '@sub_lines before wait' ); + +wait_for { scalar @sub_lines }; + +is_deeply( \@sub_lines, [ "message" ], '@sub_lines after wait' ); + +undef @lines; + +$loop->remove( $linestreamproto ); + +undef $linestreamproto; + +done_testing; + +package TestProtocol::Stream; +use base qw( IO::Async::Protocol::LineStream ); + +sub on_read_line +{ + my $self = shift; + + push @sub_lines, $_[0]; +} diff --git a/t/63handle-connect.t b/t/63handle-connect.t new file mode 100644 index 0000000..1318923 --- /dev/null +++ b/t/63handle-connect.t @@ -0,0 +1,84 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; + +use IO::Async::Loop; + +use IO::Async::Handle; + +use IO::Async::OS; + +use IO::Socket::INET; +use Socket qw( SOCK_STREAM ); + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# Try connect(2)ing to a socket we've just created +my $listensock = IO::Socket::INET->new( + Type => SOCK_STREAM, + LocalAddr => 'localhost', + LocalPort => 0, + Listen => 1 +) or die "Cannot create listensock - $!"; + +my $addr = $listensock->sockname; + +# ->connect to plain addr +{ + my $handle = IO::Async::Handle->new( + on_read_ready => sub {}, + on_write_ready => sub {}, + ); + + $loop->add( $handle ); + + my $f = $handle->connect( addr => [ 'inet', 'stream', 0, $addr ] ); + + ok( defined $f, '$handle->connect Future defined' ); + + wait_for { $f->is_ready }; + $f->is_failed and $f->get; + + ok( defined $handle->read_handle, '$handle->read_handle defined after ->connect addr' ); + is( $handle->read_handle->peerport, $listensock->sockport, '$handle->read_handle->peerport after ->connect addr' ); + + $listensock->accept; # drop it + + $loop->remove( $handle ); +} + +# ->connect to host/service +{ + my $handle = IO::Async::Handle->new( + on_read_ready => sub {}, + on_write_ready => sub {}, + ); + + $loop->add( $handle ); + + my $f = $handle->connect( + family => "inet", + socktype => "stream", + host => $listensock->sockhost, + service => $listensock->sockport, + ); + + wait_for { $f->is_ready }; + $f->is_failed and $f->get; + + ok( defined $handle->read_handle, '$handle->read_handle defined after ->connect host/service' ); + is( $handle->read_handle->peerport, $listensock->sockport, '$handle->read_handle->peerport after ->connect host/service' ); + + $listensock->accept; # drop it + + $loop->remove( $handle ); +} + +done_testing; diff --git a/t/64handle-bind.t b/t/64handle-bind.t new file mode 100644 index 0000000..772f9e6 --- /dev/null +++ b/t/64handle-bind.t @@ -0,0 +1,36 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use IO::Async::Test; + +use Test::More; + +use IO::Async::Loop; + +use IO::Async::Handle; + +my $loop = IO::Async::Loop->new_builtin; + +testing_loop( $loop ); + +# ->bind a UDP service +{ + my $recv_count; + + my $receiver = IO::Async::Handle->new( + on_read_ready => sub { $recv_count++ }, + on_write_ready => sub { }, + ); + $loop->add( $receiver ); + + $receiver->bind( + service => "0", + socktype => "dgram", + )->get; + + ok( $receiver->read_handle->sockport, '$receiver bound to a read handle' ); +} + +done_testing; diff --git a/t/99pod.t b/t/99pod.t new file mode 100644 index 0000000..eb319fb --- /dev/null +++ b/t/99pod.t @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More; + +eval "use Test::Pod 1.00"; +plan skip_all => "Test::Pod 1.00 required for testing POD" if $@; + +all_pod_files_ok(); diff --git a/t/StupidLoop.pm b/t/StupidLoop.pm new file mode 100644 index 0000000..0b4fc9a --- /dev/null +++ b/t/StupidLoop.pm @@ -0,0 +1,8 @@ +package t::StupidLoop; + +use strict; +use base qw( IO::Async::Loop ); + +sub new { return bless {}, shift; } + +1; diff --git a/t/TimeAbout.pm b/t/TimeAbout.pm new file mode 100644 index 0000000..86b3d1f --- /dev/null +++ b/t/TimeAbout.pm @@ -0,0 +1,31 @@ +package t::TimeAbout; + +use Test::More; +use Time::HiRes qw( time ); + +use constant AUT => $ENV{TEST_QUICK_TIMERS} ? 0.1 : 1; + +use Exporter 'import'; +our @EXPORT = qw( time_about ); + +# Kindof like Test::Timer only we use Time::HiRes +# We'll be quite lenient on the time taken, in case of heavy test machine load +sub time_about +{ + my ( $code, $target, $name ) = @_; + + my $lower = $target*0.75; + my $upper = $target*1.5 + 1; + + my $now = time; + $code->(); + my $took = (time - $now) / AUT; + + cmp_ok( $took, '>', $lower, "$name took at least $lower" ); + cmp_ok( $took, '<', $upper * 3, "$name took no more than $upper" ); + if( $took > $upper and $took <= $upper * 3 ) { + diag( "$name took longer than $upper - this may just be an indication of a busy testing machine rather than a bug" ); + } +} + +0x55AA; |