lib/AnyEvent/Handle.pm

package AnyEvent::Handle;

no warnings;
use strict;

use AnyEvent ();
use AnyEvent::Util ();
use Scalar::Util ();
use Carp ();
use Fcntl ();
use Errno qw/EAGAIN EINTR/;

=head1 NAME

AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent

This module is experimental.

=cut

our $VERSION = '0.04';

=head1 SYNOPSIS

   use AnyEvent;
   use AnyEvent::Handle;

   my $cv = AnyEvent->condvar;

   my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);

   #TODO

   # or use the constructor to pass the callback:

   my $ae_fh2 =
      AnyEvent::Handle->new (
         fh => \*STDIN,
         on_eof => sub {
            $cv->broadcast;
         },
         #TODO
      );

   $cv->wait;

=head1 DESCRIPTION

This module is a helper module to make it easier to do event-based I/O on
filehandles. For utility functions for doing non-blocking connects and accepts
on sockets see L<AnyEvent::Util>.

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

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

=head1 METHODS

=over 4

=item B<new (%args)>

The constructor supports these arguments (all as key => value pairs).

=over 4

=item fh => $filehandle [MANDATORY]

The filehandle this L<AnyEvent::Handle> object will operate on.

NOTE: The filehandle will be set to non-blocking (using
AnyEvent::Util::fh_nonblocking).

=item on_eof => $cb->($self)

Set the callback to be called on EOF.

While not mandatory, it is highly recommended to set an eof callback,
otherwise you might end up with a closed socket while you are still
waiting for data.

=item on_error => $cb->($self)

This is the fatal error callback, that is called when, well, a fatal error
ocurs, such as not being able to resolve the hostname, failure to connect
or a read error.

The object will not be in a usable state when this callback has been
called.

On callback entrance, the value of C<$!> contains the operating system
error (or C<ENOSPC> or C<EPIPE>).

While not mandatory, it is I<highly> recommended to set this callback, as
you will not be notified of errors otherwise. The default simply calls
die.

=item on_read => $cb->($self)

This sets the default read callback, which is called when data arrives
and no read request is in the queue.

To access (and remove data from) the read buffer, use the C<< ->rbuf >>
method or acces sthe C<$self->{rbuf}> member directly.

When an EOF condition is detected then AnyEvent::Handle will first try to
feed all the remaining data to the queued callbacks and C<on_read> before
calling the C<on_eof> callback. If no progress can be made, then a fatal
error will be raised (with C<$!> set to C<EPIPE>).

=item on_drain => $cb->()

This sets the callback that is called when the write buffer becomes empty
(or when the callback is set and the buffer is empty already).

To append to the write buffer, use the C<< ->push_write >> method.

=item rbuf_max => <bytes>

If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
when the read buffer ever (strictly) exceeds this size. This is useful to
avoid denial-of-service attacks.

