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.

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

=cut

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

   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 form 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 $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} || 8192, length $self->{rbuf};

         if ($len > 0) {
            if (defined $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;
      });
   }
}

=back

=head1 AUTHOR

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

=cut

1; # End of AnyEvent::Handle
Revision:	1.16
Committed:	Fri May 23 05:16:57 2008 UTC (16 years ago) by root
Branch:	MAIN
Changes since 1.15:	+8 -4 lines
Log Message:	implement vc 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	=back
145
146	=cut
147
148	sub new {
149	my $class = shift;
150
151	my $self = bless { @_ }, $class;
152
153	$self->{fh} or Carp::croak "mandatory argument fh is missing";
154
155	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
156
157	$self->on_eof (delete $self->{on_eof} ) if $self->{on_eof};
158	$self->on_error (delete $self->{on_error}) if $self->{on_error};
159	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
160	$self->on_read (delete $self->{on_read} ) if $self->{on_read};
161
162	$self->start_read;
163
164	$self
165	}
166
167	sub _shutdown {
168	my ($self) = @_;
169
170	delete $self->{rw};
171	delete $self->{ww};
172	delete $self->{fh};
173	}
174
175	sub error {
176	my ($self) = @_;
177
178	{
179	local $!;
180	$self->_shutdown;
181	}
182
183	if ($self->{on_error}) {
184	$self->{on_error}($self);
185	} else {
186	die "AnyEvent::Handle uncaught fatal error: $!";
187	}
188	}
189
190	=item $fh = $handle->fh
191
192	This method returns the filehandle of the L<AnyEvent::Handle> object.
193
194	=cut
195
196	sub fh { $_[0]->{fh} }
197
198	=item $handle->on_error ($cb)
199
200	Replace the current C<on_error> callback (see the C<on_error> constructor argument).
201
202	=cut
203
204	sub on_error {
205	$_[0]{on_error} = $_[1];
206	}
207
208	=item $handle->on_eof ($cb)
209
210	Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
211
212	=cut
213
214	sub on_eof {
215	$_[0]{on_eof} = $_[1];
216	}
217
218	#############################################################################
219
220	=back
221
222	=head2 WRITE QUEUE
223
224	AnyEvent::Handle manages two queues per handle, one for writing and one
225	for reading.
226
227	The write queue is very simple: you can add data to its end, and
228	AnyEvent::Handle will automatically try to get rid of it for you.
229
230	When data could be writtena nd the write buffer is shorter then the low
231	water mark, the C<on_drain> callback will be invoked.
232
233	=over 4
234
235	=item $handle->on_drain ($cb)
236
237	Sets the C<on_drain> callback or clears it (see the description of
238	C<on_drain> in the constructor).
239
240	=cut
241
242	sub on_drain {
243	my ($self, $cb) = @_;
244
245	$self->{on_drain} = $cb;
246
247	$cb->($self)
248	if $cb && $self->{low_water_mark} >= length $self->{wbuf};
249	}
250
251	=item $handle->push_write ($data)
252
253	Queues the given scalar to be written. You can push as much data as you
254	want (only limited by the available memory), as C<AnyEvent::Handle>
255	buffers it independently of the kernel.
256
257	=cut
258
259	sub push_write {
260	my ($self, $data) = @_;
261
262	$self->{wbuf} .= $data;
263
264	unless ($self->{ww}) {
265	Scalar::Util::weaken $self;
266	my $cb = sub {
267	my $len = syswrite $self->{fh}, $self->{wbuf};
268
269	if ($len > 0) {
270	substr $self->{wbuf}, 0, $len, "";
271
272
273	$self->{on_drain}($self)
274	if $self->{low_water_mark} >= length $self->{wbuf}
275	&& $self->{on_drain};
276
277	delete $self->{ww} unless length $self->{wbuf};
278	} elsif ($! != EAGAIN && $! != EINTR) {
279	$self->error;
280	}
281	};
282
283	$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
284
285	$cb->($self);
286	};
287	}
288
289	#############################################################################
290
291	=back
292
293	=head2 READ QUEUE
294
295	AnyEvent::Handle manages two queues per handle, one for writing and one
296	for reading.
297
298	The read queue is more complex than the write queue. It can be used in two
299	ways, the "simple" way, using only C<on_read> and the "complex" way, using
300	a queue.
301
302	In the simple case, you just install an C<on_read> callback and whenever
303	new data arrives, it will be called. You can then remove some data (if
304	enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
305	or not.
306
307	In the more complex case, you want to queue multiple callbacks. In this
308	case, AnyEvent::Handle will call the first queued callback each time new
309	data arrives and removes it when it has done its job (see C<push_read>,
310	below).
311
312	This way you can, for example, push three line-reads, followed by reading
313	a chunk of data, and AnyEvent::Handle will execute them in order.
314
315	Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
316	the specified number of bytes which give an XML datagram.
317
318	# in the default state, expect some header bytes
319	$handle->on_read (sub {
320	# some data is here, now queue the length-header-read (4 octets)
321	shift->unshift_read_chunk (4, sub {
322	# header arrived, decode
323	my $len = unpack "N", $_[1];
324
325	# now read the payload
326	shift->unshift_read_chunk ($len, sub {
327	my $xml = $_[1];
328	# handle xml
329	});
330	});
331	});
332
333	Example 2: Implement a client for a protocol that replies either with
334	"OK" and another line or "ERROR" for one request, and 64 bytes for the
335	second request. Due tot he availability of a full queue, we can just
336	pipeline sending both requests and manipulate the queue as necessary in
337	the callbacks:
338
339	# request one
340	$handle->push_write ("request 1\015\012");
341
342	# we expect "ERROR" or "OK" as response, so push a line read
343	$handle->push_read_line (sub {
344	# if we got an "OK", we have to _prepend_ another line,
345	# so it will be read before the second request reads its 64 bytes
346	# which are already in the queue when this callback is called
347	# we don't do this in case we got an error
348	if ($_[1] eq "OK") {
349	$_[0]->unshift_read_line (sub {
350	my $response = $_[1];
351	...
352	});
353	}
354	});
355
356	# request two
357	$handle->push_write ("request 2\015\012");
358
359	# simply read 64 bytes, always
360	$handle->push_read_chunk (64, sub {
361	my $response = $_[1];
362	...
363	});
364
365	=over 4
366
367	=cut
368
369	sub _drain_rbuf {
370	my ($self) = @_;
371
372	return if $self->{in_drain};
373	local $self->{in_drain} = 1;
374
375	while (my $len = length $self->{rbuf}) {
376	no strict 'refs';
377	if (my $cb = shift @{ $self->{queue} }) {
378	if (!$cb->($self)) {
379	if ($self->{eof}) {
380	# no progress can be made (not enough data and no data forthcoming)
381	$! = &Errno::EPIPE; return $self->error;
382	}
383
384	unshift @{ $self->{queue} }, $cb;
385	return;
386	}
387	} elsif ($self->{on_read}) {
388	$self->{on_read}($self);
389
390	if (
391	$self->{eof} # if no further data will arrive
392	&& $len == length $self->{rbuf} # and no data has been consumed
393	&& !@{ $self->{queue} } # and the queue is still empty
394	&& $self->{on_read} # and we still want to read data
395	) {
396	# then no progress can be made
397	$! = &Errno::EPIPE; return $self->error;
398	}
399	} else {
400	# read side becomes idle
401	delete $self->{rw};
402	return;
403	}
404	}
405
406	if ($self->{eof}) {
407	$self->_shutdown;
408	$self->{on_eof}($self)
409	if $self->{on_eof};
410	}
411	}
412
413	=item $handle->on_read ($cb)
414
415	This replaces the currently set C<on_read> callback, or clears it (when
416	the new callback is C<undef>). See the description of C<on_read> in the
417	constructor.
418
419	=cut
420
421	sub on_read {
422	my ($self, $cb) = @_;
423
424	$self->{on_read} = $cb;
425	}
426
427	=item $handle->rbuf
428
429	Returns the read buffer (as a modifiable lvalue).
430
431	You can access the read buffer directly as the C<< ->{rbuf} >> member, if
432	you want.
433
434	NOTE: The read buffer should only be used or modified if the C<on_read>,
435	C<push_read> or C<unshift_read> methods are used. The other read methods
436	automatically manage the read buffer.
437
438	=cut
439
440	sub rbuf : lvalue {
441	$_[0]{rbuf}
442	}
443
444	=item $handle->push_read ($cb)
445
446	=item $handle->unshift_read ($cb)
447
448	Append the given callback to the end of the queue (C<push_read>) or
449	prepend it (C<unshift_read>).
450
451	The callback is called each time some additional read data arrives.
452
453	It must check wether enough data is in the read buffer already.
454
455	If not enough data is available, it must return the empty list or a false
456	value, in which case it will be called repeatedly until enough data is
457	available (or an error condition is detected).
458
459	If enough data was available, then the callback must remove all data it is
460	interested in (which can be none at all) and return a true value. After returning
461	true, it will be removed from the queue.
462
463	=cut
464
465	sub push_read {
466	my ($self, $cb) = @_;
467
468	push @{ $self->{queue} }, $cb;
469	$self->_drain_rbuf;
470	}
471
472	sub unshift_read {
473	my ($self, $cb) = @_;
474
475	push @{ $self->{queue} }, $cb;
476	$self->_drain_rbuf;
477	}
478
479	=item $handle->push_read_chunk ($len, $cb->($self, $data))
480
481	=item $handle->unshift_read_chunk ($len, $cb->($self, $data))
482
483	Append the given callback to the end of the queue (C<push_read_chunk>) or
484	prepend it (C<unshift_read_chunk>).
485
486	The callback will be called only once C<$len> bytes have been read, and
487	these C<$len> bytes will be passed to the callback.
488
489	=cut
490
491	sub _read_chunk($$) {
492	my ($self, $len, $cb) = @_;
493
494	sub {
495	$len <= length $_[0]{rbuf} or return;
496	$cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
497	1
498	}
499	}
500
501	sub push_read_chunk {
502	$_[0]->push_read (&_read_chunk);
503	}
504
505
506	sub unshift_read_chunk {
507	$_[0]->unshift_read (&_read_chunk);
508	}
509
510	=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
511
512	=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
513
514	Append the given callback to the end of the queue (C<push_read_line>) or
515	prepend it (C<unshift_read_line>).
516
517	The callback will be called only once a full line (including the end of
518	line marker, C<$eol>) has been read. This line (excluding the end of line
519	marker) will be passed to the callback as second argument (C<$line>), and
520	the end of line marker as the third argument (C<$eol>).
521
522	The end of line marker, C<$eol>, can be either a string, in which case it
523	will be interpreted as a fixed record end marker, or it can be a regex
524	object (e.g. created by C<qr>), in which case it is interpreted as a
525	regular expression.
526
527	The end of line marker argument C<$eol> is optional, if it is missing (NOT
528	undef), then C<qr\|\015?\012\|> is used (which is good for most internet
529	protocols).
530
531	Partial lines at the end of the stream will never be returned, as they are
532	not marked by the end of line marker.
533
534	=cut
535
536	sub _read_line($$) {
537	my $self = shift;
538	my $cb = pop;
539	my $eol = @_ ? shift : qr\|(\015?\012)\|;
540	my $pos;
541
542	$eol = quotemeta $eol unless ref $eol;
543	$eol = qr\|^(.*?)($eol)\|s;
544
545	sub {
546	$_[0]{rbuf} =~ s/$eol// or return;
547
548	$cb->($_[0], $1, $2);
549	1
550	}
551	}
552
553	sub push_read_line {
554	$_[0]->push_read (&_read_line);
555	}
556
557	sub unshift_read_line {
558	$_[0]->unshift_read (&_read_line);
559	}
560
561	=item $handle->stop_read
562
563	=item $handle->start_read
564
565	In rare cases you actually do not want to read anything form the
566	socket. In this case you can call C<stop_read>. Neither C<on_read> no
567	any queued callbacks will be executed then. To start readign again, call
568	C<start_read>.
569
570	=cut
571
572	sub stop_read {
573	my ($self) = @_;
574
575	delete $self->{rw};
576	}
577
578	sub start_read {
579	my ($self) = @_;
580
581	unless ($self->{rw} \|\| $self->{eof}) {
582	Scalar::Util::weaken $self;
583
584	$self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
585	my $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} \|\| 8192, length $self->{rbuf};
586
587	if ($len > 0) {
588	if (defined $self->{rbuf_max}) {
589	if ($self->{rbuf_max} < length $self->{rbuf}) {
590	$! = &Errno::ENOSPC; return $self->error;
591	}
592	}
593
594	} elsif (defined $len) {
595	$self->{eof} = 1;
596	delete $self->{rw};
597
598	} elsif ($! != EAGAIN && $! != EINTR) {
599	return $self->error;
600	}
601
602	$self->_drain_rbuf;
603	});
604	}
605	}
606
607	=back
608
609	=head1 AUTHOR
610
611	Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
612
613	=cut
614
615	1; # End of AnyEvent::Handle