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 _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}->(\$_[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 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 $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}->($rbuf)
               : $self->_drain_rbuf;

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

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

=back

=head1 AUTHOR

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

=cut

1; # End of AnyEvent::Handle
Revision:	1.17
Committed:	Sat May 24 04:17:45 2008 UTC (15 years, 11 months ago) by root
Branch:	MAIN
Changes since 1.16:	+27 -14 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	root	1.15	This module is experimental.
18
19	elmex	1.1	=cut
20
21	root	1.15	our $VERSION = '0.04';
22	elmex	1.1
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	root	1.8	#TODO
33	elmex	1.1
34	elmex	1.2	# 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	root	1.8	#TODO
43	elmex	1.2	);
44
45	elmex	1.1	$cv->wait;
46
47			=head1 DESCRIPTION
48
49	root	1.8	This module is a helper module to make it easier to do event-based I/O on
50	elmex	1.13	filehandles. For utility functions for doing non-blocking connects and accepts
51			on sockets see L<AnyEvent::Util>.
52	root	1.8
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	elmex	1.1
57	root	1.8	All callbacks will be invoked with the handle object as their first
58			argument.
59	elmex	1.1
60			=head1 METHODS
61
62			=over 4
63
64			=item B<new (%args)>
65
66	root	1.8	The constructor supports these arguments (all as key => value pairs).
67	elmex	1.1
68			=over 4
69
70	root	1.8	=item fh => $filehandle [MANDATORY]
71	elmex	1.1
72			The filehandle this L<AnyEvent::Handle> object will operate on.
73
74	root	1.8	NOTE: The filehandle will be set to non-blocking (using
75			AnyEvent::Util::fh_nonblocking).
76
77	root	1.16	=item on_eof => $cb->($self)
78	root	1.10
79			Set the callback to be called on EOF.
80	root	1.8
81	root	1.16	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	root	1.10	=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	root	1.8
91			The object will not be in a usable state when this callback has been
92			called.
93
94	root	1.10	On callback entrance, the value of C<$!> contains the operating system
95	root	1.8	error (or C<ENOSPC> or C<EPIPE>).
96
97	root	1.10	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	root	1.8
101			=item on_read => $cb->($self)
102
103			This sets the default read callback, which is called when data arrives
104	root	1.10	and no read request is in the queue.
105	root	1.8
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	elmex	1.1
114	root	1.8	=item on_drain => $cb->()
115	elmex	1.1
116	root	1.8	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	elmex	1.1
119	root	1.8	To append to the write buffer, use the C<< ->push_write >> method.
120	elmex	1.2
121	root	1.8	=item rbuf_max => <bytes>
122	elmex	1.2
123	root	1.8	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	elmex	1.2
127	root	1.8	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	elmex	1.2
133	root	1.8	=item read_size => <bytes>
134	elmex	1.2
135	root	1.8	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	elmex	1.2
144	elmex	1.1	=back
145
146			=cut
147
148			sub new {
149	root	1.8	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	elmex	1.1
157	root	1.16	$self->on_eof (delete $self->{on_eof} ) if $self->{on_eof};
158	root	1.10	$self->on_error (delete $self->{on_error}) if $self->{on_error};
159	root	1.8	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
160			$self->on_read (delete $self->{on_read} ) if $self->{on_read};
161	elmex	1.1
162	root	1.10	$self->start_read;
163
164	root	1.8	$self
165			}
166	elmex	1.2
167	root	1.8	sub _shutdown {
168			my ($self) = @_;
169	elmex	1.2
170	root	1.8	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	elmex	1.1	}
182
183	root	1.10	if ($self->{on_error}) {
184			$self->{on_error}($self);
185			} else {
186			die "AnyEvent::Handle uncaught fatal error: $!";
187			}
188	elmex	1.1	}
189
190	root	1.8	=item $fh = $handle->fh
191	elmex	1.1
192			This method returns the filehandle of the L<AnyEvent::Handle> object.
193
194			=cut
195
196			sub fh { $_[0]->{fh} }
197
198	root	1.8	=item $handle->on_error ($cb)
199	elmex	1.1
200	root	1.8	Replace the current C<on_error> callback (see the C<on_error> constructor argument).
201	elmex	1.1
202	root	1.8	=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	elmex	1.1
212			=cut
213
214	root	1.8	sub on_eof {
215			$_[0]{on_eof} = $_[1];
216			}
217
218	root	1.9	#############################################################################
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	root	1.8	=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	elmex	1.1	my ($self, $cb) = @_;
244
245	root	1.8	$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	root	1.17	sub _drain_wbuf {
260			my ($self) = @_;
261	root	1.8
262			unless ($self->{ww}) {
263			Scalar::Util::weaken $self;
264			my $cb = sub {
265			my $len = syswrite $self->{fh}, $self->{wbuf};
266
267			if ($len > 0) {
268			substr $self->{wbuf}, 0, $len, "";
269
270			$self->{on_drain}($self)
271			if $self->{low_water_mark} >= length $self->{wbuf}
272			&& $self->{on_drain};
273
274			delete $self->{ww} unless length $self->{wbuf};
275			} elsif ($! != EAGAIN && $! != EINTR) {
276			$self->error;
277	elmex	1.1	}
278	root	1.8	};
279
280			$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
281
282			$cb->($self);
283			};
284			}
285
286	root	1.17	sub push_write {
287			my $self = shift;
288
289			if ($self->{filter_w}) {
290			$self->{filter_w}->(\$_[0]);
291			} else {
292			$self->{wbuf} .= $_[0];
293			$self->_drain_wbuf;
294			}
295			}
296
297	root	1.8	#############################################################################
298
299	root	1.9	=back
300
301			=head2 READ QUEUE
302
303			AnyEvent::Handle manages two queues per handle, one for writing and one
304			for reading.
305
306			The read queue is more complex than the write queue. It can be used in two
307			ways, the "simple" way, using only C<on_read> and the "complex" way, using
308			a queue.
309
310			In the simple case, you just install an C<on_read> callback and whenever
311			new data arrives, it will be called. You can then remove some data (if
312			enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
313			or not.
314
315			In the more complex case, you want to queue multiple callbacks. In this
316			case, AnyEvent::Handle will call the first queued callback each time new
317			data arrives and removes it when it has done its job (see C<push_read>,
318			below).
319
320			This way you can, for example, push three line-reads, followed by reading
321			a chunk of data, and AnyEvent::Handle will execute them in order.
322
323			Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
324			the specified number of bytes which give an XML datagram.
325
326			# in the default state, expect some header bytes
327			$handle->on_read (sub {
328			# some data is here, now queue the length-header-read (4 octets)
329			shift->unshift_read_chunk (4, sub {
330			# header arrived, decode
331			my $len = unpack "N", $_[1];
332
333			# now read the payload
334			shift->unshift_read_chunk ($len, sub {
335			my $xml = $_[1];
336			# handle xml
337			});
338			});
339			});
340
341			Example 2: Implement a client for a protocol that replies either with
342			"OK" and another line or "ERROR" for one request, and 64 bytes for the
343			second request. Due tot he availability of a full queue, we can just
344			pipeline sending both requests and manipulate the queue as necessary in
345			the callbacks:
346
347			# request one
348			$handle->push_write ("request 1\015\012");
349
350			# we expect "ERROR" or "OK" as response, so push a line read
351			$handle->push_read_line (sub {
352			# if we got an "OK", we have to _prepend_ another line,
353			# so it will be read before the second request reads its 64 bytes
354			# which are already in the queue when this callback is called
355			# we don't do this in case we got an error
356			if ($_[1] eq "OK") {
357			$_[0]->unshift_read_line (sub {
358			my $response = $_[1];
359			...
360			});
361			}
362			});
363
364			# request two
365			$handle->push_write ("request 2\015\012");
366
367			# simply read 64 bytes, always
368			$handle->push_read_chunk (64, sub {
369			my $response = $_[1];
370			...
371			});
372
373			=over 4
374
375	root	1.10	=cut
376
377	root	1.8	sub _drain_rbuf {
378			my ($self) = @_;
379	elmex	1.1
380	root	1.17	if (
381			defined $self->{rbuf_max}
382			&& $self->{rbuf_max} < length $self->{rbuf}
383			) {
384			$! = &Errno::ENOSPC; return $self->error;
385			}
386
387	root	1.11	return if $self->{in_drain};
388	root	1.8	local $self->{in_drain} = 1;
389	elmex	1.1
390	root	1.8	while (my $len = length $self->{rbuf}) {
391			no strict 'refs';
392	root	1.10	if (my $cb = shift @{ $self->{queue} }) {
393			if (!$cb->($self)) {
394			if ($self->{eof}) {
395			# no progress can be made (not enough data and no data forthcoming)
396			$! = &Errno::EPIPE; return $self->error;
397			}
398
399			unshift @{ $self->{queue} }, $cb;
400	root	1.8	return;
401			}
402			} elsif ($self->{on_read}) {
403			$self->{on_read}($self);
404
405			if (
406			$self->{eof} # if no further data will arrive
407			&& $len == length $self->{rbuf} # and no data has been consumed
408			&& !@{ $self->{queue} } # and the queue is still empty
409			&& $self->{on_read} # and we still want to read data
410			) {
411			# then no progress can be made
412			$! = &Errno::EPIPE; return $self->error;
413	elmex	1.1	}
414	root	1.8	} else {
415			# read side becomes idle
416			delete $self->{rw};
417			return;
418			}
419			}
420
421			if ($self->{eof}) {
422			$self->_shutdown;
423	root	1.16	$self->{on_eof}($self)
424			if $self->{on_eof};
425	root	1.8	}
426	elmex	1.1	}
427
428	root	1.8	=item $handle->on_read ($cb)
429	elmex	1.1
430	root	1.8	This replaces the currently set C<on_read> callback, or clears it (when
431			the new callback is C<undef>). See the description of C<on_read> in the
432			constructor.
433	elmex	1.1
434	root	1.8	=cut
435
436			sub on_read {
437			my ($self, $cb) = @_;
438	elmex	1.1
439	root	1.8	$self->{on_read} = $cb;
440	elmex	1.1	}
441
442	root	1.8	=item $handle->rbuf
443
444			Returns the read buffer (as a modifiable lvalue).
445	elmex	1.1
446	root	1.8	You can access the read buffer directly as the C<< ->{rbuf} >> member, if
447			you want.
448	elmex	1.1
449	root	1.8	NOTE: The read buffer should only be used or modified if the C<on_read>,
450			C<push_read> or C<unshift_read> methods are used. The other read methods
451			automatically manage the read buffer.
452	elmex	1.1
453			=cut
454
455	elmex	1.2	sub rbuf : lvalue {
456	root	1.8	$_[0]{rbuf}
457	elmex	1.2	}
458	elmex	1.1
459	root	1.8	=item $handle->push_read ($cb)
460
461			=item $handle->unshift_read ($cb)
462
463			Append the given callback to the end of the queue (C<push_read>) or
464			prepend it (C<unshift_read>).
465
466			The callback is called each time some additional read data arrives.
467	elmex	1.1
468	root	1.8	It must check wether enough data is in the read buffer already.
469	elmex	1.1
470	root	1.8	If not enough data is available, it must return the empty list or a false
471			value, in which case it will be called repeatedly until enough data is
472			available (or an error condition is detected).
473
474			If enough data was available, then the callback must remove all data it is
475			interested in (which can be none at all) and return a true value. After returning
476			true, it will be removed from the queue.
477	elmex	1.1
478			=cut
479
480	root	1.8	sub push_read {
481			my ($self, $cb) = @_;
482	elmex	1.1
483	root	1.8	push @{ $self->{queue} }, $cb;
484			$self->_drain_rbuf;
485	elmex	1.1	}
486
487	root	1.8	sub unshift_read {
488			my ($self, $cb) = @_;
489
490			push @{ $self->{queue} }, $cb;
491			$self->_drain_rbuf;
492			}
493	elmex	1.1
494	root	1.8	=item $handle->push_read_chunk ($len, $cb->($self, $data))
495	elmex	1.1
496	root	1.8	=item $handle->unshift_read_chunk ($len, $cb->($self, $data))
497	elmex	1.1
498	root	1.8	Append the given callback to the end of the queue (C<push_read_chunk>) or
499			prepend it (C<unshift_read_chunk>).
500	elmex	1.1
501	root	1.8	The callback will be called only once C<$len> bytes have been read, and
502			these C<$len> bytes will be passed to the callback.
503	elmex	1.1
504			=cut
505
506	root	1.8	sub _read_chunk($$) {
507	root	1.10	my ($self, $len, $cb) = @_;
508	elmex	1.1
509	root	1.8	sub {
510			$len <= length $_[0]{rbuf} or return;
511	elmex	1.12	$cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
512	root	1.8	1
513			}
514			}
515
516			sub push_read_chunk {
517	root	1.10	$_[0]->push_read (&_read_chunk);
518	root	1.8	}
519	elmex	1.1
520	elmex	1.5
521	root	1.8	sub unshift_read_chunk {
522	root	1.10	$_[0]->unshift_read (&_read_chunk);
523	elmex	1.1	}
524
525	root	1.8	=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
526	elmex	1.1
527	root	1.8	=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
528	elmex	1.1
529	root	1.8	Append the given callback to the end of the queue (C<push_read_line>) or
530			prepend it (C<unshift_read_line>).
531	elmex	1.1
532	root	1.8	The callback will be called only once a full line (including the end of
533			line marker, C<$eol>) has been read. This line (excluding the end of line
534			marker) will be passed to the callback as second argument (C<$line>), and
535			the end of line marker as the third argument (C<$eol>).
536	elmex	1.1
537	root	1.8	The end of line marker, C<$eol>, can be either a string, in which case it
538			will be interpreted as a fixed record end marker, or it can be a regex
539			object (e.g. created by C<qr>), in which case it is interpreted as a
540			regular expression.
541	elmex	1.1
542	root	1.8	The end of line marker argument C<$eol> is optional, if it is missing (NOT
543			undef), then C<qr\|\015?\012\|> is used (which is good for most internet
544			protocols).
545	elmex	1.1
546	root	1.8	Partial lines at the end of the stream will never be returned, as they are
547			not marked by the end of line marker.
548	elmex	1.1
549	root	1.8	=cut
550	elmex	1.1
551	root	1.8	sub _read_line($$) {
552	root	1.10	my $self = shift;
553	root	1.8	my $cb = pop;
554			my $eol = @_ ? shift : qr\|(\015?\012)\|;
555			my $pos;
556	elmex	1.1
557	root	1.14	$eol = quotemeta $eol unless ref $eol;
558			$eol = qr\|^(.*?)($eol)\|s;
559	elmex	1.1
560	root	1.8	sub {
561			$_[0]{rbuf} =~ s/$eol// or return;
562	elmex	1.1
563	elmex	1.12	$cb->($_[0], $1, $2);
564	root	1.8	1
565			}
566			}
567	elmex	1.1
568	root	1.8	sub push_read_line {
569	root	1.10	$_[0]->push_read (&_read_line);
570			}
571
572			sub unshift_read_line {
573			$_[0]->unshift_read (&_read_line);
574			}
575
576			=item $handle->stop_read
577
578			=item $handle->start_read
579
580			In rare cases you actually do not want to read anything form the
581			socket. In this case you can call C<stop_read>. Neither C<on_read> no
582			any queued callbacks will be executed then. To start readign again, call
583			C<start_read>.
584
585			=cut
586
587			sub stop_read {
588			my ($self) = @_;
589	elmex	1.1
590	root	1.10	delete $self->{rw};
591	root	1.8	}
592	elmex	1.1
593	root	1.10	sub start_read {
594			my ($self) = @_;
595
596			unless ($self->{rw} \|\| $self->{eof}) {
597			Scalar::Util::weaken $self;
598
599			$self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
600	root	1.17	my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
601			my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} \|\| 8192, length $$rbuf;
602	root	1.10
603			if ($len > 0) {
604	root	1.17	$self->{filter_r}
605			? $self->{filter_r}->($rbuf)
606			: $self->_drain_rbuf;
607	root	1.10
608			} elsif (defined $len) {
609	root	1.17	delete $self->{rw};
610	root	1.10	$self->{eof} = 1;
611	root	1.17	$self->_drain_rbuf;
612	root	1.10
613			} elsif ($! != EAGAIN && $! != EINTR) {
614			return $self->error;
615			}
616			});
617			}
618	elmex	1.1	}
619
620			=back
621
622			=head1 AUTHOR
623
624	root	1.8	Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
625	elmex	1.1
626			=cut
627
628			1; # End of AnyEvent::Handle