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

=cut

our $VERSION = '0.02';

=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 (and sockets, see L<AnyEvent::Socket> for an easy way to make
non-blocking resolves and connects).

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_error => $cb->($self) [MANDATORY]

This is the fatal error callback, that is called when 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 opertaing system
error (or C<ENOSPC> or C<EPIPE>).

=item on_eof => $cb->($self) [MANDATORY]

Set the callback to be called on EOF.

=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.  If the read callback is C<undef>
or has never been set, than AnyEvent::Handle will cease reading from the
filehandle.

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.

=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;

   $self->on_error ((delete $self->{on_error}) or Carp::croak "mandatory argument on_error is missing");
   $self->on_eof   ((delete $self->{on_eof}  ) or Carp::croak "mandatory argument on_eof is missing");

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

   $self
}

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

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

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

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

   $self->{on_error}($self);
}

=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 push_write {
   my ($self, $data) = @_;

   $self->{wbuf} .= $data;

   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);
   };
}

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

=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

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

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

   while (my $len = length $self->{rbuf}) {
      no strict 'refs';
      if (@{ $self->{queue} }) {
         if ($self->{queue}[0]($self)) {
            shift @{ $self->{queue} };
         } elsif ($self->{eof}) {
            # no progress can be made (not enough data and no data forthcoming)
            $! = &Errno::EPIPE; return $self->error;
         } else {
            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);
   }
}

=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;

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

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

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

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

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

         $self->_drain_rbuf;
      });
   }
}

=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 ($len, $cb) = @_;

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

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

   $self->push_read (_read_chunk $len, $cb);
}


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

   $self->unshift_read (_read_chunk $len, $cb);
}

=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 $cb = pop;
   my $eol = @_ ? shift : qr|(\015?\012)|;
   my $pos;

   $eol = qr|(\Q$eol\E)| unless ref $eol;
   $eol = qr|^(.*?)($eol)|;

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

      $cb->($1, $2);
      1
   }
}

sub push_read_line {
   my $self = shift;

   $self->push_read (&_read_line);
}

sub unshift_read_line {
   my $self = shift;

   $self->unshift_read (&_read_line);
}

=back

=head1 AUTHOR

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

=cut

