lib/AnyEvent/Handle.pm

package AnyEvent::Handle;

use warnings;
use strict;

use AnyEvent;
use IO::Handle;
use Errno qw/EAGAIN EINTR/;

=head1 NAME

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

=head1 VERSION

Version 0.01

=cut

our $VERSION = '0.01';

=head1 SYNOPSIS

   use AnyEvent;
   use AnyEvent::Handle;

   my $cv = AnyEvent->condvar;

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

   $ae_fh->readlines (sub {
      my ($ae_fh, @lines) = @_;
      for (@lines) {
         chomp;
         print "Line: $_";
      }
      $cv->broadcast;
   });

   $cv->wait;

=head1 DESCRIPTION

This module is a helper module to make it easier to do non-blocking I/O
on filehandles (and sockets, see L<AnyEvent::Socket>).

The event loop is provided by L<AnyEvent>.

=head1 METHODS

=over 4

=item B<new (%args)>

The constructor has these arguments:

=over 4

=item fh => $filehandle

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

NOTE: The filehandle will be set to non-blocking.

=item read_block_size => $size

The default read block size use for reads via the C<on_read>
method.

=back

=cut

sub new {
   my $this  = shift;
   my $class = ref($this) || $this;
   my $self  = {
      read_block_size => 4096,
      rbuf            => '',
      @_
   };
   bless $self, $class;

   $self->{fh}->blocking (0) if $self->{fh};

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

   } elsif ($self->{on_readline}) {
      $self->readlines ($self->{on_readline});
   }

   return $self
}

=item B<fh>

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

=cut

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

=item B<on_read ($callback)>

This method installs a C<$callback> that will be called
when new data arrived. You can access the read buffer via the C<rbuf>
method (see below).

The first argument of the C<$callback> will be the L<AnyEvent::Handle> object.

=cut

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

   unless (defined $self->{on_read}) {
      delete $self->{on_read_w};
      return;
   }
  
   $self->{on_read_w} =
      AnyEvent->io (poll => 'r', fh => $self->{fh}, cb => sub {
         #d# warn "READ:[$self->{read_size}] $self->{read_block_size} : ".length ($self->{rbuf})."\n";
         my $rbuf_len = length $self->{rbuf};
         my $l;
         if (defined $self->{read_size}) {
            $l = sysread $self->{fh}, $self->{rbuf},
                         ($self->{read_size} - $rbuf_len), $rbuf_len;
         } else {
            $l = sysread $self->{fh}, $self->{rbuf}, $self->{read_block_size}, $rbuf_len;
         }
         #d# warn "READL $l [$self->{rbuf}]\n";

         if (not defined $l) {
            return if $! == EAGAIN || $! == EINTR;
            $self->{on_error}->($self, $!) if $self->{on_error};
            delete $self->{on_read_w};

         } elsif ($l == 0) {
            $self->{on_eof}->($self) if $self->{on_eof};
            delete $self->{on_read_w};

         } else {
            $self->{on_read}->($self);
         }
      });
}

=item B<on_error ($callback)>

Whenever a read or write operation resulted in an error the C<$callback>
will be called.

The first argument of C<$callback> will be the L<AnyEvent::Handle> object itself
and the second argument will be the value of C<$!>.

=cut

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

=item B<on_eof ($callback)>

Installs the C<$callback> that will be called when the end of file is
encountered in a read operation this C<$callback> will be called. The first
argument will be the L<AnyEvent::Handle> object itself.

=cut

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

=item B<rbuf>

Returns a reference to the read buffer.

NOTE: The read buffer should only be used or modified if the C<on_read>
method is used directly. The C<read> and C<readlines> methods will provide
the read data to their callbacks.

=cut

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

=item B<read ($len, $callback)>

Will read exactly C<$len> bytes from the filehandle and call the C<$callback>
if done so. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
object itself and the second argument the read data.

NOTE: This method will override any callbacks installed via the C<on_read> method.

=cut

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

   $self->{read_cb} = $cb;
   my $old_blk_size = $self->{read_block_size};
   $self->{read_block_size} = $len;

   $self->on_read (sub {
      #d# warn "OFOFO $len || ".length($_[0]->{rbuf})."||\n";

      if ($len == length $_[0]->{rbuf}) {
         $_[0]->{read_block_size} = $old_blk_size;
         $_[0]->on_read (undef);
         $_[0]->{read_cb}->($_[0], (substr $self->{rbuf}, 0, $len, ''));
      }
   });
}

=item B<readlines ($callback)>

=item B<readlines ($sep, $callback)>

This method will read lines from the filehandle, seperated by C<$sep> or C<"\n">
if C<$sep> is not provided. C<$sep> will be used as part of a regex, so it can be
a regex itself and won't be quoted!

