view release on metacpan or  search on metacpan

lib/AnyEvent/Handle.pm  view on Meta::CPAN

=head1 NAME

AnyEvent::Handle - non-blocking I/O on streaming handles via AnyEvent

=head1 SYNOPSIS

   use AnyEvent;
   use AnyEvent::Handle;

   my $cv = AnyEvent->condvar;

   my $hdl; $hdl = new AnyEvent::Handle
      fh => \*STDIN,
      on_error => sub {
         my ($hdl, $fatal, $msg) = @_;
         AE::log error => $msg;
         $hdl->destroy;
         $cv->send;
      };

   # send some request line
   $hdl->push_write ("getinfo\015\012");

   # read the response line
   $hdl->push_read (line => sub {
      my ($hdl, $line) = @_;
      say "got line <$line>";
      $cv->send;
   });

   $cv->recv;

=head1 DESCRIPTION

This is a helper module to make it easier to do event-based I/O
on stream-based filehandles (sockets, pipes, and other stream
things). Specifically, it doesn't work as expected on files, packet-based
sockets or similar things.

The L<AnyEvent::Intro> tutorial contains some well-documented
AnyEvent::Handle examples.

In the following, where the documentation refers to "bytes", it means
characters. As sysread and syswrite are used for all I/O, their
treatment of characters applies to this module as well.

At the very minimum, you should specify C<fh> or C<connect>, and the
C<on_error> callback.

All callbacks will be invoked with the handle object as their first
argument.

=cut

package AnyEvent::Handle;

use Scalar::Util ();
use List::Util ();
use Carp ();
use Errno qw(EAGAIN EWOULDBLOCK EINTR);

use AnyEvent (); BEGIN { AnyEvent::common_sense }
use AnyEvent::Util qw(WSAEWOULDBLOCK);

our $VERSION = $AnyEvent::VERSION;

sub _load_func($) {
   my $func = $_[0];

   unless (defined &$func) {
      my $pkg = $func;
      do {
         $pkg =~ s/::[^:]+$//
            or return;
         eval "require $pkg";
      } until defined &$func;
   }

   \&$func
}

sub MAX_READ_SIZE() { 131072 }

=head1 METHODS

=over 4

lib/AnyEvent/Handle.pm  view on Meta::CPAN

sub register_read_type($$) {
   $RH{$_[0]} = $_[1];
}

sub push_read {
   my $self = shift;
   my $cb = pop;

   if (@_) {
      my $type = shift;

      $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
             or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_read")
            ->($self, $cb, @_);
   }

   push @{ $self->{_queue} }, $cb;
   $self->_drain_rbuf;
}

sub unshift_read {
   my $self = shift;
   my $cb = pop;

   if (@_) {
      my $type = shift;

      $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
             or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::unshift_read")
            ->($self, $cb, @_);
   }

   unshift @{ $self->{_queue} }, $cb;
   $self->_drain_rbuf;
}

=item $handle->push_read (type => @args, $cb)

=item $handle->unshift_read (type => @args, $cb)

Instead of providing a callback that parses the data itself you can chose
between a number of predefined parsing formats, for chunks of data, lines
etc. You can also specify the (fully qualified) name of a package, in
which case AnyEvent tries to load the package and then expects to find the
C<anyevent_read_type> function inside (see "custom read types", below).

Predefined types are (if you have ideas for additional types, feel free to
drop by and tell us):

=over 4

=item chunk => $octets, $cb->($handle, $data)

Invoke the callback only once C<$octets> bytes have been read. Pass the
data read to the callback. The callback will never be called with less
data.

Example: read 2 bytes.

   $handle->push_read (chunk => 2, sub {
      say "yay " . unpack "H*", $_[1];
   });

=cut

register_read_type chunk => sub {
   my ($self, $cb, $len) = @_;

   sub {
      $len <= length $_[0]{rbuf} or return;
      $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
      1
   }
};

=item line => [$eol, ]$cb->($handle, $line, $eol)

The callback will be called only once a full line (including the end of
line marker, C<$eol>) has been read. This line (excluding the end of line
marker) will be passed to the callback as second argument (C<$line>), and
the end of line marker as the third argument (C<$eol>).

The end of line marker, C<$eol>, can be either a string, in which case it
will be interpreted as a fixed record end marker, or it can be a regex
object (e.g. created by C<qr>), in which case it is interpreted as a
regular expression.

The end of line marker argument C<$eol> is optional, if it is missing (NOT
undef), then C<qr|\015?\012|> is used (which is good for most internet
protocols).

Partial lines at the end of the stream will never be returned, as they are
not marked by the end of line marker.

=cut

register_read_type line => sub {
   my ($self, $cb, $eol) = @_;

   if (@_ < 3) {
      # this is faster then the generic code below
      sub {
         (my $pos = index $_[0]{rbuf}, "\012") >= 0
            or return;

         (my $str = substr $_[0]{rbuf}, 0, $pos + 1, "") =~ s/(\015?\012)\Z// or die;
         $cb->($_[0], $str, "$1");
         1
      }
   } else {
      $eol = quotemeta $eol unless ref $eol;
      $eol = qr|^(.*?)($eol)|s;

      sub {
         $_[0]{rbuf} =~ s/$eol// or return;

         $cb->($_[0], "$1", "$2");
         1
      }
   }
};