For example, a server accepting connections from untrusted sources should
be configured to accept only so-and-so much data that it cannot act on
(for example, when expecting a line, an attacker could send an unlimited
amount of data without a callback ever being called as long as the line
isn't finished).

=item read_size => <bytes>

The default read block size (the amount of bytes this module will try to read
on each [loop iteration). Default: C<4096>.

=item low_water_mark => <bytes>

Sets the amount of bytes (default: C<0>) that make up an "empty" write
buffer: If the write reaches this size or gets even samller it is
considered empty.

=item tls => "accept" | "connect" | Net::SSLeay::SSL object

When this parameter is given, it enables TLS (SSL) mode, that means it
will start making tls handshake and will transparently encrypt/decrypt
data.

For the TLS server side, use C<accept>, and for the TLS client side of a
connection, use C<connect> mode.

You can also provide your own TLS connection object, but you have
to make sure that you call either C<Net::SSLeay::set_connect_state>
or C<Net::SSLeay::set_accept_state> on it before you pass it to
AnyEvent::Handle.

=item tls_ctx => $ssl_ctx

Use the given Net::SSLeay::CTX object to create the new TLS connection
(unless a connection object was specified directly). If this parameter is
missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.

=back

=cut

sub new {
   my $class = shift;

   my $self = bless { @_ }, $class;

   $self->{fh} or Carp::croak "mandatory argument fh is missing";

   AnyEvent::Util::fh_nonblocking $self->{fh}, 1;

   if ($self->{tls}) {
      require Net::SSLeay;
      $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
   }

   $self->on_eof   (delete $self->{on_eof}  ) if $self->{on_eof};
   $self->on_error (delete $self->{on_error}) if $self->{on_error};
   $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
   $self->on_read  (delete $self->{on_read} ) if $self->{on_read};

   $self->start_read;

   $self
}

sub _shutdown {
   my ($self) = @_;

   delete $self->{rw};
   delete $self->{ww};
   delete $self->{fh};
}

sub error {
   my ($self) = @_;

   {
      local $!;
      $self->_shutdown;
   }

   if ($self->{on_error}) {
      $self->{on_error}($self);
   } else {
      die "AnyEvent::Handle uncaught fatal error: $!";
   }
}

=item $fh = $handle->fh

This method returns the filehandle of the L<AnyEvent::Handle> object.

=cut

sub fh { $_[0]->{fh} }

=item $handle->on_error ($cb)

Replace the current C<on_error> callback (see the C<on_error> constructor argument).

=cut

sub on_error {
   $_[0]{on_error} = $_[1];
}

=item $handle->on_eof ($cb)

Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).

=cut

sub on_eof {
   $_[0]{on_eof} = $_[1];
}

#############################################################################

=back

=head2 WRITE QUEUE

AnyEvent::Handle manages two queues per handle, one for writing and one
for reading.

The write queue is very simple: you can add data to its end, and
AnyEvent::Handle will automatically try to get rid of it for you.

When data could be writtena nd the write buffer is shorter then the low
water mark, the C<on_drain> callback will be invoked.

=over 4

=item $handle->on_drain ($cb)

Sets the C<on_drain> callback or clears it (see the description of
C<on_drain> in the constructor).

=cut

sub on_drain {
   my ($self, $cb) = @_;

   $self->{on_drain} = $cb;

   $cb->($self)
      if $cb && $self->{low_water_mark} >= length $self->{wbuf};
}

=item $handle->push_write ($data)

Queues the given scalar to be written. You can push as much data as you
want (only limited by the available memory), as C<AnyEvent::Handle>
buffers it independently of the kernel.

=cut

sub _drain_wbuf {
   my ($self) = @_;

   unless ($self->{ww}) {
      Scalar::Util::weaken $self;
      my $cb = sub {
         my $len = syswrite $self->{fh}, $self->{wbuf};

         if ($len > 0) {
            substr $self->{wbuf}, 0, $len, "";

            $self->{on_drain}($self)
               if $self->{low_water_mark} >= length $self->{wbuf}
                  && $self->{on_drain};

            delete $self->{ww} unless length $self->{wbuf};
         } elsif ($! != EAGAIN && $! != EINTR) {
            $self->error;
         }
      };

      $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);

      $cb->($self);
   };
}

sub push_write {
   my $self = shift;

   if ($self->{filter_w}) {
      $self->{filter_w}->($self, \$_[0]);
   } else {
      $self->{wbuf} .= $_[0];
      $self->_drain_wbuf;
   }
}

#############################################################################

=back

=head2 READ QUEUE

AnyEvent::Handle manages two queues per handle, one for writing and one
for reading.

The read queue is more complex than the write queue. It can be used in two
ways, the "simple" way, using only C<on_read> and the "complex" way, using
a queue.

