Acme-Sort-Sleep

 view release on metacpan or  search on metacpan

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

 use IO::Async::Loop;
 use IO::Async::Loop::Poll;

 my $loop_poll = IO::Async::Loop::Poll->new;

 my $loop = IO::Async::Loop->new;

While it is not advised to do so under normal circumstances, if the program
really wishes to construct more than one Loop object, it can call the
constructor C<really_new>, or invoke one of the subclass-specific constructors
directly.

The list of candidates is formed from the following choices, in this order:

=over 4

=item * $ENV{IO_ASYNC_LOOP}

If this environment variable is set, it should contain a comma-separated list
of subclass names. These names may or may not be fully-qualified; if a name
does not contain C<::> then it will have C<IO::Async::Loop::> prepended to it.
This allows the end-user to specify a particular choice to fit the needs of
his use of a program using L<IO::Async>.

=item * $IO::Async::Loop::LOOP

If this scalar is set, it should contain a comma-separated list of subclass
names. These may or may not be fully-qualified, as with the above case. This
allows a program author to suggest a loop module to use.

In cases where the module subclass is a hard requirement, such as GTK programs
using C<Glib>, it would be better to use the module specifically and invoke
its constructor directly.

=item * IO::Async::OS->LOOP_PREFER_CLASSES

The L<IO::Async::OS> hints module for the given OS is then consulted to see if
it suggests any other module classes specific to the given operating system.

=item * $^O

The module called C<IO::Async::Loop::$^O> is tried next. This allows specific
OSes, such as the ever-tricky C<MSWin32>, to provide an implementation that
might be more efficient than the generic ones, or even work at all.

This option is now discouraged in favour of the L<IO::Async::OS> hint instead.
At some future point it may be removed entirely, given as currently only
C<linux> uses it.

=item * Poll and Select

Finally, if no other choice has been made by now, the built-in C<Poll> module
is chosen. This should always work, but in case it doesn't, the C<Select>
module will be chosen afterwards as a last-case attempt. If this also fails,
then the magic constructor itself will throw an exception.

=back

If any of the explicitly-requested loop types (C<$ENV{IO_ASYNC_LOOP}> or
C<$IO::Async::Loop::LOOP>) fails to load then a warning is printed detailing
the error.

Implementors of new C<IO::Async::Loop> subclasses should see the notes about
C<API_VERSION> below.

=cut

sub __try_new
{
   my ( $class ) = @_;

   ( my $file = "$class.pm" ) =~ s{::}{/}g;

   eval {
      local $SIG{__WARN__} = sub {};
      require $file;
   } or return;

   my $self;
   $self = eval { $class->new } and return $self;

   # Oh dear. We've loaded the code OK but for some reason the constructor
   # wasn't happy. Being polite we ought really to unload the file again,
   # but perl doesn't actually provide us a way to do this.

   return undef;
}

sub new
{
   return our $ONE_TRUE_LOOP ||= shift->really_new;
}

# Ensure that the loop is DESTROYed recursively at exit time, before GD happens
END {
   undef our $ONE_TRUE_LOOP;
}