The C<$callback> will be called when at least one
line could be read. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
object itself and the rest of the arguments will be the read lines.

NOTE: This method will override any callbacks installed via the C<on_read> method.

=cut

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

   if (ref $NL) {
      $cb = $NL;
      $NL = "\n";
   }

   $self->{on_readline} = $cb;

   $self->on_read (sub {
      my @lines;
      push @lines, $1 while $_[0]->{rbuf} =~ s/(.*)$NL//;
      $self->{on_readline}->($_[0], @lines);
   });
}

=item B<write ($data)>

=item B<write ($callback)>

=item B<write ($data, $callback)>

This method will write C<$data> to the filehandle and call the C<$callback>
afterwards. If only C<$callback> is provided it will be called when the
write buffer becomes empty the next time (or immediately if it already is empty).

=cut

sub write {
   my ($self, $data, $cb) = @_;
   if (ref $data) { $cb = $data; undef $data }
   push @{$self->{write_bufs}}, [$data, $cb];
   $self->_check_writer;
}

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

   if ($self->{write_w}) {
      unless ($self->{write_cb}) {
         while (@{$self->{write_bufs}} && not defined $self->{write_bufs}->[0]->[1]) {
            my $wba = shift @{$self->{write_bufs}};
            $self->{wbuf} .= $wba->[0];
         }
      }
      return;
   }

   my $wba = shift @{$self->{write_bufs}}
      or return;

   unless (defined $wba->[0]) {
      $wba->[1]->($self) if $wba->[1];
      $self->_check_writer;
      return;
   }

   $self->{wbuf}     = $wba->[0];
   $self->{write_cb} = $wba->[1];

   $self->{write_w} =
      AnyEvent->io (poll => 'w', fh => $self->{fh}, cb => sub {
         my $l = syswrite $self->{fh}, $self->{wbuf}, length $self->{wbuf};

         if (not defined $l) {
            return if $! == EAGAIN || $! == EINTR;
            delete $self->{write_w};

            $self->{on_error}->($self, $!) if $self->{on_error};

         } else {
            substr $self->{wbuf}, 0, $l, '';

            if (length ($self->{wbuf}) == 0) {
               $self->{write_cb}->($self) if $self->{write_cb};

               delete $self->{write_w};
               delete $self->{wbuf};
               delete $self->{write_cb};

               $self->_check_writer;
            }
         }
      });
}

=back

=head1 AUTHOR

Robin Redeker, C<< <elmex at ta-sa.org> >>

=head1 BUGS

Please report any bugs or feature requests to
C<bug-io-anyevent at rt.cpan.org>, or through the web interface at
L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=IO-AnyEvent>.
I will be notified, and then you'll automatically be notified of progress on
your bug as I make changes.

=head1 SUPPORT

You can find documentation for this module with the perldoc command.

    perldoc AnyEvent::Handle

You can also look for information at:

=over 4

=item * AnnoCPAN: Annotated CPAN documentation

L<http://annocpan.org/dist/IO-AnyEvent>

=item * CPAN Ratings

L<http://cpanratings.perl.org/d/IO-AnyEvent>

=item * RT: CPAN's request tracker

L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=IO-AnyEvent>

=item * Search CPAN

L<http://search.cpan.org/dist/IO-AnyEvent>

=back

=head1 ACKNOWLEDGEMENTS

=head1 COPYRIGHT & LICENSE

Copyright 2008 Robin Redeker, all rights reserved.

This program is free software; you can redistribute it and/or modify it
under the same terms as Perl itself.

=cut