1; # End of AnyEvent::Handle
Revision:	1.9
Committed:	Fri May 2 16:07:46 2008 UTC (16 years, 1 month ago) by root
Branch:	MAIN
Changes since 1.8:	+93 -2 lines
Log Message:	* empty log message *
#	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	=cut
18
19	our $VERSION = '0.02';
20
21	=head1 SYNOPSIS
22
23	use AnyEvent;
24	use AnyEvent::Handle;
25
26	my $cv = AnyEvent->condvar;
27
28	my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);
29
30	#TODO
31
32	# or use the constructor to pass the callback:
33
34	my $ae_fh2 =
35	AnyEvent::Handle->new (
36	fh => \*STDIN,
37	on_eof => sub {
38	$cv->broadcast;
39	},
40	#TODO
41	);
42
43	$cv->wait;
44
45	=head1 DESCRIPTION
46
47	This module is a helper module to make it easier to do event-based I/O on
48	filehandles (and sockets, see L<AnyEvent::Socket> for an easy way to make
49	non-blocking resolves and connects).
50
51	In the following, when the documentation refers to of "bytes" then this
52	means characters. As sysread and syswrite are used for all I/O, their
53	treatment of characters applies to this module as well.
54
55	All callbacks will be invoked with the handle object as their first
56	argument.
57
58	=head1 METHODS
59
60	=over 4
61
62	=item B<new (%args)>
63
64	The constructor supports these arguments (all as key => value pairs).
65
66	=over 4
67
68	=item fh => $filehandle [MANDATORY]
69
70	The filehandle this L<AnyEvent::Handle> object will operate on.
71
72	NOTE: The filehandle will be set to non-blocking (using
73	AnyEvent::Util::fh_nonblocking).
74
75	=item on_error => $cb->($self) [MANDATORY]
76
77	This is the fatal error callback, that is called when a fatal error ocurs,
78	such as not being able to resolve the hostname, failure to connect or a
79	read error.
80
81	The object will not be in a usable state when this callback has been
82	called.
83
84	On callback entrance, the value of C<$!> contains the opertaing system
85	error (or C<ENOSPC> or C<EPIPE>).
86
87	=item on_eof => $cb->($self) [MANDATORY]
88
89	Set the callback to be called on EOF.
90
91	=item on_read => $cb->($self)
92
93	This sets the default read callback, which is called when data arrives
94	and no read request is in the queue. If the read callback is C<undef>
95	or has never been set, than AnyEvent::Handle will cease reading from the
96	filehandle.
97
98	To access (and remove data from) the read buffer, use the C<< ->rbuf >>
99	method or acces sthe C<$self->{rbuf}> member directly.
100
101	When an EOF condition is detected then AnyEvent::Handle will first try to
102	feed all the remaining data to the queued callbacks and C<on_read> before
103	calling the C<on_eof> callback. If no progress can be made, then a fatal
104	error will be raised (with C<$!> set to C<EPIPE>).
105
106	=item on_drain => $cb->()
107
108	This sets the callback that is called when the write buffer becomes empty
109	(or when the callback is set and the buffer is empty already).
110
111	To append to the write buffer, use the C<< ->push_write >> method.
112
113	=item rbuf_max => <bytes>
114
115	If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
116	when the read buffer ever (strictly) exceeds this size. This is useful to
117	avoid denial-of-service attacks.
118
119	For example, a server accepting connections from untrusted sources should
120	be configured to accept only so-and-so much data that it cannot act on
121	(for example, when expecting a line, an attacker could send an unlimited
122	amount of data without a callback ever being called as long as the line
123	isn't finished).
124
125	=item read_size => <bytes>
126
127	The default read block size (the amount of bytes this module will try to read
128	on each [loop iteration). Default: C<4096>.
129
130	=item low_water_mark => <bytes>
131
132	Sets the amount of bytes (default: C<0>) that make up an "empty" write
133	buffer: If the write reaches this size or gets even samller it is
134	considered empty.
135
136	=back
137
138	=cut
139
140	sub new {
141	my $class = shift;
142
143	my $self = bless { @_ }, $class;
144
145	$self->{fh} or Carp::croak "mandatory argument fh is missing";
146
147	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
148
149	$self->on_error ((delete $self->{on_error}) or Carp::croak "mandatory argument on_error is missing");
150	$self->on_eof ((delete $self->{on_eof} ) or Carp::croak "mandatory argument on_eof is missing");
151
152	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
153	$self->on_read (delete $self->{on_read} ) if $self->{on_read};
154
155	$self
156	}
157
158	sub _shutdown {
159	my ($self) = @_;
160
161	delete $self->{rw};
162	delete $self->{ww};
163	delete $self->{fh};
164	}
165
166	sub error {
167	my ($self) = @_;
168
169	{
170	local $!;
171	$self->_shutdown;
172	}
173
174	$self->{on_error}($self);
175	}
176
177	=item $fh = $handle->fh
178
179	This method returns the filehandle of the L<AnyEvent::Handle> object.
180
181	=cut
182
183	sub fh { $_[0]->{fh} }
184
185	=item $handle->on_error ($cb)
186
187	Replace the current C<on_error> callback (see the C<on_error> constructor argument).
188
189	=cut
190
191	sub on_error {
192	$_[0]{on_error} = $_[1];
193	}
194
195	=item $handle->on_eof ($cb)
196
197	Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
198
199	=cut
200
201	sub on_eof {
202	$_[0]{on_eof} = $_[1];
203	}
204
205	#############################################################################
206
207	=back
208
209	=head2 WRITE QUEUE
210
211	AnyEvent::Handle manages two queues per handle, one for writing and one
212	for reading.
213
214	The write queue is very simple: you can add data to its end, and
215	AnyEvent::Handle will automatically try to get rid of it for you.
216
217	When data could be writtena nd the write buffer is shorter then the low
218	water mark, the C<on_drain> callback will be invoked.
219
220	=over 4
221
222	=item $handle->on_drain ($cb)
223
224	Sets the C<on_drain> callback or clears it (see the description of
225	C<on_drain> in the constructor).
226
227	=cut
228
229	sub on_drain {
230	my ($self, $cb) = @_;
231
232	$self->{on_drain} = $cb;
233
234	$cb->($self)
235	if $cb && $self->{low_water_mark} >= length $self->{wbuf};
236	}
237
238	=item $handle->push_write ($data)
239
240	Queues the given scalar to be written. You can push as much data as you
241	want (only limited by the available memory), as C<AnyEvent::Handle>
242	buffers it independently of the kernel.
243
244	=cut
245
246	sub push_write {
247	my ($self, $data) = @_;
248
249	$self->{wbuf} .= $data;
250
251	unless ($self->{ww}) {
252	Scalar::Util::weaken $self;
253	my $cb = sub {
254	my $len = syswrite $self->{fh}, $self->{wbuf};
255
256	if ($len > 0) {
257	substr $self->{wbuf}, 0, $len, "";
258
259
260	$self->{on_drain}($self)
261	if $self->{low_water_mark} >= length $self->{wbuf}
262	&& $self->{on_drain};
263
264	delete $self->{ww} unless length $self->{wbuf};
265	} elsif ($! != EAGAIN && $! != EINTR) {
266	$self->error;
267	}
268	};
269
270	$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
271
272	$cb->($self);
273	};
274	}
275
276	#############################################################################
277
278	=back
279
280	=head2 READ QUEUE
281
282	AnyEvent::Handle manages two queues per handle, one for writing and one
283	for reading.
284
285	The read queue is more complex than the write queue. It can be used in two
286	ways, the "simple" way, using only C<on_read> and the "complex" way, using
287	a queue.
288
289	In the simple case, you just install an C<on_read> callback and whenever
290	new data arrives, it will be called. You can then remove some data (if
291	enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
292	or not.
293
294	In the more complex case, you want to queue multiple callbacks. In this
295	case, AnyEvent::Handle will call the first queued callback each time new
296	data arrives and removes it when it has done its job (see C<push_read>,
297	below).
298
299	This way you can, for example, push three line-reads, followed by reading
300	a chunk of data, and AnyEvent::Handle will execute them in order.
301
302	Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
303	the specified number of bytes which give an XML datagram.
304
305	# in the default state, expect some header bytes
306	$handle->on_read (sub {
307	# some data is here, now queue the length-header-read (4 octets)
308	shift->unshift_read_chunk (4, sub {
309	# header arrived, decode
310	my $len = unpack "N", $_[1];
311
312	# now read the payload
313	shift->unshift_read_chunk ($len, sub {
314	my $xml = $_[1];
315	# handle xml
316	});
317	});
318	});
319
320	Example 2: Implement a client for a protocol that replies either with
321	"OK" and another line or "ERROR" for one request, and 64 bytes for the
322	second request. Due tot he availability of a full queue, we can just
323	pipeline sending both requests and manipulate the queue as necessary in
324	the callbacks:
325
326	# request one
327	$handle->push_write ("request 1\015\012");
328
329	# we expect "ERROR" or "OK" as response, so push a line read
330	$handle->push_read_line (sub {
331	# if we got an "OK", we have to _prepend_ another line,
332	# so it will be read before the second request reads its 64 bytes
333	# which are already in the queue when this callback is called
334	# we don't do this in case we got an error
335	if ($_[1] eq "OK") {
336	$_[0]->unshift_read_line (sub {
337	my $response = $_[1];
338	...
339	});
340	}
341	});
342
343	# request two
344	$handle->push_write ("request 2\015\012");
345
346	# simply read 64 bytes, always
347	$handle->push_read_chunk (64, sub {
348	my $response = $_[1];
349	...
350	});
351
352	=over 4
353
354	sub _drain_rbuf {
355	my ($self) = @_;
356
357	return if exists $self->{in_drain};
358	local $self->{in_drain} = 1;
359
360	while (my $len = length $self->{rbuf}) {
361	no strict 'refs';
362	if (@{ $self->{queue} }) {
363	if ($self->{queue}[0]($self)) {
364	shift @{ $self->{queue} };
365	} elsif ($self->{eof}) {
366	# no progress can be made (not enough data and no data forthcoming)
367	$! = &Errno::EPIPE; return $self->error;
368	} else {
369	return;
370	}
371	} elsif ($self->{on_read}) {
372	$self->{on_read}($self);
373
374	if (
375	$self->{eof} # if no further data will arrive
376	&& $len == length $self->{rbuf} # and no data has been consumed
377	&& !@{ $self->{queue} } # and the queue is still empty
378	&& $self->{on_read} # and we still want to read data
379	) {
380	# then no progress can be made
381	$! = &Errno::EPIPE; return $self->error;
382	}
383	} else {
384	# read side becomes idle
385	delete $self->{rw};
386	return;
387	}
388	}
389
390	if ($self->{eof}) {
391	$self->_shutdown;
392	$self->{on_eof}($self);
393	}
394	}
395
396	=item $handle->on_read ($cb)
397
398	This replaces the currently set C<on_read> callback, or clears it (when
399	the new callback is C<undef>). See the description of C<on_read> in the
400	constructor.
401
402	=cut
403
404	sub on_read {
405	my ($self, $cb) = @_;
406
407	$self->{on_read} = $cb;
408
409	unless ($self->{rw} \|\| $self->{eof}) {
410	Scalar::Util::weaken $self;
411
412	$self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
413	my $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} \|\| 8192, length $self->{rbuf};
414
415	if ($len > 0) {
416	if (exists $self->{rbuf_max}) {
417	if ($self->{rbuf_max} < length $self->{rbuf}) {
418	$! = &Errno::ENOSPC; return $self->error;
419	}
420	}
421
422	} elsif (defined $len) {
423	$self->{eof} = 1;
424	delete $self->{rw};
425
426	} elsif ($! != EAGAIN && $! != EINTR) {
427	return $self->error;
428	}
429
430	$self->_drain_rbuf;
431	});
432	}
433	}
434
435	=item $handle->rbuf
436
437	Returns the read buffer (as a modifiable lvalue).
438
439	You can access the read buffer directly as the C<< ->{rbuf} >> member, if
440	you want.
441
442	NOTE: The read buffer should only be used or modified if the C<on_read>,
443	C<push_read> or C<unshift_read> methods are used. The other read methods
444	automatically manage the read buffer.
445
446	=cut
447
448	sub rbuf : lvalue {
449	$_[0]{rbuf}
450	}
451
452	=item $handle->push_read ($cb)
453
454	=item $handle->unshift_read ($cb)
455
456	Append the given callback to the end of the queue (C<push_read>) or
457	prepend it (C<unshift_read>).
458
459	The callback is called each time some additional read data arrives.
460
461	It must check wether enough data is in the read buffer already.
462
463	If not enough data is available, it must return the empty list or a false
464	value, in which case it will be called repeatedly until enough data is
465	available (or an error condition is detected).
466
467	If enough data was available, then the callback must remove all data it is
468	interested in (which can be none at all) and return a true value. After returning
469	true, it will be removed from the queue.
470
471	=cut
472
473	sub push_read {
474	my ($self, $cb) = @_;
475
476	push @{ $self->{queue} }, $cb;
477	$self->_drain_rbuf;
478	}
479
480	sub unshift_read {
481	my ($self, $cb) = @_;
482
483	push @{ $self->{queue} }, $cb;
484	$self->_drain_rbuf;
485	}
486
487	=item $handle->push_read_chunk ($len, $cb->($self, $data))
488
489	=item $handle->unshift_read_chunk ($len, $cb->($self, $data))
490
491	Append the given callback to the end of the queue (C<push_read_chunk>) or
492	prepend it (C<unshift_read_chunk>).
493
494	The callback will be called only once C<$len> bytes have been read, and
495	these C<$len> bytes will be passed to the callback.
496
497	=cut
498
499	sub _read_chunk($$) {
500	my ($len, $cb) = @_;
501
502	sub {
503	$len <= length $_[0]{rbuf} or return;
504	$cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
505	1
506	}
507	}
508
509	sub push_read_chunk {
510	my ($self, $len, $cb) = @_;
511
512	$self->push_read (_read_chunk $len, $cb);
513	}
514
515
516	sub unshift_read_chunk {
517	my ($self, $len, $cb) = @_;
518
519	$self->unshift_read (_read_chunk $len, $cb);
520	}
521
522	=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
523
524	=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
525
526	Append the given callback to the end of the queue (C<push_read_line>) or
527	prepend it (C<unshift_read_line>).
528
529	The callback will be called only once a full line (including the end of
530	line marker, C<$eol>) has been read. This line (excluding the end of line
531	marker) will be passed to the callback as second argument (C<$line>), and
532	the end of line marker as the third argument (C<$eol>).
533
534	The end of line marker, C<$eol>, can be either a string, in which case it
535	will be interpreted as a fixed record end marker, or it can be a regex
536	object (e.g. created by C<qr>), in which case it is interpreted as a
537	regular expression.
538
539	The end of line marker argument C<$eol> is optional, if it is missing (NOT
540	undef), then C<qr\|\015?\012\|> is used (which is good for most internet
541	protocols).
542
543	Partial lines at the end of the stream will never be returned, as they are
544	not marked by the end of line marker.
545
546	=cut
547
548	sub _read_line($$) {
549	my $cb = pop;
550	my $eol = @_ ? shift : qr\|(\015?\012)\|;
551	my $pos;
552
553	$eol = qr\|(\Q$eol\E)\| unless ref $eol;
554	$eol = qr\|^(.*?)($eol)\|;
555
556	sub {
557	$_[0]{rbuf} =~ s/$eol// or return;
558
559	$cb->($1, $2);
560	1
561	}
562	}
563
564	sub push_read_line {
565	my $self = shift;
566
567	$self->push_read (&_read_line);
568	}
569
570	sub unshift_read_line {
571	my $self = shift;
572
573	$self->unshift_read (&_read_line);
574	}
575
576	=back
577
578	=head1 AUTHOR
579
580	Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
581
582	=cut
583
584	1; # End of AnyEvent::Handle