In the simple case, you just install an C<on_read> callback and whenever
new data arrives, it will be called. You can then remove some data (if
enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
or not.

In the more complex case, you want to queue multiple callbacks. In this
case, AnyEvent::Handle will call the first queued callback each time new
data arrives and removes it when it has done its job (see C<push_read>,
below).

This way you can, for example, push three line-reads, followed by reading
a chunk of data, and AnyEvent::Handle will execute them in order.

Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
the specified number of bytes which give an XML datagram.

   # in the default state, expect some header bytes
   $handle->on_read (sub {
      # some data is here, now queue the length-header-read (4 octets)
      shift->unshift_read_chunk (4, sub {
         # header arrived, decode
         my $len = unpack "N", $_[1];

         # now read the payload
         shift->unshift_read_chunk ($len, sub {
            my $xml = $_[1];
            # handle xml
         });
      });
   });

Example 2: Implement a client for a protocol that replies either with
"OK" and another line or "ERROR" for one request, and 64 bytes for the
second request. Due tot he availability of a full queue, we can just
pipeline sending both requests and manipulate the queue as necessary in
the callbacks:

   # request one
   $handle->push_write ("request 1\015\012");

   # we expect "ERROR" or "OK" as response, so push a line read
   $handle->push_read_line (sub {
      # if we got an "OK", we have to _prepend_ another line,
      # so it will be read before the second request reads its 64 bytes
      # which are already in the queue when this callback is called
      # we don't do this in case we got an error
      if ($_[1] eq "OK") {
         $_[0]->unshift_read_line (sub {
            my $response = $_[1];
            ...
         });
      }
   });

   # request two
   $handle->push_write ("request 2\015\012");

   # simply read 64 bytes, always
   $handle->push_read_chunk (64, sub {
      my $response = $_[1];
      ...
   });

=over 4

=cut

sub _drain_rbuf {
   my ($self) = @_;

   if (
      defined $self->{rbuf_max}
      && $self->{rbuf_max} < length $self->{rbuf}
   ) {
      $! = &Errno::ENOSPC; return $self->error;
   }

   return if $self->{in_drain};
   local $self->{in_drain} = 1;

   while (my $len = length $self->{rbuf}) {
      no strict 'refs';
      if (my $cb = shift @{ $self->{queue} }) {
         if (!$cb->($self)) {
            if ($self->{eof}) {
               # no progress can be made (not enough data and no data forthcoming)
               $! = &Errno::EPIPE; return $self->error;
            }

            unshift @{ $self->{queue} }, $cb;
            return;
         }
      } elsif ($self->{on_read}) {
         $self->{on_read}($self);

         if (
            $self->{eof}                    # if no further data will arrive
            && $len == length $self->{rbuf} # and no data has been consumed
            && !@{ $self->{queue} }         # and the queue is still empty
            && $self->{on_read}             # and we still want to read data
         ) {
            # then no progress can be made
            $! = &Errno::EPIPE; return $self->error;
         }
      } else {
         # read side becomes idle
         delete $self->{rw};
         return;
      }
   }

   if ($self->{eof}) {
      $self->_shutdown;
      $self->{on_eof}($self)
         if $self->{on_eof};
   }
}

=item $handle->on_read ($cb)

This replaces the currently set C<on_read> callback, or clears it (when
the new callback is C<undef>). See the description of C<on_read> in the
constructor.

=cut

sub on_read {
   my ($self, $cb) = @_;

   $self->{on_read} = $cb;
}

=item $handle->rbuf

Returns the read buffer (as a modifiable lvalue).

You can access the read buffer directly as the C<< ->{rbuf} >> member, if
you want.

NOTE: The read buffer should only be used or modified if the C<on_read>,
C<push_read> or C<unshift_read> methods are used. The other read methods
automatically manage the read buffer.

=cut

sub rbuf : lvalue {
   $_[0]{rbuf}
}

=item $handle->push_read ($cb)

=item $handle->unshift_read ($cb)

Append the given callback to the end of the queue (C<push_read>) or
prepend it (C<unshift_read>).

The callback is called each time some additional read data arrives.

It must check wether enough data is in the read buffer already.

If not enough data is available, it must return the empty list or a false
value, in which case it will be called repeatedly until enough data is
available (or an error condition is detected).

If enough data was available, then the callback must remove all data it is
interested in (which can be none at all) and return a true value. After returning
true, it will be removed from the queue.

=cut

sub push_read {
   my ($self, $cb) = @_;

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

sub unshift_read {
   my ($self, $cb) = @_;

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

=item $handle->push_read_chunk ($len, $cb->($self, $data))

=item $handle->unshift_read_chunk ($len, $cb->($self, $data))

Append the given callback to the end of the queue (C<push_read_chunk>) or
prepend it (C<unshift_read_chunk>).

The callback will be called only once C<$len> bytes have been read, and
these C<$len> bytes will be passed to the callback.

=cut

sub _read_chunk($$) {
   my ($self, $len, $cb) = @_;

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

sub push_read_chunk {
   $_[0]->push_read (&_read_chunk);
}


sub unshift_read_chunk {
   $_[0]->unshift_read (&_read_chunk);
}

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

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

Append the given callback to the end of the queue (C<push_read_line>) or
prepend it (C<unshift_read_line>).

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

sub _read_line($$) {
   my $self = shift;
   my $cb = pop;
   my $eol = @_ ? shift : qr|(\015?\012)|;
   my $pos;

   $eol = quotemeta $eol unless ref $eol;
   $eol = qr|^(.*?)($eol)|s;

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

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

sub push_read_line {
   $_[0]->push_read (&_read_line);
}

sub unshift_read_line {
   $_[0]->unshift_read (&_read_line);
}

=item $handle->stop_read

=item $handle->start_read

In rare cases you actually do not want to read anything from the
socket. In this case you can call C<stop_read>. Neither C<on_read> no
any queued callbacks will be executed then. To start readign again, call
C<start_read>.

=cut

sub stop_read {
   my ($self) = @_;

   delete $self->{rw};
}

sub start_read {
   my ($self) = @_;

   unless ($self->{rw} || $self->{eof}) {
      Scalar::Util::weaken $self;

      $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
         my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
         my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;

         if ($len > 0) {
            $self->{filter_r}
               ? $self->{filter_r}->($self, $rbuf)
               : $self->_drain_rbuf;

         } elsif (defined $len) {
            delete $self->{rw};
            $self->{eof} = 1;
            $self->_drain_rbuf;

         } elsif ($! != EAGAIN && $! != EINTR) {
            return $self->error;
         }
      });
   }
}

sub _dotls {
   my ($self) = @_;

   if (length $self->{tls_wbuf}) {
      my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf});
      substr $self->{tls_wbuf}, 0, $len, "" if $len > 0;
   }

   if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) {
      $self->{wbuf} .= $buf;
      $self->_drain_wbuf;
   }

   if (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
      $self->{rbuf} .= $buf;
      $self->_drain_rbuf;
   } elsif (
      (my $err = Net::SSLeay::get_error ($self->{tls}, -1))
      != Net::SSLeay::ERROR_WANT_READ ()
   ) {
      if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
         $self->error;
      } elsif ($err == Net::SSLeay::ERROR_SSL ())  {
         $! = &Errno::EIO;
         $self->error;
      }

      # all others are fine for our purposes
   }
}