1; # End of AnyEvent::Handle
Revision:	1.1
Committed:	Sun Apr 27 16:56:17 2008 UTC (16 years, 1 month ago) by elmex
Branch:	MAIN
Log Message:	added IO::AnyEvent as AnyEvent::Handle and AnyEvent::Socket, including tests
#	Content
1	package AnyEvent::Handle;
2
3	use warnings;
4	use strict;
5
6	use AnyEvent;
7	use IO::Handle;
8	use Errno qw/EAGAIN EINTR/;
9
10	=head1 NAME
11
12	AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent
13
14	=head1 VERSION
15
16	Version 0.01
17
18	=cut
19
20	our $VERSION = '0.01';
21
22	=head1 SYNOPSIS
23
24	use AnyEvent;
25	use AnyEvent::Handle;
26
27	my $cv = AnyEvent->condvar;
28
29	my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);
30
31	$ae_fh->readlines (sub {
32	my ($ae_fh, @lines) = @_;
33	for (@lines) {
34	chomp;
35	print "Line: $_";
36	}
37	$cv->broadcast;
38	});
39
40	$cv->wait;
41
42	=head1 DESCRIPTION
43
44	This module is a helper module to make it easier to do non-blocking I/O
45	on filehandles (and sockets, see L<AnyEvent::Socket>).
46
47	The event loop is provided by L<AnyEvent>.
48
49	=head1 METHODS
50
51	=over 4
52
53	=item B<new (%args)>
54
55	The constructor has these arguments:
56
57	=over 4
58
59	=item fh => $filehandle
60
61	The filehandle this L<AnyEvent::Handle> object will operate on.
62
63	NOTE: The filehandle will be set to non-blocking.
64
65	=item read_block_size => $size
66
67	The default read block size use for reads via the C<on_read>
68	method.
69
70	=back
71
72	=cut
73
74	sub new {
75	my $this = shift;
76	my $class = ref($this) \|\| $this;
77	my $self = {
78	read_block_size => 4096,
79	rbuf => '',
80	@_
81	};
82	bless $self, $class;
83
84	$self->{fh}->blocking (0) if $self->{fh};
85
86	if ($self->{on_read}) {
87	$self->on_read ($self->{on_read});
88
89	} elsif ($self->{on_readline}) {
90	$self->readlines ($self->{on_readline});
91	}
92
93	return $self
94	}
95
96	=item B<fh>
97
98	This method returns the filehandle of the L<AnyEvent::Handle> object.
99
100	=cut
101
102	sub fh { $_[0]->{fh} }
103
104	=item B<on_read ($callback)>
105
106	This method installs a C<$callback> that will be called
107	when new data arrived. You can access the read buffer via the C<rbuf>
108	method (see below).
109
110	The first argument of the C<$callback> will be the L<AnyEvent::Handle> object.
111
112	=cut
113
114	sub on_read {
115	my ($self, $cb) = @_;
116	$self->{on_read} = $cb;
117
118	unless (defined $self->{on_read}) {
119	delete $self->{on_read_w};
120	return;
121	}
122
123	$self->{on_read_w} =
124	AnyEvent->io (poll => 'r', fh => $self->{fh}, cb => sub {
125	#d# warn "READ:[$self->{read_size}] $self->{read_block_size} : ".length ($self->{rbuf})."\n";
126	my $rbuf_len = length $self->{rbuf};
127	my $l;
128	if (defined $self->{read_size}) {
129	$l = sysread $self->{fh}, $self->{rbuf},
130	($self->{read_size} - $rbuf_len), $rbuf_len;
131	} else {
132	$l = sysread $self->{fh}, $self->{rbuf}, $self->{read_block_size}, $rbuf_len;
133	}
134	#d# warn "READL $l [$self->{rbuf}]\n";
135
136	if (not defined $l) {
137	return if $! == EAGAIN \|\| $! == EINTR;
138	$self->{on_error}->($self, $!) if $self->{on_error};
139	delete $self->{on_read_w};
140
141	} elsif ($l == 0) {
142	$self->{on_eof}->($self) if $self->{on_eof};
143	delete $self->{on_read_w};
144
145	} else {
146	$self->{on_read}->($self);
147	}
148	});
149	}
150
151	=item B<on_error ($callback)>
152
153	Whenever a read or write operation resulted in an error the C<$callback>
154	will be called.
155
156	The first argument of C<$callback> will be the L<AnyEvent::Handle> object itself
157	and the second argument will be the value of C<$!>.
158
159	=cut
160
161	sub on_error {
162	$_[0]->{on_error} = $_[1];
163	}
164
165	=item B<on_eof ($callback)>
166
167	Installs the C<$callback> that will be called when the end of file is
168	encountered in a read operation this C<$callback> will be called. The first
169	argument will be the L<AnyEvent::Handle> object itself.
170
171	=cut
172
173	sub on_eof {
174	$_[0]->{on_eof} = $_[1];
175	}
176
177	=item B<rbuf>
178
179	Returns a reference to the read buffer.
180
181	NOTE: The read buffer should only be used or modified if the C<on_read>
182	method is used directly. The C<read> and C<readlines> methods will provide
183	the read data to their callbacks.
184
185	=cut
186
187	sub rbuf : lvalue { $_[0]->{rbuf} }
188
189	=item B<read ($len, $callback)>
190
191	Will read exactly C<$len> bytes from the filehandle and call the C<$callback>
192	if done so. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
193	object itself and the second argument the read data.
194
195	NOTE: This method will override any callbacks installed via the C<on_read> method.
196
197	=cut
198
199	sub read {
200	my ($self, $len, $cb) = @_;
201
202	$self->{read_cb} = $cb;
203	my $old_blk_size = $self->{read_block_size};
204	$self->{read_block_size} = $len;
205
206	$self->on_read (sub {
207	#d# warn "OFOFO $len \|\| ".length($_[0]->{rbuf})."\|\|\n";
208
209	if ($len == length $_[0]->{rbuf}) {
210	$_[0]->{read_block_size} = $old_blk_size;
211	$_[0]->on_read (undef);
212	$_[0]->{read_cb}->($_[0], (substr $self->{rbuf}, 0, $len, ''));
213	}
214	});
215	}
216
217	=item B<readlines ($callback)>
218
219	=item B<readlines ($sep, $callback)>
220
221	This method will read lines from the filehandle, seperated by C<$sep> or C<"\n">
222	if C<$sep> is not provided. C<$sep> will be used as part of a regex, so it can be
223	a regex itself and won't be quoted!
224
225	The C<$callback> will be called when at least one
226	line could be read. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
227	object itself and the rest of the arguments will be the read lines.
228
229	NOTE: This method will override any callbacks installed via the C<on_read> method.
230
231	=cut
232
233	sub readlines {
234	my ($self, $NL, $cb) = @_;
235
236	if (ref $NL) {
237	$cb = $NL;
238	$NL = "\n";
239	}
240
241	$self->{on_readline} = $cb;
242
243	$self->on_read (sub {
244	my @lines;
245	push @lines, $1 while $_[0]->{rbuf} =~ s/(.*)$NL//;
246	$self->{on_readline}->($_[0], @lines);
247	});
248	}
249
250	=item B<write ($data)>
251
252	=item B<write ($callback)>
253
254	=item B<write ($data, $callback)>
255
256	This method will write C<$data> to the filehandle and call the C<$callback>
257	afterwards. If only C<$callback> is provided it will be called when the
258	write buffer becomes empty the next time (or immediately if it already is empty).
259
260	=cut
261
262	sub write {
263	my ($self, $data, $cb) = @_;
264	if (ref $data) { $cb = $data; undef $data }
265	push @{$self->{write_bufs}}, [$data, $cb];
266	$self->_check_writer;
267	}
268
269	sub _check_writer {
270	my ($self) = @_;
271
272	if ($self->{write_w}) {
273	unless ($self->{write_cb}) {
274	while (@{$self->{write_bufs}} && not defined $self->{write_bufs}->[0]->[1]) {
275	my $wba = shift @{$self->{write_bufs}};
276	$self->{wbuf} .= $wba->[0];
277	}
278	}
279	return;
280	}
281
282	my $wba = shift @{$self->{write_bufs}}
283	or return;
284
285	unless (defined $wba->[0]) {
286	$wba->[1]->($self) if $wba->[1];
287	$self->_check_writer;
288	return;
289	}
290
291	$self->{wbuf} = $wba->[0];
292	$self->{write_cb} = $wba->[1];
293
294	$self->{write_w} =
295	AnyEvent->io (poll => 'w', fh => $self->{fh}, cb => sub {
296	my $l = syswrite $self->{fh}, $self->{wbuf}, length $self->{wbuf};
297
298	if (not defined $l) {
299	return if $! == EAGAIN \|\| $! == EINTR;
300	delete $self->{write_w};
301
302	$self->{on_error}->($self, $!) if $self->{on_error};
303
304	} else {
305	substr $self->{wbuf}, 0, $l, '';
306
307	if (length ($self->{wbuf}) == 0) {
308	$self->{write_cb}->($self) if $self->{write_cb};
309
310	delete $self->{write_w};
311	delete $self->{wbuf};
312	delete $self->{write_cb};
313
314	$self->_check_writer;
315	}
316	}
317	});
318	}
319
320	=back
321
322	=head1 AUTHOR
323
324	Robin Redeker, C<< <elmex at ta-sa.org> >>
325
326	=head1 BUGS
327
328	Please report any bugs or feature requests to
329	C<bug-io-anyevent at rt.cpan.org>, or through the web interface at
330	L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=IO-AnyEvent>.
331	I will be notified, and then you'll automatically be notified of progress on
332	your bug as I make changes.
333
334	=head1 SUPPORT
335
336	You can find documentation for this module with the perldoc command.
337
338	perldoc AnyEvent::Handle
339
340	You can also look for information at:
341
342	=over 4
343
344	=item * AnnoCPAN: Annotated CPAN documentation
345
346	L<http://annocpan.org/dist/IO-AnyEvent>
347
348	=item * CPAN Ratings
349
350	L<http://cpanratings.perl.org/d/IO-AnyEvent>
351
352	=item * RT: CPAN's request tracker
353
354	L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=IO-AnyEvent>
355
356	=item * Search CPAN
357
358	L<http://search.cpan.org/dist/IO-AnyEvent>
359
360	=back
361
362	=head1 ACKNOWLEDGEMENTS
363
364	=head1 COPYRIGHT & LICENSE
365
366	Copyright 2008 Robin Redeker, all rights reserved.
367
368	This program is free software; you can redistribute it and/or modify it
369	under the same terms as Perl itself.
370
371	=cut
372
373	1; # End of AnyEvent::Handle