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

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

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

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.8
Committed:	Fri May 2 15:36:10 2008 UTC (16 years ago) by root
Branch:	MAIN
Changes since 1.7:	+332 -213 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	elmex	1.1	package AnyEvent::Handle;
2
3	elmex	1.6	no warnings;
4	elmex	1.1	use strict;
5
6	root	1.8	use AnyEvent ();
7			use AnyEvent::Util ();
8			use Scalar::Util ();
9			use Carp ();
10			use Fcntl ();
11	elmex	1.1	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	root	1.8	our $VERSION = '0.02';
20	elmex	1.1
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	root	1.8	#TODO
31	elmex	1.1
32	elmex	1.2	# 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	root	1.8	#TODO
41	elmex	1.2	);
42
43	elmex	1.1	$cv->wait;
44
45			=head1 DESCRIPTION
46
47	root	1.8	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	elmex	1.1
55	root	1.8	All callbacks will be invoked with the handle object as their first
56			argument.
57	elmex	1.1
58			=head1 METHODS
59
60			=over 4
61
62			=item B<new (%args)>
63
64	root	1.8	The constructor supports these arguments (all as key => value pairs).
65	elmex	1.1
66			=over 4
67
68	root	1.8	=item fh => $filehandle [MANDATORY]
69	elmex	1.1
70			The filehandle this L<AnyEvent::Handle> object will operate on.
71
72	root	1.8	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	elmex	1.1
106	root	1.8	=item on_drain => $cb->()
107	elmex	1.1
108	root	1.8	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	elmex	1.1
111	root	1.8	To append to the write buffer, use the C<< ->push_write >> method.
112	elmex	1.2
113	root	1.8	=item rbuf_max => <bytes>
114	elmex	1.2
115	root	1.8	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	elmex	1.2
119	root	1.8	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	elmex	1.2
125	root	1.8	=item read_size => <bytes>
126	elmex	1.2
127	root	1.8	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	elmex	1.2
136	elmex	1.1	=back
137
138			=cut
139
140			sub new {
141	root	1.8	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	elmex	1.1
149	root	1.8	$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	elmex	1.1
152	root	1.8	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
153			$self->on_read (delete $self->{on_read} ) if $self->{on_read};
154	elmex	1.1
155	root	1.8	$self
156			}
157	elmex	1.2
158	root	1.8	sub _shutdown {
159			my ($self) = @_;
160	elmex	1.2
161	root	1.8	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	elmex	1.1	}
173
174	root	1.8	$self->{on_error}($self);
175	elmex	1.1	}
176
177	root	1.8	=item $fh = $handle->fh
178	elmex	1.1
179			This method returns the filehandle of the L<AnyEvent::Handle> object.
180
181			=cut
182
183			sub fh { $_[0]->{fh} }
184
185	root	1.8	=item $handle->on_error ($cb)
186	elmex	1.1
187	root	1.8	Replace the current C<on_error> callback (see the C<on_error> constructor argument).
188	elmex	1.1
189	root	1.8	=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	elmex	1.1
199			=cut
200
201	root	1.8	#############################################################################
202
203			sub on_eof {
204			$_[0]{on_eof} = $_[1];
205			}
206
207			=item $handle->on_drain ($cb)
208
209			Sets the C<on_drain> callback or clears it (see the description of
210			C<on_drain> in the constructor).
211
212			=cut
213
214			sub on_drain {
215	elmex	1.1	my ($self, $cb) = @_;
216
217	root	1.8	$self->{on_drain} = $cb;
218
219			$cb->($self)
220			if $cb && $self->{low_water_mark} >= length $self->{wbuf};
221			}
222
223			=item $handle->push_write ($data)
224
225			Queues the given scalar to be written. You can push as much data as you
226			want (only limited by the available memory), as C<AnyEvent::Handle>
227			buffers it independently of the kernel.
228
229			=cut
230
231			sub push_write {
232			my ($self, $data) = @_;
233
234			$self->{wbuf} .= $data;
235
236			unless ($self->{ww}) {
237			Scalar::Util::weaken $self;
238			my $cb = sub {
239			my $len = syswrite $self->{fh}, $self->{wbuf};
240
241			if ($len > 0) {
242			substr $self->{wbuf}, 0, $len, "";
243
244
245			$self->{on_drain}($self)
246			if $self->{low_water_mark} >= length $self->{wbuf}
247			&& $self->{on_drain};
248
249			delete $self->{ww} unless length $self->{wbuf};
250			} elsif ($! != EAGAIN && $! != EINTR) {
251			$self->error;
252	elmex	1.1	}
253	root	1.8	};
254
255			$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
256
257			$cb->($self);
258			};
259			}
260
261			#############################################################################
262
263			sub _drain_rbuf {
264			my ($self) = @_;
265	elmex	1.1
266	root	1.8	return if exists $self->{in_drain};
267			local $self->{in_drain} = 1;
268	elmex	1.1
269	root	1.8	while (my $len = length $self->{rbuf}) {
270			no strict 'refs';
271			if (@{ $self->{queue} }) {
272			if ($self->{queue}[0]($self)) {
273			shift @{ $self->{queue} };
274			} elsif ($self->{eof}) {
275			# no progress can be made (not enough data and no data forthcoming)
276			$! = &Errno::EPIPE; return $self->error;
277	elmex	1.1	} else {
278	root	1.8	return;
279			}
280			} elsif ($self->{on_read}) {
281			$self->{on_read}($self);
282
283			if (
284			$self->{eof} # if no further data will arrive
285			&& $len == length $self->{rbuf} # and no data has been consumed
286			&& !@{ $self->{queue} } # and the queue is still empty
287			&& $self->{on_read} # and we still want to read data
288			) {
289			# then no progress can be made
290			$! = &Errno::EPIPE; return $self->error;
291	elmex	1.1	}
292	root	1.8	} else {
293			# read side becomes idle
294			delete $self->{rw};
295			return;
296			}
297			}
298
299			if ($self->{eof}) {
300			$self->_shutdown;
301			$self->{on_eof}($self);
302			}
303	elmex	1.1	}
304
305	root	1.8	=item $handle->on_read ($cb)
306	elmex	1.1
307	root	1.8	This replaces the currently set C<on_read> callback, or clears it (when
308			the new callback is C<undef>). See the description of C<on_read> in the
309			constructor.
310	elmex	1.1
311	root	1.8	=cut
312
313			sub on_read {
314			my ($self, $cb) = @_;
315	elmex	1.1
316	root	1.8	$self->{on_read} = $cb;
317
318			unless ($self->{rw} \|\| $self->{eof}) {
319			Scalar::Util::weaken $self;
320	elmex	1.1
321	root	1.8	$self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
322			my $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} \|\| 8192, length $self->{rbuf};
323	elmex	1.1
324	root	1.8	if ($len > 0) {
325			if (exists $self->{rbuf_max}) {
326			if ($self->{rbuf_max} < length $self->{rbuf}) {
327			$! = &Errno::ENOSPC; return $self->error;
328			}
329			}
330	elmex	1.1
331	root	1.8	} elsif (defined $len) {
332			$self->{eof} = 1;
333			delete $self->{rw};
334	elmex	1.1
335	root	1.8	} elsif ($! != EAGAIN && $! != EINTR) {
336			return $self->error;
337			}
338	elmex	1.1
339	root	1.8	$self->_drain_rbuf;
340			});
341			}
342	elmex	1.1	}
343
344	root	1.8	=item $handle->rbuf
345
346			Returns the read buffer (as a modifiable lvalue).
347	elmex	1.1
348	root	1.8	You can access the read buffer directly as the C<< ->{rbuf} >> member, if
349			you want.
350	elmex	1.1
351	root	1.8	NOTE: The read buffer should only be used or modified if the C<on_read>,
352			C<push_read> or C<unshift_read> methods are used. The other read methods
353			automatically manage the read buffer.
354	elmex	1.1
355			=cut
356
357	elmex	1.2	sub rbuf : lvalue {
358	root	1.8	$_[0]{rbuf}
359	elmex	1.2	}
360	elmex	1.1
361	root	1.8	=item $handle->push_read ($cb)
362
363			=item $handle->unshift_read ($cb)
364
365			Append the given callback to the end of the queue (C<push_read>) or
366			prepend it (C<unshift_read>).
367
368			The callback is called each time some additional read data arrives.
369	elmex	1.1
370	root	1.8	It must check wether enough data is in the read buffer already.
371	elmex	1.1
372	root	1.8	If not enough data is available, it must return the empty list or a false
373			value, in which case it will be called repeatedly until enough data is
374			available (or an error condition is detected).
375
376			If enough data was available, then the callback must remove all data it is
377			interested in (which can be none at all) and return a true value. After returning
378			true, it will be removed from the queue.
379	elmex	1.1
380			=cut
381
382	root	1.8	sub push_read {
383			my ($self, $cb) = @_;
384	elmex	1.1
385	root	1.8	push @{ $self->{queue} }, $cb;
386			$self->_drain_rbuf;
387	elmex	1.1	}
388
389	root	1.8	sub unshift_read {
390			my ($self, $cb) = @_;
391
392			push @{ $self->{queue} }, $cb;
393			$self->_drain_rbuf;
394			}
395	elmex	1.1
396	root	1.8	=item $handle->push_read_chunk ($len, $cb->($self, $data))
397	elmex	1.1
398	root	1.8	=item $handle->unshift_read_chunk ($len, $cb->($self, $data))
399	elmex	1.1
400	root	1.8	Append the given callback to the end of the queue (C<push_read_chunk>) or
401			prepend it (C<unshift_read_chunk>).
402	elmex	1.1
403	root	1.8	The callback will be called only once C<$len> bytes have been read, and
404			these C<$len> bytes will be passed to the callback.
405	elmex	1.1
406			=cut
407
408	root	1.8	sub _read_chunk($$) {
409			my ($len, $cb) = @_;
410	elmex	1.1
411	root	1.8	sub {
412			$len <= length $_[0]{rbuf} or return;
413			$cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
414			1
415			}
416			}
417
418			sub push_read_chunk {
419			my ($self, $len, $cb) = @_;
420	elmex	1.5
421	root	1.8	$self->push_read (_read_chunk $len, $cb);
422			}
423	elmex	1.1
424	elmex	1.5
425	root	1.8	sub unshift_read_chunk {
426			my ($self, $len, $cb) = @_;
427	elmex	1.1
428	root	1.8	$self->unshift_read (_read_chunk $len, $cb);
429	elmex	1.1	}
430
431	root	1.8	=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
432	elmex	1.1
433	root	1.8	=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
434	elmex	1.1
435	root	1.8	Append the given callback to the end of the queue (C<push_read_line>) or
436			prepend it (C<unshift_read_line>).
437	elmex	1.1
438	root	1.8	The callback will be called only once a full line (including the end of
439			line marker, C<$eol>) has been read. This line (excluding the end of line
440			marker) will be passed to the callback as second argument (C<$line>), and
441			the end of line marker as the third argument (C<$eol>).
442	elmex	1.1
443	root	1.8	The end of line marker, C<$eol>, can be either a string, in which case it
444			will be interpreted as a fixed record end marker, or it can be a regex
445			object (e.g. created by C<qr>), in which case it is interpreted as a
446			regular expression.
447	elmex	1.1
448	root	1.8	The end of line marker argument C<$eol> is optional, if it is missing (NOT
449			undef), then C<qr\|\015?\012\|> is used (which is good for most internet
450			protocols).
451	elmex	1.1
452	root	1.8	Partial lines at the end of the stream will never be returned, as they are
453			not marked by the end of line marker.
454	elmex	1.1
455	root	1.8	=cut
456	elmex	1.1
457	root	1.8	sub _read_line($$) {
458			my $cb = pop;
459			my $eol = @_ ? shift : qr\|(\015?\012)\|;
460			my $pos;
461	elmex	1.1
462	root	1.8	$eol = qr\|(\Q$eol\E)\| unless ref $eol;
463			$eol = qr\|^(.*?)($eol)\|;
464	elmex	1.1
465	root	1.8	sub {
466			$_[0]{rbuf} =~ s/$eol// or return;
467	elmex	1.1
468	root	1.8	$cb->($1, $2);
469			1
470			}
471			}
472	elmex	1.1
473	root	1.8	sub push_read_line {
474			my $self = shift;
475	elmex	1.1
476	root	1.8	$self->push_read (&_read_line);
477			}
478	elmex	1.1
479	root	1.8	sub unshift_read_line {
480			my $self = shift;
481	elmex	1.1
482	root	1.8	$self->unshift_read (&_read_line);
483	elmex	1.1	}
484
485			=back
486
487			=head1 AUTHOR
488
489	root	1.8	Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
490	elmex	1.1
491			=cut
492
493			1; # End of AnyEvent::Handle