# TODO: maybe document...
sub starttls {
   my ($self, $ssl, $ctx) = @_;

   if ($ssl eq "accept") {
      $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
      Net::SSLeay::set_accept_state ($ssl);
   } elsif ($ssl eq "connect") {
      $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
      Net::SSLeay::set_connect_state ($ssl);
   }

   $self->{tls} = $ssl;

   $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
   $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());

   Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});

   $self->{filter_w} = sub {
      $_[0]{tls_wbuf} .= ${$_[1]};
      &_dotls;
   };
   $self->{filter_r} = sub {
      Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]});
      &_dotls;
   };
}

sub DESTROY {
   my $self = shift;

   Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
}

=item AnyEvent::Handle::TLS_CTX

This function creates and returns the Net::SSLeay::CTX object used by
default for TLS mode.

The context is created like this:

   Net::SSLeay::load_error_strings;
   Net::SSLeay::SSLeay_add_ssl_algorithms;
   Net::SSLeay::randomize;

   my $CTX = Net::SSLeay::CTX_new;

   Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL

=cut

our $TLS_CTX;

sub TLS_CTX() {
   $TLS_CTX || do {
      require Net::SSLeay;

      Net::SSLeay::load_error_strings ();
      Net::SSLeay::SSLeay_add_ssl_algorithms ();
      Net::SSLeay::randomize ();

      $TLS_CTX = Net::SSLeay::CTX_new ();

      Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());

      $TLS_CTX
   }
}

=back

=head1 AUTHOR

Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.

=cut