sub really_new
{
   shift;  # We're going to ignore the class name actually given
   my $self;

   my @candidates;

   push @candidates, split( m/,/, $ENV{IO_ASYNC_LOOP} ) if defined $ENV{IO_ASYNC_LOOP};

   push @candidates, split( m/,/, $LOOP ) if defined $LOOP;

   foreach my $class ( @candidates ) {
      $class =~ m/::/ or $class = "IO::Async::Loop::$class";
      $self = __try_new( $class ) and return $self;

      my ( $topline ) = split m/\n/, $@; # Ignore all the other lines; they'll be require's verbose output
      warn "Unable to use $class - $topline\n";
   }

   unless( $LOOP_NO_OS ) {
      foreach my $class ( IO::Async::OS->LOOP_PREFER_CLASSES, "IO::Async::Loop::$^O" ) {
         $class =~ m/::/ or $class = "IO::Async::Loop::$class";
         $self = __try_new( $class ) and return $self;

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

sub later
{
   my $self = shift;
   my ( $code ) = @_;

   return $self->watch_idle( when => 'later', code => $code );
}

=head2 spawn_child

   $loop->spawn_child( %params )

This method creates a new child process to run a given code block or command.
For more detail, see the C<spawn_child> method on the
L<IO::Async::ChildManager> class.

=cut

sub spawn_child
{
   my $self = shift;
   my %params = @_;

   my $childmanager = $self->{childmanager} ||=
      $self->__new_feature( "IO::Async::ChildManager" );

   $childmanager->spawn_child( %params );
}

=head2 open_child

   $pid = $loop->open_child( %params )

This creates a new child process to run the given code block or command, and
attaches filehandles to it that the parent will watch. This method is a light
wrapper around constructing a new L<IO::Async::Process> object, provided
largely for backward compatibility. New code ought to construct such an object
directly, as it may provide more features than are available here.

The C<%params> hash takes the following keys:

=over 8

=item command => ARRAY or STRING

=item code => CODE

The command or code to run in the child process (as per the C<spawn> method)

=item on_finish => CODE

A continuation to be called when the child process exits and has closed all of
the filehandles that were set up for it. It will be invoked in the following
way:

 $on_finish->( $pid, $exitcode )

The second argument is passed the plain perl C<$?> value.

=item on_error => CODE

Optional continuation to be called when the child code block throws an
exception, or the command could not be C<exec(2)>ed. It will be invoked in the
following way (as per C<spawn>)

 $on_error->( $pid, $exitcode, $dollarbang, $dollarat )

If this continuation is not supplied, then C<on_finish> is used instead. The
value of C<$!> and C<$@> will not be reported.

=item setup => ARRAY

Optional reference to an array to pass to the underlying C<spawn> method.

=back

In addition, the hash takes keys that define how to set up file descriptors in
the child process. (If the C<setup> array is also given, these operations will
be performed after those specified by C<setup>.)

=over 8

=item fdI<n> => HASH

A hash describing how to set up file descriptor I<n>. The hash may contain one
of the following sets of keys:

=over 4

=item on_read => CODE

The child will be given the writing end of a pipe. The reading end will be
wrapped by an L<IO::Async::Stream> using this C<on_read> callback function.

=item from => STRING

The child will be given the reading end of a pipe. The string given by the
C<from> parameter will be written to the child. When all of the data has been
written the pipe will be closed.

=back

=item stdin => ...

=item stdout => ...

=item stderr => ...

Shortcuts for C<fd0>, C<fd1> and C<fd2> respectively.

=back

=cut

sub open_child
{
   my $self = shift;
   my %params = @_;

   my $on_finish = delete $params{on_finish};
   ref $on_finish or croak "Expected 'on_finish' to be a reference";
   $params{on_finish} = sub {
      my ( $process, $exitcode ) = @_;
      $on_finish->( $process->pid, $exitcode );
   };

   if( my $on_error = delete $params{on_error} ) {
      ref $on_error or croak "Expected 'on_error' to be a reference";

      $params{on_exception} = sub {
         my ( $process, $exception, $errno, $exitcode ) = @_;
         # Swap order
         $on_error->( $process->pid, $exitcode, $errno, $exception );
      };
   }

   $params{on_exit} and croak "Cannot pass 'on_exit' parameter through ChildManager->open";

   require IO::Async::Process;
   my $process = IO::Async::Process->new( %params );

   $self->add( $process );

   return $process->pid;
}

=head2 run_child

   $pid = $loop->run_child( %params )

This creates a new child process to run the given code block or command,
capturing its STDOUT and STDERR streams. When the process exits, a
continuation is invoked being passed the exitcode, and content of the streams.

=over 8

=item command => ARRAY or STRING

=item code => CODE

The command or code to run in the child process (as per the C<spawn_child>
method)

=item on_finish => CODE

A continuation to be called when the child process exits and closed its STDOUT
and STDERR streams. It will be invoked in the following way:

 $on_finish->( $pid, $exitcode, $stdout, $stderr )

The second argument is passed the plain perl C<$?> value.

=item stdin => STRING

Optional. String to pass in to the child process's STDIN stream.

=item setup => ARRAY

Optional reference to an array to pass to the underlying C<spawn> method.

=back

This method is intended mainly as an IO::Async-compatible replacement for the
perl C<readpipe> function (`backticks`), allowing it to replace

  my $output = `command here`;

with

 $loop->run_child(
    command => "command here", 
    on_finish => sub {
       my ( undef, $exitcode, $output ) = @_;

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

=head2 set_resolver

   $loop->set_resolver( $resolver )

Sets the internally-stored L<IO::Async::Resolver> object. In most cases this
method should not be required, but it may be used to provide an alternative
resolver for special use-cases.

=cut

sub set_resolver
{
   my $self = shift;
   my ( $resolver ) = @_;

   $resolver->can( $_ ) or croak "Resolver is unsuitable as it does not implement $_"
      for qw( resolve getaddrinfo getnameinfo );

   $self->{resolver} = $resolver;

   $self->add( $resolver );
}

=head2 resolve

   @result = $loop->resolve( %params )->get

This method performs a single name resolution operation. It uses an
internally-stored L<IO::Async::Resolver> object. For more detail, see the
C<resolve> method on the L<IO::Async::Resolver> class.

=cut

sub resolve
{
   my $self = shift;
   my ( %params ) = @_;

   $self->resolver->resolve( %params );
}

=head2 connect

   $handle|$socket = $loop->connect( %params )->get

This method performs a non-blocking connection to a given address or set of
addresses, returning a L<IO::Async::Future> which represents the operation. On
completion, the future will yield the connected socket handle, or the given
L<IO::Async::Handle> object.

There are two modes of operation. Firstly, a list of addresses can be provided
which will be tried in turn. Alternatively as a convenience, if a host and
service name are provided instead of a list of addresses, these will be
resolved using the underlying loop's C<resolve> method into the list of
addresses.

When attempting to connect to any among a list of addresses, there may be
failures among the first attempts, before a valid connection is made. For
example, the resolver may have returned some IPv6 addresses, but only IPv4
routes are valid on the system. In this case, the first C<connect(2)> syscall
will fail. This isn't yet a fatal error, if there are more addresses to try,
perhaps some IPv4 ones.

For this reason, it is possible that the operation eventually succeeds even
though some system calls initially fail. To be aware of individual failures,
the optional C<on_fail> callback can be used. This will be invoked on each
individual C<socket(2)> or C<connect(2)> failure, which may be useful for
debugging or logging.

Because this module simply uses the C<getaddrinfo> resolver, it will be fully
IPv6-aware if the underlying platform's resolver is. This allows programs to
be fully IPv6-capable.

In plain address mode, the C<%params> hash takes the following keys:

=over 8

=item addrs => ARRAY

Reference to an array of (possibly-multiple) address structures to attempt to
connect to. Each should be in the layout described for C<addr>. Such a layout
is returned by the C<getaddrinfo> named resolver.

=item addr => HASH or ARRAY

Shortcut for passing a single address to connect to; it may be passed directly
with this key, instead of in another array on its own. This should be in a
format recognised by L<IO::Async::OS>'s C<extract_addrinfo> method.

This example shows how to use the C<Socket> functions to construct one for TCP
port 8001 on address 10.0.0.1:

 $loop->connect(
    addr => {
       family   => "inet",
       socktype => "stream",
       port     => 8001,
       ip       => "10.0.0.1",
    },
    ...
 );

This example shows another way to connect to a UNIX socket at F<echo.sock>.

 $loop->connect(
    addr => {
       family   => "unix",
       socktype => "stream",
       path     => "echo.sock",
    },
    ...
 );

=item local_addrs => ARRAY

=item local_addr => HASH or ARRAY

Optional. Similar to the C<addrs> or C<addr> parameters, these specify a local
address or set of addresses to C<bind(2)> the socket to before
C<connect(2)>ing it.

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

When performing the resolution step too, the C<addrs> or C<addr> keys are
ignored, and instead the following keys are taken:

=over 8

=item host => STRING

=item service => STRING

The hostname and service name to connect to.

=item local_host => STRING

=item local_service => STRING

Optional. The hostname and/or service name to C<bind(2)> the socket to locally
before connecting to the peer.

=item family => INT

=item socktype => INT

=item protocol => INT

=item flags => INT

Optional. Other arguments to pass along with C<host> and C<service> to the
C<getaddrinfo> call.

=item socktype => STRING

Optionally may instead be one of the values C<'stream'>, C<'dgram'> or
C<'raw'> to stand for C<SOCK_STREAM>, C<SOCK_DGRAM> or C<SOCK_RAW>. This
utility is provided to allow the caller to avoid a separate C<use Socket> only
for importing these constants.

=back

It is necessary to pass the C<socktype> hint to the resolver when resolving
the host/service names into an address, as some OS's C<getaddrinfo> functions
require this hint. A warning is emitted if neither C<socktype> nor C<protocol>
hint is defined when performing a C<getaddrinfo> lookup. To avoid this warning
while still specifying no particular C<socktype> hint (perhaps to invoke some
OS-specific behaviour), pass C<0> as the C<socktype> value.

In either case, it also accepts the following arguments:

=over 8

=item handle => IO::Async::Handle

Optional. If given a L<IO::Async::Handle> object or a subclass (such as
L<IO::Async::Stream> or L<IO::Async::Socket> its handle will be set to the
newly-connected socket on success, and that handle used as the result of the
future instead.

=item on_fail => CODE

Optional. After an individual C<socket(2)> or C<connect(2)> syscall has failed,
this callback is invoked to inform of the error. It is passed the name of the
syscall that failed, the arguments that were passed to it, and the error it
generated. I.e.

 $on_fail->( "socket", $family, $socktype, $protocol, $! );

 $on_fail->( "bind", $sock, $address, $! );

 $on_fail->( "connect", $sock, $address, $! );

Because of the "try all" nature when given a list of multiple addresses, this
callback may be invoked multiple times, even before an eventual success.

=back

This method accepts an C<extensions> parameter; see the C<EXTENSIONS> section
below.

=head2 connect (void)

   $loop->connect( %params )

When not returning a future, additional parameters can be given containing the
continuations to invoke on success or failure.

=over 8

=item on_connected => CODE

A continuation that is invoked on a successful C<connect(2)> call to a valid
socket. It will be passed the connected socket handle, as an C<IO::Socket>
object.

 $on_connected->( $handle )

=item on_stream => CODE

An alternative to C<on_connected>, a continuation that is passed an instance
of L<IO::Async::Stream> when the socket is connected. This is provided as a
convenience for the common case that a Stream object is required as the
transport for a Protocol object.

 $on_stream->( $stream )

=item on_socket => CODE

Similar to C<on_stream>, but constructs an instance of L<IO::Async::Socket>.
This is most useful for C<SOCK_DGRAM> or C<SOCK_RAW> sockets.

 $on_socket->( $socket )

=item on_connect_error => CODE

A continuation that is invoked after all of the addresses have been tried, and
none of them succeeded. It will be passed the most significant error that
occurred, and the name of the operation it occurred in. Errors from the
C<connect(2)> syscall are considered most significant, then C<bind(2)>, then
finally C<socket(2)>.

 $on_connect_error->( $syscall, $! )

=item on_resolve_error => CODE

A continuation that is invoked when the name resolution attempt fails. This is
invoked in the same way as the C<on_error> continuation for the C<resolve>
method.

=back

=cut

sub connect
{
   my $self = shift;
   my ( %params ) = @_;

   my $extensions;
   if( $extensions = delete $params{extensions} and @$extensions ) {
      my ( $ext, @others ) = @$extensions;

      my $method = "${ext}_connect";
      # TODO: Try to 'require IO::Async::$ext'

      $self->can( $method ) or croak "Extension method '$method' is not available";

      return $self->$method(
         %params,
         ( @others ? ( extensions => \@others ) : () ),
      );
   }

   my $handle = $params{handle};

   my $on_done;
   # Legacy callbacks
   if( my $on_connected = delete $params{on_connected} ) {
      $on_done = $on_connected;
   }
   elsif( my $on_stream = delete $params{on_stream} ) {
      defined $handle and croak "Cannot pass 'on_stream' with a handle object as well";

      require IO::Async::Stream;
      # TODO: It doesn't make sense to put a SOCK_DGRAM in an
      # IO::Async::Stream but currently we don't detect this
      $handle = IO::Async::Stream->new;
      $on_done = $on_stream;
   }
   elsif( my $on_socket = delete $params{on_socket} ) {
      defined $handle and croak "Cannot pass 'on_socket' with a handle object as well";

      require IO::Async::Socket;
      $handle = IO::Async::Socket->new;
      $on_done = $on_socket;
   }
   elsif( !defined wantarray ) {
      croak "Expected 'on_connected' or 'on_stream' callback or to return a Future";
   }

   my $on_connect_error;
   if( $on_connect_error = $params{on_connect_error} ) {
      # OK
   }
   elsif( !defined wantarray ) {
      croak "Expected 'on_connect_error' callback";
   }

   my $on_resolve_error;
   if( $on_resolve_error = $params{on_resolve_error} ) {
      # OK
   }
   elsif( !defined wantarray and exists $params{host} || exists $params{local_host} ) {
      croak "Expected 'on_resolve_error' callback or to return a Future";
   }

   my $connector = $self->{connector} ||= $self->__new_feature( "IO::Async::Internals::Connector" );

   my $future = $connector->connect( %params );

   $future = $future->then( sub {
      $handle->set_handle( shift );
      return Future->done( $handle )
   }) if $handle;

   $future->on_done( $on_done ) if $on_done;
   $future->on_fail( sub {
      $on_connect_error->( @_[2,3] ) if $on_connect_error and $_[1] eq "connect";
      $on_resolve_error->( $_[2] )   if $on_resolve_error and $_[1] eq "resolve";
   } );

   return $future if defined wantarray;

   # Caller is not going to keep hold of the Future, so we have to ensure it
   # stays alive somehow
   $future->on_ready( sub { undef $future } ); # intentional cycle
}

=head2 listen

   $listener = $loop->listen( %params )->get

This method sets up a listening socket and arranges for an acceptor callback
to be invoked each time a new connection is accepted on the socket. Internally
it creates an instance of L<IO::Async::Listener> and adds it to the Loop if
not given one in the arguments.

Addresses may be given directly, or they may be looked up using the system's
name resolver, or a socket handle may be given directly.

If multiple addresses are given, or resolved from the service and hostname,
then each will be attempted in turn until one succeeds.

In named resolver mode, the C<%params> hash takes the following keys:

=over 8

=item service => STRING

The service name to listen on.

=item host => STRING

The hostname to listen on. Optional. Will listen on all addresses if not
supplied.

=item family => INT

=item socktype => INT

=item protocol => INT

=item flags => INT

Optional. Other arguments to pass along with C<host> and C<service> to the
C<getaddrinfo> call.

=item socktype => STRING

Optionally may instead be one of the values C<'stream'>, C<'dgram'> or
C<'raw'> to stand for C<SOCK_STREAM>, C<SOCK_DGRAM> or C<SOCK_RAW>. This
utility is provided to allow the caller to avoid a separate C<use Socket> only
for importing these constants.

=back

It is necessary to pass the C<socktype> hint to the resolver when resolving
the host/service names into an address, as some OS's C<getaddrinfo> functions
require this hint. A warning is emitted if neither C<socktype> nor C<protocol>
hint is defined when performing a C<getaddrinfo> lookup. To avoid this warning
while still specifying no particular C<socktype> hint (perhaps to invoke some
OS-specific behaviour), pass C<0> as the C<socktype> value.

In plain address mode, the C<%params> hash takes the following keys:

=over 8

=item addrs => ARRAY

Reference to an array of (possibly-multiple) address structures to attempt to
listen on. Each should be in the layout described for C<addr>. Such a layout
is returned by the C<getaddrinfo> named resolver.

=item addr => ARRAY

Shortcut for passing a single address to listen on; it may be passed directly
with this key, instead of in another array of its own. This should be in a
format recognised by L<IO::Async::OS>'s C<extract_addrinfo> method. See also
the C<EXAMPLES> section.

=back

In direct socket handle mode, the following keys are taken:

=over 8

=item handle => IO

The listening socket handle.

=back

In either case, the following keys are also taken:

=over 8

=item on_fail => CODE

Optional. A callback that is invoked if a syscall fails while attempting to
create a listening sockets. It is passed the name of the syscall that failed,
the arguments that were passed to it, and the error generated. I.e.

 $on_fail->( "socket", $family, $socktype, $protocol, $! );

 $on_fail->( "sockopt", $sock, $optname, $optval, $! );

 $on_fail->( "bind", $sock, $address, $! );

 $on_fail->( "listen", $sock, $queuesize, $! );

=item queuesize => INT

Optional. The queue size to pass to the C<listen(2)> calls. If not supplied,
then 3 will be given instead.

=item reuseaddr => BOOL

Optional. If true or not supplied then the C<SO_REUSEADDR> socket option will
be set. To prevent this, pass a false value such as 0.

=item v6only => BOOL

Optional. If defined, sets or clears the C<IPV6_V6ONLY> socket option on
C<PF_INET6> sockets. This option disables the ability of C<PF_INET6> socket to
accept connections from C<AF_INET> addresses. Not all operating systems allow
this option to be disabled.

=back

An alternative which gives more control over the listener, is to create the
L<IO::Async::Listener> object directly and add it explicitly to the Loop.

This method accepts an C<extensions> parameter; see the C<EXTENSIONS> section
below.

=head2 listen (void)

   $loop->listen( %params )

When not returning a future, additional parameters can be given containing the
continuations to invoke on success or failure.

=over 8

=item on_notifier => CODE

Optional. A callback that is invoked when the Listener object is ready to
receive connections. The callback is passed the Listener object itself.

 $on_notifier->( $listener )

If this callback is required, it may instead be better to construct the
Listener object directly.

=item on_listen => CODE

Optional. A callback that is invoked when the listening socket is ready.
Typically this would be used in the name resolver case, in order to inspect
the socket's sockname address, or otherwise inspect the filehandle.

 $on_listen->( $socket )

=item on_listen_error => CODE

A continuation this is invoked after all of the addresses have been tried, and
none of them succeeded. It will be passed the most significant error that
occurred, and the name of the operation it occurred in. Errors from the
C<listen(2)> syscall are considered most significant, then C<bind(2)>, then
C<sockopt(2)>, then finally C<socket(2)>.

=item on_resolve_error => CODE

A continuation that is invoked when the name resolution attempt fails. This is
invoked in the same way as the C<on_error> continuation for the C<resolve>
method.

=back

=cut

sub listen
{
   my $self = shift;
   my ( %params ) = @_;

   my $remove_on_error;
   my $listener = $params{listener} ||= do {
      $remove_on_error++;

      require IO::Async::Listener;

      # Our wrappings of these don't want $listener
      my %listenerparams;
      for (qw( on_accept on_stream on_socket )) {
         next unless exists $params{$_};
         croak "Cannot ->listen with '$_' and 'listener'" if $params{listener};

         my $code = delete $params{$_};
         $listenerparams{$_} = sub {
            shift;
            goto &$code;
         };
      }

      my $listener = IO::Async::Listener->new( %listenerparams );
      $self->add( $listener );
      $listener
   };

   my $extensions;
   if( $extensions = delete $params{extensions} and @$extensions ) {
      my ( $ext, @others ) = @$extensions;

      # We happen to know we break older IO::Async::SSL
      if( $ext eq "SSL" and $IO::Async::SSL::VERSION < '0.12001' ) {
         croak "IO::Async::SSL version too old; need at least 0.12_001; found $IO::Async::SSL::VERSION";
      }

      my $method = "${ext}_listen";
      # TODO: Try to 'require IO::Async::$ext'

      $self->can( $method ) or croak "Extension method '$method' is not available";

      my $f = $self->$method(
         %params,
         ( @others ? ( extensions => \@others ) : () ),
      );
      $f->on_fail( sub { $self->remove( $listener ) } ) if $remove_on_error;

      return $f;
   }

   my $on_notifier = delete $params{on_notifier}; # optional

   my $on_listen_error  = delete $params{on_listen_error};
   my $on_resolve_error = delete $params{on_resolve_error};

   # Shortcut
   if( $params{addr} and not $params{addrs} ) {
      $params{addrs} = [ delete $params{addr} ];
   }

   my $f;
   if( my $handle = delete $params{handle} ) {
      $f = $self->_listen_handle( $listener, $handle, %params );
   }
   elsif( my $addrs = delete $params{addrs} ) {
      $on_listen_error or defined wantarray or
         croak "Expected 'on_listen_error' or to return a Future";
      $f = $self->_listen_addrs( $listener, $addrs, %params );
   }
   elsif( defined $params{service} ) {
      $on_listen_error or defined wantarray or
         croak "Expected 'on_listen_error' or to return a Future";
      $on_resolve_error or defined wantarray or
         croak "Expected 'on_resolve_error' or to return a Future";
      $f = $self->_listen_hostservice( $listener, delete $params{host}, delete $params{service}, %params );
   }
   else {
      croak "Expected either 'service' or 'addrs' or 'addr' arguments";
   }

   $f->on_done( $on_notifier ) if $on_notifier;
   if( my $on_listen = $params{on_listen} ) {
      $f->on_done( sub { $on_listen->( shift->read_handle ) } );
   }
   $f->on_fail( sub {
      my ( $message, $how, @rest ) = @_;
      $on_listen_error->( @rest )  if $on_listen_error  and $how eq "listen";
      $on_resolve_error->( @rest ) if $on_resolve_error and $how eq "resolve";
   });
   $f->on_fail( sub { $self->remove( $listener ) } ) if $remove_on_error;

   return $f if defined wantarray;

   # Caller is not going to keep hold of the Future, so we have to ensure it
   # stays alive somehow
   $f->on_ready( sub { undef $f } ); # intentional cycle
}

sub _listen_handle
{
   my $self = shift;
   my ( $listener, $handle, %params ) = @_;

   $listener->configure( handle => $handle );
   return $self->new_future->done( $listener );
}

sub _listen_addrs
{
   my $self = shift;
   my ( $listener, $addrs, %params ) = @_;

   my $queuesize = $params{queuesize} || 3;

   my $on_fail = $params{on_fail};
   !defined $on_fail or ref $on_fail or croak "Expected 'on_fail' to be a reference";

   my $reuseaddr = 1;
   $reuseaddr = 0 if defined $params{reuseaddr} and not $params{reuseaddr};

   my $v6only = $params{v6only};

   my ( $listenerr, $binderr, $sockopterr, $socketerr );

   foreach my $addr ( @$addrs ) {
      my ( $family, $socktype, $proto, $address ) = IO::Async::OS->extract_addrinfo( $addr );

      my $sock;

      unless( $sock = IO::Async::OS->socket( $family, $socktype, $proto ) ) {
         $socketerr = $!;
         $on_fail->( socket => $family, $socktype, $proto, $! ) if $on_fail;
         next;
      }

      if( $reuseaddr ) {
         unless( $sock->sockopt( SO_REUSEADDR, 1 ) ) {
            $sockopterr = $!;
            $on_fail->( sockopt => $sock, SO_REUSEADDR, 1, $! ) if $on_fail;
            next;
         }
      }

      if( defined $v6only and $family == AF_INET6 ) {
         unless( $sock->setsockopt( IPPROTO_IPV6, IPV6_V6ONLY, $v6only ) ) {
            $sockopterr = $!;
            $on_fail->( sockopt => $sock, IPV6_V6ONLY, $v6only, $! ) if $on_fail;
            next;
         }
      }

      unless( $sock->bind( $address ) ) {
         $binderr = $!;
         $on_fail->( bind => $sock, $address, $! ) if $on_fail;
         next;
      }

      unless( $sock->listen( $queuesize ) ) {
         $listenerr = $!;
         $on_fail->( listen => $sock, $queuesize, $! ) if $on_fail;
         next;
      }

      return $self->_listen_handle( $listener, $sock, %params );
   }

   my $f = $self->new_future;
   return $f->fail( "Cannot listen() - $listenerr",      listen => listen  => $listenerr  ) if $listenerr;
   return $f->fail( "Cannot bind() - $binderr",          listen => bind    => $binderr    ) if $binderr;
   return $f->fail( "Cannot setsockopt() - $sockopterr", listen => sockopt => $sockopterr ) if $sockopterr;
   return $f->fail( "Cannot socket() - $socketerr",      listen => socket  => $socketerr  ) if $socketerr;
   die 'Oops; $loop->listen failed but no error cause was found';
}

sub _listen_hostservice
{
   my $self = shift;
   my ( $listener, $host, $service, %params ) = @_;

   $host ||= "";
   defined $service or $service = ""; # might be 0

   my %gai_hints;
   exists $params{$_} and $gai_hints{$_} = $params{$_} for qw( family socktype protocol flags );

   defined $gai_hints{socktype} or defined $gai_hints{protocol} or
      carp "Attempting to ->listen without either 'socktype' or 'protocol' hint is not portable";

   $self->resolver->getaddrinfo(
      host    => $host,
      service => $service,
      passive => 1,
      %gai_hints,
   )->then( sub {
      my @addrs = @_;
      $self->_listen_addrs( $listener, \@addrs, %params );
   });
}

=head1 OS ABSTRACTIONS

Because the Magic Constructor searches for OS-specific subclasses of the Loop,
several abstractions of OS services are provided, in case specific OSes need
to give different implementations on that OS.

=cut

=head2 signame2num

   $signum = $loop->signame2num( $signame )

Legacy wrappers around L<IO::Async::OS> functions.

=cut

sub signame2num { shift; IO::Async::OS->signame2num( @_ ) }

=head2 time

   $time = $loop->time

Returns the current UNIX time in fractional seconds. This is currently
equivalent to C<Time::HiRes::time> but provided here as a utility for
programs to obtain the time current used by L<IO::Async> for its own timing
purposes.

=cut

sub time
{
   my $self = shift;
   return Time::HiRes::time;

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

=cut

# This class specifically does NOT implement this method, so that subclasses
# are forced to. The constructor will be checking....
sub __watch_io
{
   my $self = shift;
   my %params = @_;

   my $handle = delete $params{handle} or croak "Expected 'handle'";
   defined eval { $handle->fileno } or croak "Expected that 'handle' has defined ->fileno";

   # Silent "upgrade" to O_NONBLOCK
   $handle->blocking and $handle->blocking(0);

   my $watch = ( $self->{iowatches}->{$handle->fileno} ||= [] );

   $watch->[0] = $handle;

   if( exists $params{on_read_ready} ) {
      $watch->[1] = delete $params{on_read_ready};
   }

   if( exists $params{on_write_ready} ) {
      $watch->[2] = delete $params{on_write_ready};
   }

   if( exists $params{on_hangup} ) {
      $self->_CAN_ON_HANGUP or croak "Cannot watch_io for 'on_hangup' in ".ref($self);
      $watch->[3] = delete $params{on_hangup};
   }

   keys %params and croak "Unrecognised keys for ->watch_io - " . join( ", ", keys %params );
}

=head2 unwatch_io

   $loop->unwatch_io( %params )

This method removes a watch on an IO handle which was previously installed by
C<watch_io>.

The C<%params> hash takes the following keys:

=over 8

=item handle => IO

The IO handle to remove the watch for.

=item on_read_ready => BOOL

If true, remove the watch for read-readiness.

=item on_write_ready => BOOL

If true, remove the watch for write-readiness.

=back

Either or both callbacks may be removed at once. It is not an error to attempt
to remove a callback that is not present. If both callbacks were provided to
the C<watch_io> method and only one is removed by this method, the other shall
remain.

=cut

sub __unwatch_io
{
   my $self = shift;
   my %params = @_;

   my $handle = delete $params{handle} or croak "Expected 'handle'";

   my $watch = $self->{iowatches}->{$handle->fileno} or return;

   if( delete $params{on_read_ready} ) {
      undef $watch->[1];
   }

   if( delete $params{on_write_ready} ) {
      undef $watch->[2];
   }

   if( delete $params{on_hangup} ) {
      $self->_CAN_ON_HANGUP or croak "Cannot watch_io for 'on_hangup' in ".ref($self);
      undef $watch->[3];
   }

   if( not $watch->[1] and not $watch->[2] and not $watch->[3] ) {
      delete $self->{iowatches}->{$handle->fileno};
   }

   keys %params and croak "Unrecognised keys for ->unwatch_io - " . join( ", ", keys %params );
}

=head2 watch_signal

   $loop->watch_signal( $signal, $code )

This method adds a new signal handler to watch the given signal.

=over 8

=item $signal

The name of the signal to watch to. This should be a bare name like C<TERM>.

=item $code

A CODE reference to the handling callback.

=back

There can only be one callback per signal name. Registering a new one will
remove an existing one.

Applications should use a L<IO::Async::Signal> object, or call
C<attach_signal> instead of using this method.

This and C<unwatch_signal> are optional; a subclass may implement neither, or