1; # End of AnyEvent::Handle
Revision:	1.19
Committed:	Sat May 24 05:57:11 2008 UTC (15 years, 11 months ago) by root
Branch:	MAIN
Changes since 1.18:	+126 -0 lines
Log Message:	ssl support
#	Content
1	package AnyEvent::Handle;
2
3	no warnings;
4	use strict;
5
6	use AnyEvent ();
7	use AnyEvent::Util ();
8	use Scalar::Util ();
9	use Carp ();
10	use Fcntl ();
11	use Errno qw/EAGAIN EINTR/;
12
13	=head1 NAME
14
15	AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent
16
17	This module is experimental.
18
19	=cut
20
21	our $VERSION = '0.04';
22
23	=head1 SYNOPSIS
24
25	use AnyEvent;
26	use AnyEvent::Handle;
27
28	my $cv = AnyEvent->condvar;
29
30	my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);
31
32	#TODO
33
34	# or use the constructor to pass the callback:
35
36	my $ae_fh2 =
37	AnyEvent::Handle->new (
38	fh => \*STDIN,
39	on_eof => sub {
40	$cv->broadcast;
41	},
42	#TODO
43	);
44
45	$cv->wait;
46
47	=head1 DESCRIPTION
48
49	This module is a helper module to make it easier to do event-based I/O on
50	filehandles. For utility functions for doing non-blocking connects and accepts
51	on sockets see L<AnyEvent::Util>.
52
53	In the following, when the documentation refers to of "bytes" then this
54	means characters. As sysread and syswrite are used for all I/O, their
55	treatment of characters applies to this module as well.
56
57	All callbacks will be invoked with the handle object as their first
58	argument.
59
60	=head1 METHODS
61
62	=over 4
63
64	=item B<new (%args)>
65
66	The constructor supports these arguments (all as key => value pairs).
67
68	=over 4
69
70	=item fh => $filehandle [MANDATORY]
71
72	The filehandle this L<AnyEvent::Handle> object will operate on.
73
74	NOTE: The filehandle will be set to non-blocking (using
75	AnyEvent::Util::fh_nonblocking).
76
77	=item on_eof => $cb->($self)
78
79	Set the callback to be called on EOF.
80
81	While not mandatory, it is highly recommended to set an eof callback,
82	otherwise you might end up with a closed socket while you are still
83	waiting for data.
84
85	=item on_error => $cb->($self)
86
87	This is the fatal error callback, that is called when, well, a fatal error
88	ocurs, such as not being able to resolve the hostname, failure to connect
89	or a read error.
90
91	The object will not be in a usable state when this callback has been
92	called.
93
94	On callback entrance, the value of C<$!> contains the operating system
95	error (or C<ENOSPC> or C<EPIPE>).
96
97	While not mandatory, it is I<highly> recommended to set this callback, as
98	you will not be notified of errors otherwise. The default simply calls
99	die.
100
101	=item on_read => $cb->($self)
102
103	This sets the default read callback, which is called when data arrives
104	and no read request is in the queue.
105
106	To access (and remove data from) the read buffer, use the C<< ->rbuf >>
107	method or acces sthe C<$self->{rbuf}> member directly.
108
109	When an EOF condition is detected then AnyEvent::Handle will first try to
110	feed all the remaining data to the queued callbacks and C<on_read> before
111	calling the C<on_eof> callback. If no progress can be made, then a fatal
112	error will be raised (with C<$!> set to C<EPIPE>).
113
114	=item on_drain => $cb->()
115
116	This sets the callback that is called when the write buffer becomes empty
117	(or when the callback is set and the buffer is empty already).
118
119	To append to the write buffer, use the C<< ->push_write >> method.
120
121	=item rbuf_max => <bytes>
122
123	If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
124	when the read buffer ever (strictly) exceeds this size. This is useful to
125	avoid denial-of-service attacks.
126
127	For example, a server accepting connections from untrusted sources should
128	be configured to accept only so-and-so much data that it cannot act on
129	(for example, when expecting a line, an attacker could send an unlimited
130	amount of data without a callback ever being called as long as the line
131	isn't finished).
132
133	=item read_size => <bytes>
134
135	The default read block size (the amount of bytes this module will try to read
136	on each [loop iteration). Default: C<4096>.
137
138	=item low_water_mark => <bytes>
139
140	Sets the amount of bytes (default: C<0>) that make up an "empty" write
141	buffer: If the write reaches this size or gets even samller it is
142	considered empty.
143
144	=item tls => "accept" \| "connect" \| Net::SSLeay::SSL object
145
146	When this parameter is given, it enables TLS (SSL) mode, that means it
147	will start making tls handshake and will transparently encrypt/decrypt
148	data.
149
150	For the TLS server side, use C<accept>, and for the TLS client side of a
151	connection, use C<connect> mode.
152
153	You can also provide your own TLS connection object, but you have
154	to make sure that you call either C<Net::SSLeay::set_connect_state>
155	or C<Net::SSLeay::set_accept_state> on it before you pass it to
156	AnyEvent::Handle.
157
158	=item tls_ctx => $ssl_ctx
159
160	Use the given Net::SSLeay::CTX object to create the new TLS connection
161	(unless a connection object was specified directly). If this parameter is
162	missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
163
164	=back
165
166	=cut
167
168	sub new {
169	my $class = shift;
170
171	my $self = bless { @_ }, $class;
172
173	$self->{fh} or Carp::croak "mandatory argument fh is missing";
174
175	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
176
177	if ($self->{tls}) {
178	require Net::SSLeay;
179	$self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
180	}
181
182	$self->on_eof (delete $self->{on_eof} ) if $self->{on_eof};
183	$self->on_error (delete $self->{on_error}) if $self->{on_error};
184	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
185	$self->on_read (delete $self->{on_read} ) if $self->{on_read};
186
187	$self->start_read;
188
189	$self
190	}
191
192	sub _shutdown {
193	my ($self) = @_;
194
195	delete $self->{rw};
196	delete $self->{ww};
197	delete $self->{fh};
198	}
199
200	sub error {
201	my ($self) = @_;
202
203	{
204	local $!;
205	$self->_shutdown;
206	}
207
208	if ($self->{on_error}) {
209	$self->{on_error}($self);
210	} else {
211	die "AnyEvent::Handle uncaught fatal error: $!";
212	}
213	}
214
215	=item $fh = $handle->fh
216
217	This method returns the filehandle of the L<AnyEvent::Handle> object.
218
219	=cut
220
221	sub fh { $_[0]->{fh} }
222
223	=item $handle->on_error ($cb)
224
225	Replace the current C<on_error> callback (see the C<on_error> constructor argument).
226
227	=cut
228
229	sub on_error {
230	$_[0]{on_error} = $_[1];
231	}
232
233	=item $handle->on_eof ($cb)
234
235	Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
236
237	=cut
238
239	sub on_eof {
240	$_[0]{on_eof} = $_[1];
241	}
242
243	#############################################################################
244
245	=back
246
247	=head2 WRITE QUEUE
248
249	AnyEvent::Handle manages two queues per handle, one for writing and one
250	for reading.
251
252	The write queue is very simple: you can add data to its end, and
253	AnyEvent::Handle will automatically try to get rid of it for you.
254
255	When data could be writtena nd the write buffer is shorter then the low
256	water mark, the C<on_drain> callback will be invoked.
257
258	=over 4
259
260	=item $handle->on_drain ($cb)
261
262	Sets the C<on_drain> callback or clears it (see the description of
263	C<on_drain> in the constructor).
264
265	=cut
266
267	sub on_drain {
268	my ($self, $cb) = @_;
269
270	$self->{on_drain} = $cb;
271
272	$cb->($self)
273	if $cb && $self->{low_water_mark} >= length $self->{wbuf};
274	}
275
276	=item $handle->push_write ($data)
277
278	Queues the given scalar to be written. You can push as much data as you
279	want (only limited by the available memory), as C<AnyEvent::Handle>
280	buffers it independently of the kernel.
281
282	=cut
283
284	sub _drain_wbuf {
285	my ($self) = @_;
286
287	unless ($self->{ww}) {
288	Scalar::Util::weaken $self;
289	my $cb = sub {
290	my $len = syswrite $self->{fh}, $self->{wbuf};
291
292	if ($len > 0) {
293	substr $self->{wbuf}, 0, $len, "";
294
295	$self->{on_drain}($self)
296	if $self->{low_water_mark} >= length $self->{wbuf}
297	&& $self->{on_drain};
298
299	delete $self->{ww} unless length $self->{wbuf};
300	} elsif ($! != EAGAIN && $! != EINTR) {
301	$self->error;
302	}
303	};
304
305	$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
306
307	$cb->($self);
308	};
309	}
310
311	sub push_write {
312	my $self = shift;
313
314	if ($self->{filter_w}) {
315	$self->{filter_w}->($self, \$_[0]);
316	} else {
317	$self->{wbuf} .= $_[0];
318	$self->_drain_wbuf;
319	}
320	}
321
322	#############################################################################
323
324	=back
325
326	=head2 READ QUEUE
327
328	AnyEvent::Handle manages two queues per handle, one for writing and one
329	for reading.
330
331	The read queue is more complex than the write queue. It can be used in two
332	ways, the "simple" way, using only C<on_read> and the "complex" way, using
333	a queue.
334
335	In the simple case, you just install an C<on_read> callback and whenever
336	new data arrives, it will be called. You can then remove some data (if
337	enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
338	or not.
339
340	In the more complex case, you want to queue multiple callbacks. In this
341	case, AnyEvent::Handle will call the first queued callback each time new
342	data arrives and removes it when it has done its job (see C<push_read>,
343	below).
344
345	This way you can, for example, push three line-reads, followed by reading
346	a chunk of data, and AnyEvent::Handle will execute them in order.
347
348	Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
349	the specified number of bytes which give an XML datagram.
350
351	# in the default state, expect some header bytes
352	$handle->on_read (sub {
353	# some data is here, now queue the length-header-read (4 octets)
354	shift->unshift_read_chunk (4, sub {
355	# header arrived, decode
356	my $len = unpack "N", $_[1];
357
358	# now read the payload
359	shift->unshift_read_chunk ($len, sub {
360	my $xml = $_[1];
361	# handle xml
362	});
363	});
364	});
365
366	Example 2: Implement a client for a protocol that replies either with
367	"OK" and another line or "ERROR" for one request, and 64 bytes for the
368	second request. Due tot he availability of a full queue, we can just
369	pipeline sending both requests and manipulate the queue as necessary in
370	the callbacks:
371
372	# request one
373	$handle->push_write ("request 1\015\012");
374
375	# we expect "ERROR" or "OK" as response, so push a line read
376	$handle->push_read_line (sub {
377	# if we got an "OK", we have to _prepend_ another line,
378	# so it will be read before the second request reads its 64 bytes
379	# which are already in the queue when this callback is called
380	# we don't do this in case we got an error
381	if ($_[1] eq "OK") {
382	$_[0]->unshift_read_line (sub {
383	my $response = $_[1];
384	...
385	});
386	}
387	});
388
389	# request two
390	$handle->push_write ("request 2\015\012");
391
392	# simply read 64 bytes, always
393	$handle->push_read_chunk (64, sub {
394	my $response = $_[1];
395	...
396	});
397
398	=over 4
399
400	=cut
401
402	sub _drain_rbuf {
403	my ($self) = @_;
404
405	if (
406	defined $self->{rbuf_max}
407	&& $self->{rbuf_max} < length $self->{rbuf}
408	) {
409	$! = &Errno::ENOSPC; return $self->error;
410	}
411
412	return if $self->{in_drain};
413	local $self->{in_drain} = 1;
414
415	while (my $len = length $self->{rbuf}) {
416	no strict 'refs';
417	if (my $cb = shift @{ $self->{queue} }) {
418	if (!$cb->($self)) {
419	if ($self->{eof}) {
420	# no progress can be made (not enough data and no data forthcoming)
421	$! = &Errno::EPIPE; return $self->error;
422	}
423
424	unshift @{ $self->{queue} }, $cb;
425	return;
426	}
427	} elsif ($self->{on_read}) {
428	$self->{on_read}($self);
429
430	if (
431	$self->{eof} # if no further data will arrive
432	&& $len == length $self->{rbuf} # and no data has been consumed
433	&& !@{ $self->{queue} } # and the queue is still empty
434	&& $self->{on_read} # and we still want to read data
435	) {
436	# then no progress can be made
437	$! = &Errno::EPIPE; return $self->error;
438	}
439	} else {
440	# read side becomes idle
441	delete $self->{rw};
442	return;
443	}
444	}
445
446	if ($self->{eof}) {
447	$self->_shutdown;
448	$self->{on_eof}($self)
449	if $self->{on_eof};
450	}
451	}
452
453	=item $handle->on_read ($cb)
454
455	This replaces the currently set C<on_read> callback, or clears it (when
456	the new callback is C<undef>). See the description of C<on_read> in the
457	constructor.
458
459	=cut
460
461	sub on_read {
462	my ($self, $cb) = @_;
463
464	$self->{on_read} = $cb;
465	}
466
467	=item $handle->rbuf
468
469	Returns the read buffer (as a modifiable lvalue).
470
471	You can access the read buffer directly as the C<< ->{rbuf} >> member, if
472	you want.
473
474	NOTE: The read buffer should only be used or modified if the C<on_read>,
475	C<push_read> or C<unshift_read> methods are used. The other read methods
476	automatically manage the read buffer.
477
478	=cut
479
480	sub rbuf : lvalue {
481	$_[0]{rbuf}
482	}
483
484	=item $handle->push_read ($cb)
485
486	=item $handle->unshift_read ($cb)
487
488	Append the given callback to the end of the queue (C<push_read>) or
489	prepend it (C<unshift_read>).
490
491	The callback is called each time some additional read data arrives.
492
493	It must check wether enough data is in the read buffer already.
494
495	If not enough data is available, it must return the empty list or a false
496	value, in which case it will be called repeatedly until enough data is
497	available (or an error condition is detected).
498
499	If enough data was available, then the callback must remove all data it is
500	interested in (which can be none at all) and return a true value. After returning
501	true, it will be removed from the queue.
502
503	=cut
504
505	sub push_read {
506	my ($self, $cb) = @_;
507
508	push @{ $self->{queue} }, $cb;
509	$self->_drain_rbuf;
510	}
511
512	sub unshift_read {
513	my ($self, $cb) = @_;
514
515	push @{ $self->{queue} }, $cb;
516	$self->_drain_rbuf;
517	}
518
519	=item $handle->push_read_chunk ($len, $cb->($self, $data))
520
521	=item $handle->unshift_read_chunk ($len, $cb->($self, $data))
522
523	Append the given callback to the end of the queue (C<push_read_chunk>) or
524	prepend it (C<unshift_read_chunk>).
525
526	The callback will be called only once C<$len> bytes have been read, and
527	these C<$len> bytes will be passed to the callback.
528
529	=cut
530
531	sub _read_chunk($$) {
532	my ($self, $len, $cb) = @_;
533
534	sub {
535	$len <= length $_[0]{rbuf} or return;
536	$cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
537	1
538	}
539	}
540
541	sub push_read_chunk {
542	$_[0]->push_read (&_read_chunk);
543	}
544
545
546	sub unshift_read_chunk {
547	$_[0]->unshift_read (&_read_chunk);
548	}
549
550	=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
551
552	=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
553
554	Append the given callback to the end of the queue (C<push_read_line>) or
555	prepend it (C<unshift_read_line>).
556
557	The callback will be called only once a full line (including the end of
558	line marker, C<$eol>) has been read. This line (excluding the end of line
559	marker) will be passed to the callback as second argument (C<$line>), and
560	the end of line marker as the third argument (C<$eol>).
561
562	The end of line marker, C<$eol>, can be either a string, in which case it
563	will be interpreted as a fixed record end marker, or it can be a regex
564	object (e.g. created by C<qr>), in which case it is interpreted as a
565	regular expression.
566
567	The end of line marker argument C<$eol> is optional, if it is missing (NOT
568	undef), then C<qr\|\015?\012\|> is used (which is good for most internet
569	protocols).
570
571	Partial lines at the end of the stream will never be returned, as they are
572	not marked by the end of line marker.
573
574	=cut
575
576	sub _read_line($$) {
577	my $self = shift;
578	my $cb = pop;
579	my $eol = @_ ? shift : qr\|(\015?\012)\|;
580	my $pos;
581
582	$eol = quotemeta $eol unless ref $eol;
583	$eol = qr\|^(.*?)($eol)\|s;
584
585	sub {
586	$_[0]{rbuf} =~ s/$eol// or return;
587
588	$cb->($_[0], $1, $2);
589	1
590	}
591	}
592
593	sub push_read_line {
594	$_[0]->push_read (&_read_line);
595	}
596
597	sub unshift_read_line {
598	$_[0]->unshift_read (&_read_line);
599	}
600
601	=item $handle->stop_read
602
603	=item $handle->start_read
604
605	In rare cases you actually do not want to read anything from the
606	socket. In this case you can call C<stop_read>. Neither C<on_read> no
607	any queued callbacks will be executed then. To start readign again, call
608	C<start_read>.
609
610	=cut
611
612	sub stop_read {
613	my ($self) = @_;
614
615	delete $self->{rw};
616	}
617
618	sub start_read {
619	my ($self) = @_;
620
621	unless ($self->{rw} \|\| $self->{eof}) {
622	Scalar::Util::weaken $self;
623
624	$self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
625	my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
626	my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} \|\| 8192, length $$rbuf;
627
628	if ($len > 0) {
629	$self->{filter_r}
630	? $self->{filter_r}->($self, $rbuf)
631	: $self->_drain_rbuf;
632
633	} elsif (defined $len) {
634	delete $self->{rw};
635	$self->{eof} = 1;
636	$self->_drain_rbuf;
637
638	} elsif ($! != EAGAIN && $! != EINTR) {
639	return $self->error;
640	}
641	});
642	}
643	}
644
645	sub _dotls {
646	my ($self) = @_;
647
648	if (length $self->{tls_wbuf}) {
649	my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf});
650	substr $self->{tls_wbuf}, 0, $len, "" if $len > 0;
651	}
652
653	if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) {
654	$self->{wbuf} .= $buf;
655	$self->_drain_wbuf;
656	}
657
658	if (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
659	$self->{rbuf} .= $buf;
660	$self->_drain_rbuf;
661	} elsif (
662	(my $err = Net::SSLeay::get_error ($self->{tls}, -1))
663	!= Net::SSLeay::ERROR_WANT_READ ()
664	) {
665	if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
666	$self->error;
667	} elsif ($err == Net::SSLeay::ERROR_SSL ()) {
668	$! = &Errno::EIO;
669	$self->error;
670	}
671
672	# all others are fine for our purposes
673	}
674	}
675
676	# TODO: maybe document...
677	sub starttls {
678	my ($self, $ssl, $ctx) = @_;
679
680	if ($ssl eq "accept") {
681	$ssl = Net::SSLeay::new ($ctx \|\| TLS_CTX ());
682	Net::SSLeay::set_accept_state ($ssl);
683	} elsif ($ssl eq "connect") {
684	$ssl = Net::SSLeay::new ($ctx \|\| TLS_CTX ());
685	Net::SSLeay::set_connect_state ($ssl);
686	}
687
688	$self->{tls} = $ssl;
689
690	$self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
691	$self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
692
693	Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});
694
695	$self->{filter_w} = sub {
696	$_[0]{tls_wbuf} .= ${$_[1]};
697	&_dotls;
698	};
699	$self->{filter_r} = sub {
700	Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]});
701	&_dotls;
702	};
703	}
704
705	sub DESTROY {
706	my $self = shift;
707
708	Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
709	}
710
711	=item AnyEvent::Handle::TLS_CTX
712
713	This function creates and returns the Net::SSLeay::CTX object used by
714	default for TLS mode.
715
716	The context is created like this:
717
718	Net::SSLeay::load_error_strings;
719	Net::SSLeay::SSLeay_add_ssl_algorithms;
720	Net::SSLeay::randomize;
721
722	my $CTX = Net::SSLeay::CTX_new;
723
724	Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
725
726	=cut
727
728	our $TLS_CTX;
729
730	sub TLS_CTX() {
731	$TLS_CTX \|\| do {
732	require Net::SSLeay;
733
734	Net::SSLeay::load_error_strings ();
735	Net::SSLeay::SSLeay_add_ssl_algorithms ();
736	Net::SSLeay::randomize ();
737
738	$TLS_CTX = Net::SSLeay::CTX_new ();
739
740	Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
741
742	$TLS_CTX
743	}
744	}
745
746	=back
747
748	=head1 AUTHOR
749
750	Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
751
752	=cut
753
754	1; # End of AnyEvent::Handle