ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.8 by root, Fri May 2 15:36:10 2008 UTC vs.
Revision 1.45 by root, Thu May 29 00:20:39 2008 UTC

2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util (); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw(EAGAIN EINTR);
12 12
13=head1 NAME 13=head1 NAME
14 14
15AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17=cut 17=cut
18 18
19our $VERSION = '0.02'; 19our $VERSION = '0.04';
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
25 25
26 my $cv = AnyEvent->condvar; 26 my $cv = AnyEvent->condvar;
27 27
28 my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN); 28 my $handle =
29
30 #TODO
31
32 # or use the constructor to pass the callback:
33
34 my $ae_fh2 =
35 AnyEvent::Handle->new ( 29 AnyEvent::Handle->new (
36 fh => \*STDIN, 30 fh => \*STDIN,
37 on_eof => sub { 31 on_eof => sub {
38 $cv->broadcast; 32 $cv->broadcast;
39 }, 33 },
40 #TODO
41 ); 34 );
42 35
43 $cv->wait; 36 # send some request line
37 $handle->push_write ("getinfo\015\012");
38
39 # read the response line
40 $handle->push_read (line => sub {
41 my ($handle, $line) = @_;
42 warn "read line <$line>\n";
43 $cv->send;
44 });
45
46 $cv->recv;
44 47
45=head1 DESCRIPTION 48=head1 DESCRIPTION
46 49
47This module is a helper module to make it easier to do event-based I/O on 50This module is a helper module to make it easier to do event-based I/O on
48filehandles (and sockets, see L<AnyEvent::Socket> for an easy way to make 51filehandles. For utility functions for doing non-blocking connects and accepts
49non-blocking resolves and connects). 52on sockets see L<AnyEvent::Util>.
50 53
51In the following, when the documentation refers to of "bytes" then this 54In the following, when the documentation refers to of "bytes" then this
52means characters. As sysread and syswrite are used for all I/O, their 55means characters. As sysread and syswrite are used for all I/O, their
53treatment of characters applies to this module as well. 56treatment of characters applies to this module as well.
54 57
70The filehandle this L<AnyEvent::Handle> object will operate on. 73The filehandle this L<AnyEvent::Handle> object will operate on.
71 74
72NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
73AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
74 77
75=item on_error => $cb->($self) [MANDATORY] 78=item on_eof => $cb->($handle)
76 79
80Set the callback to be called on EOF.
81
82While not mandatory, it is highly recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still
84waiting for data.
85
86=item on_error => $cb->($handle)
87
77This is the fatal error callback, that is called when a fatal error ocurs, 88This is the fatal error callback, that is called when, well, a fatal error
78such as not being able to resolve the hostname, failure to connect or a 89occurs, such as not being able to resolve the hostname, failure to connect
79read error. 90or a read error.
80 91
81The object will not be in a usable state when this callback has been 92The object will not be in a usable state when this callback has been
82called. 93called.
83 94
84On callback entrance, the value of C<$!> contains the opertaing system 95On callback entrance, the value of C<$!> contains the operating system
85error (or C<ENOSPC> or C<EPIPE>). 96error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
86 97
87=item on_eof => $cb->($self) [MANDATORY] 98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
88 100
89Set the callback to be called on EOF. 101While not mandatory, it is I<highly> recommended to set this callback, as
102you will not be notified of errors otherwise. The default simply calls
103die.
90 104
91=item on_read => $cb->($self) 105=item on_read => $cb->($handle)
92 106
93This sets the default read callback, which is called when data arrives 107This sets the default read callback, which is called when data arrives
94and no read request is in the queue. If the read callback is C<undef> 108and no read request is in the queue.
95or has never been set, than AnyEvent::Handle will cease reading from the
96filehandle.
97 109
98To access (and remove data from) the read buffer, use the C<< ->rbuf >> 110To access (and remove data from) the read buffer, use the C<< ->rbuf >>
99method or acces sthe C<$self->{rbuf}> member directly. 111method or access the C<$handle->{rbuf}> member directly.
100 112
101When an EOF condition is detected then AnyEvent::Handle will first try to 113When an EOF condition is detected then AnyEvent::Handle will first try to
102feed all the remaining data to the queued callbacks and C<on_read> before 114feed all the remaining data to the queued callbacks and C<on_read> before
103calling the C<on_eof> callback. If no progress can be made, then a fatal 115calling the C<on_eof> callback. If no progress can be made, then a fatal
104error will be raised (with C<$!> set to C<EPIPE>). 116error will be raised (with C<$!> set to C<EPIPE>).
105 117
106=item on_drain => $cb->() 118=item on_drain => $cb->($handle)
107 119
108This sets the callback that is called when the write buffer becomes empty 120This 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). 121(or when the callback is set and the buffer is empty already).
110 122
111To append to the write buffer, use the C<< ->push_write >> method. 123To append to the write buffer, use the C<< ->push_write >> method.
124
125=item timeout => $fractional_seconds
126
127If non-zero, then this enables an "inactivity" timeout: whenever this many
128seconds pass without a successful read or write on the underlying file
129handle, the C<on_timeout> callback will be invoked (and if that one is
130missing, an C<ETIMEDOUT> error will be raised).
131
132Note that timeout processing is also active when you currently do not have
133any outstanding read or write requests: If you plan to keep the connection
134idle then you should disable the timout temporarily or ignore the timeout
135in the C<on_timeout> callback.
136
137Zero (the default) disables this timeout.
138
139=item on_timeout => $cb->($handle)
140
141Called whenever the inactivity timeout passes. If you return from this
142callback, then the timeout will be reset as if some activity had happened,
143so this condition is not fatal in any way.
112 144
113=item rbuf_max => <bytes> 145=item rbuf_max => <bytes>
114 146
115If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 147If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
116when the read buffer ever (strictly) exceeds this size. This is useful to 148when the read buffer ever (strictly) exceeds this size. This is useful to
131 163
132Sets the amount of bytes (default: C<0>) that make up an "empty" write 164Sets the amount of bytes (default: C<0>) that make up an "empty" write
133buffer: If the write reaches this size or gets even samller it is 165buffer: If the write reaches this size or gets even samller it is
134considered empty. 166considered empty.
135 167
168=item tls => "accept" | "connect" | Net::SSLeay::SSL object
169
170When this parameter is given, it enables TLS (SSL) mode, that means it
171will start making tls handshake and will transparently encrypt/decrypt
172data.
173
174TLS mode requires Net::SSLeay to be installed (it will be loaded
175automatically when you try to create a TLS handle).
176
177For the TLS server side, use C<accept>, and for the TLS client side of a
178connection, use C<connect> mode.
179
180You can also provide your own TLS connection object, but you have
181to make sure that you call either C<Net::SSLeay::set_connect_state>
182or C<Net::SSLeay::set_accept_state> on it before you pass it to
183AnyEvent::Handle.
184
185See the C<starttls> method if you need to start TLs negotiation later.
186
187=item tls_ctx => $ssl_ctx
188
189Use the given Net::SSLeay::CTX object to create the new TLS connection
190(unless a connection object was specified directly). If this parameter is
191missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
192
193=item json => JSON or JSON::XS object
194
195This is the json coder object used by the C<json> read and write types.
196
197If you don't supply it, then AnyEvent::Handle will create and use a
198suitable one, which will write and expect UTF-8 encoded JSON texts.
199
200Note that you are responsible to depend on the JSON module if you want to
201use this functionality, as AnyEvent does not have a dependency itself.
202
203=item filter_r => $cb
204
205=item filter_w => $cb
206
207These exist, but are undocumented at this time.
208
136=back 209=back
137 210
138=cut 211=cut
139 212
140sub new { 213sub new {
144 217
145 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 218 $self->{fh} or Carp::croak "mandatory argument fh is missing";
146 219
147 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 220 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
148 221
149 $self->on_error ((delete $self->{on_error}) or Carp::croak "mandatory argument on_error is missing"); 222 if ($self->{tls}) {
150 $self->on_eof ((delete $self->{on_eof} ) or Carp::croak "mandatory argument on_eof is missing"); 223 require Net::SSLeay;
224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
225 }
151 226
227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
228# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
229# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
152 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
153 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 231
232 $self->{_activity} = AnyEvent->now;
233 $self->_timeout;
234
235 $self->start_read;
154 236
155 $self 237 $self
156} 238}
157 239
158sub _shutdown { 240sub _shutdown {
159 my ($self) = @_; 241 my ($self) = @_;
160 242
161 delete $self->{rw}; 243 delete $self->{_rw};
162 delete $self->{ww}; 244 delete $self->{_ww};
163 delete $self->{fh}; 245 delete $self->{fh};
164} 246}
165 247
166sub error { 248sub error {
167 my ($self) = @_; 249 my ($self) = @_;
169 { 251 {
170 local $!; 252 local $!;
171 $self->_shutdown; 253 $self->_shutdown;
172 } 254 }
173 255
174 $self->{on_error}($self); 256 $self->{on_error}($self)
257 if $self->{on_error};
258
259 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
175} 260}
176 261
177=item $fh = $handle->fh 262=item $fh = $handle->fh
178 263
179This method returns the filehandle of the L<AnyEvent::Handle> object. 264This method returns the file handle of the L<AnyEvent::Handle> object.
180 265
181=cut 266=cut
182 267
183sub fh { $_[0]->{fh} } 268sub fh { $_[0]{fh} }
184 269
185=item $handle->on_error ($cb) 270=item $handle->on_error ($cb)
186 271
187Replace the current C<on_error> callback (see the C<on_error> constructor argument). 272Replace the current C<on_error> callback (see the C<on_error> constructor argument).
188 273
196 281
197Replace the current C<on_eof> callback (see the C<on_eof> constructor argument). 282Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
198 283
199=cut 284=cut
200 285
201#############################################################################
202
203sub on_eof { 286sub on_eof {
204 $_[0]{on_eof} = $_[1]; 287 $_[0]{on_eof} = $_[1];
205} 288}
289
290=item $handle->on_timeout ($cb)
291
292Replace the current C<on_timeout> callback, or disables the callback
293(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
294argument.
295
296=cut
297
298sub on_timeout {
299 $_[0]{on_timeout} = $_[1];
300}
301
302#############################################################################
303
304=item $handle->timeout ($seconds)
305
306Configures (or disables) the inactivity timeout.
307
308=cut
309
310sub timeout {
311 my ($self, $timeout) = @_;
312
313 $self->{timeout} = $timeout;
314 $self->_timeout;
315}
316
317# reset the timeout watcher, as neccessary
318# also check for time-outs
319sub _timeout {
320 my ($self) = @_;
321
322 if ($self->{timeout}) {
323 my $NOW = AnyEvent->now;
324
325 # when would the timeout trigger?
326 my $after = $self->{_activity} + $self->{timeout} - $NOW;
327
328 # now or in the past already?
329 if ($after <= 0) {
330 $self->{_activity} = $NOW;
331
332 if ($self->{on_timeout}) {
333 $self->{on_timeout}->($self);
334 } else {
335 $! = Errno::ETIMEDOUT;
336 $self->error;
337 }
338
339 # callbakx could have changed timeout value, optimise
340 return unless $self->{timeout};
341
342 # calculate new after
343 $after = $self->{timeout};
344 }
345
346 Scalar::Util::weaken $self;
347
348 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
349 delete $self->{_tw};
350 $self->_timeout;
351 });
352 } else {
353 delete $self->{_tw};
354 }
355}
356
357#############################################################################
358
359=back
360
361=head2 WRITE QUEUE
362
363AnyEvent::Handle manages two queues per handle, one for writing and one
364for reading.
365
366The write queue is very simple: you can add data to its end, and
367AnyEvent::Handle will automatically try to get rid of it for you.
368
369When data could be written and the write buffer is shorter then the low
370water mark, the C<on_drain> callback will be invoked.
371
372=over 4
206 373
207=item $handle->on_drain ($cb) 374=item $handle->on_drain ($cb)
208 375
209Sets the C<on_drain> callback or clears it (see the description of 376Sets the C<on_drain> callback or clears it (see the description of
210C<on_drain> in the constructor). 377C<on_drain> in the constructor).
226want (only limited by the available memory), as C<AnyEvent::Handle> 393want (only limited by the available memory), as C<AnyEvent::Handle>
227buffers it independently of the kernel. 394buffers it independently of the kernel.
228 395
229=cut 396=cut
230 397
231sub push_write { 398sub _drain_wbuf {
232 my ($self, $data) = @_; 399 my ($self) = @_;
233 400
234 $self->{wbuf} .= $data; 401 if (!$self->{_ww} && length $self->{wbuf}) {
235 402
236 unless ($self->{ww}) {
237 Scalar::Util::weaken $self; 403 Scalar::Util::weaken $self;
404
238 my $cb = sub { 405 my $cb = sub {
239 my $len = syswrite $self->{fh}, $self->{wbuf}; 406 my $len = syswrite $self->{fh}, $self->{wbuf};
240 407
241 if ($len > 0) { 408 if ($len >= 0) {
242 substr $self->{wbuf}, 0, $len, ""; 409 substr $self->{wbuf}, 0, $len, "";
243 410
411 $self->{_activity} = AnyEvent->now;
244 412
245 $self->{on_drain}($self) 413 $self->{on_drain}($self)
246 if $self->{low_water_mark} >= length $self->{wbuf} 414 if $self->{low_water_mark} >= length $self->{wbuf}
247 && $self->{on_drain}; 415 && $self->{on_drain};
248 416
249 delete $self->{ww} unless length $self->{wbuf}; 417 delete $self->{_ww} unless length $self->{wbuf};
250 } elsif ($! != EAGAIN && $! != EINTR) { 418 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
251 $self->error; 419 $self->error;
252 } 420 }
253 }; 421 };
254 422
423 # try to write data immediately
424 $cb->();
425
426 # if still data left in wbuf, we need to poll
255 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 427 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
256 428 if length $self->{wbuf};
257 $cb->($self);
258 }; 429 };
259} 430}
260 431
432our %WH;
433
434sub register_write_type($$) {
435 $WH{$_[0]} = $_[1];
436}
437
438sub push_write {
439 my $self = shift;
440
441 if (@_ > 1) {
442 my $type = shift;
443
444 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
445 ->($self, @_);
446 }
447
448 if ($self->{filter_w}) {
449 $self->{filter_w}->($self, \$_[0]);
450 } else {
451 $self->{wbuf} .= $_[0];
452 $self->_drain_wbuf;
453 }
454}
455
456=item $handle->push_write (type => @args)
457
458=item $handle->unshift_write (type => @args)
459
460Instead of formatting your data yourself, you can also let this module do
461the job by specifying a type and type-specific arguments.
462
463Predefined types are (if you have ideas for additional types, feel free to
464drop by and tell us):
465
466=over 4
467
468=item netstring => $string
469
470Formats the given value as netstring
471(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
472
473=back
474
475=cut
476
477register_write_type netstring => sub {
478 my ($self, $string) = @_;
479
480 sprintf "%d:%s,", (length $string), $string
481};
482
483=item json => $array_or_hashref
484
485Encodes the given hash or array reference into a JSON object. Unless you
486provide your own JSON object, this means it will be encoded to JSON text
487in UTF-8.
488
489JSON objects (and arrays) are self-delimiting, so you can write JSON at
490one end of a handle and read them at the other end without using any
491additional framing.
492
493The generated JSON text is guaranteed not to contain any newlines: While
494this module doesn't need delimiters after or between JSON texts to be
495able to read them, many other languages depend on that.
496
497A simple RPC protocol that interoperates easily with others is to send
498JSON arrays (or objects, although arrays are usually the better choice as
499they mimic how function argument passing works) and a newline after each
500JSON text:
501
502 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
503 $handle->push_write ("\012");
504
505An AnyEvent::Handle receiver would simply use the C<json> read type and
506rely on the fact that the newline will be skipped as leading whitespace:
507
508 $handle->push_read (json => sub { my $array = $_[1]; ... });
509
510Other languages could read single lines terminated by a newline and pass
511this line into their JSON decoder of choice.
512
513=cut
514
515register_write_type json => sub {
516 my ($self, $ref) = @_;
517
518 require JSON;
519
520 $self->{json} ? $self->{json}->encode ($ref)
521 : JSON::encode_json ($ref)
522};
523
524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525
526This function (not method) lets you add your own types to C<push_write>.
527Whenever the given C<type> is used, C<push_write> will invoke the code
528reference with the handle object and the remaining arguments.
529
530The code reference is supposed to return a single octet string that will
531be appended to the write buffer.
532
533Note that this is a function, and all types registered this way will be
534global, so try to use unique names.
535
536=cut
537
261############################################################################# 538#############################################################################
539
540=back
541
542=head2 READ QUEUE
543
544AnyEvent::Handle manages two queues per handle, one for writing and one
545for reading.
546
547The read queue is more complex than the write queue. It can be used in two
548ways, the "simple" way, using only C<on_read> and the "complex" way, using
549a queue.
550
551In the simple case, you just install an C<on_read> callback and whenever
552new data arrives, it will be called. You can then remove some data (if
553enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
554or not.
555
556In the more complex case, you want to queue multiple callbacks. In this
557case, AnyEvent::Handle will call the first queued callback each time new
558data arrives and removes it when it has done its job (see C<push_read>,
559below).
560
561This way you can, for example, push three line-reads, followed by reading
562a chunk of data, and AnyEvent::Handle will execute them in order.
563
564Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
565the specified number of bytes which give an XML datagram.
566
567 # in the default state, expect some header bytes
568 $handle->on_read (sub {
569 # some data is here, now queue the length-header-read (4 octets)
570 shift->unshift_read_chunk (4, sub {
571 # header arrived, decode
572 my $len = unpack "N", $_[1];
573
574 # now read the payload
575 shift->unshift_read_chunk ($len, sub {
576 my $xml = $_[1];
577 # handle xml
578 });
579 });
580 });
581
582Example 2: Implement a client for a protocol that replies either with
583"OK" and another line or "ERROR" for one request, and 64 bytes for the
584second request. Due tot he availability of a full queue, we can just
585pipeline sending both requests and manipulate the queue as necessary in
586the callbacks:
587
588 # request one
589 $handle->push_write ("request 1\015\012");
590
591 # we expect "ERROR" or "OK" as response, so push a line read
592 $handle->push_read_line (sub {
593 # if we got an "OK", we have to _prepend_ another line,
594 # so it will be read before the second request reads its 64 bytes
595 # which are already in the queue when this callback is called
596 # we don't do this in case we got an error
597 if ($_[1] eq "OK") {
598 $_[0]->unshift_read_line (sub {
599 my $response = $_[1];
600 ...
601 });
602 }
603 });
604
605 # request two
606 $handle->push_write ("request 2\015\012");
607
608 # simply read 64 bytes, always
609 $handle->push_read_chunk (64, sub {
610 my $response = $_[1];
611 ...
612 });
613
614=over 4
615
616=cut
262 617
263sub _drain_rbuf { 618sub _drain_rbuf {
264 my ($self) = @_; 619 my ($self) = @_;
265 620
621 if (
622 defined $self->{rbuf_max}
623 && $self->{rbuf_max} < length $self->{rbuf}
624 ) {
625 $! = &Errno::ENOSPC;
626 $self->error;
627 }
628
266 return if exists $self->{in_drain}; 629 return if $self->{in_drain};
267 local $self->{in_drain} = 1; 630 local $self->{in_drain} = 1;
268 631
269 while (my $len = length $self->{rbuf}) { 632 while (my $len = length $self->{rbuf}) {
270 no strict 'refs'; 633 no strict 'refs';
271 if (@{ $self->{queue} }) { 634 if (my $cb = shift @{ $self->{_queue} }) {
272 if ($self->{queue}[0]($self)) { 635 unless ($cb->($self)) {
273 shift @{ $self->{queue} };
274 } elsif ($self->{eof}) { 636 if ($self->{_eof}) {
275 # no progress can be made (not enough data and no data forthcoming) 637 # no progress can be made (not enough data and no data forthcoming)
276 $! = &Errno::EPIPE; return $self->error; 638 $! = &Errno::EPIPE;
277 } else { 639 $self->error;
640 }
641
642 unshift @{ $self->{_queue} }, $cb;
278 return; 643 return;
279 } 644 }
280 } elsif ($self->{on_read}) { 645 } elsif ($self->{on_read}) {
281 $self->{on_read}($self); 646 $self->{on_read}($self);
282 647
283 if ( 648 if (
284 $self->{eof} # if no further data will arrive 649 $self->{_eof} # if no further data will arrive
285 && $len == length $self->{rbuf} # and no data has been consumed 650 && $len == length $self->{rbuf} # and no data has been consumed
286 && !@{ $self->{queue} } # and the queue is still empty 651 && !@{ $self->{_queue} } # and the queue is still empty
287 && $self->{on_read} # and we still want to read data 652 && $self->{on_read} # and we still want to read data
288 ) { 653 ) {
289 # then no progress can be made 654 # then no progress can be made
290 $! = &Errno::EPIPE; return $self->error; 655 $! = &Errno::EPIPE;
656 $self->error;
291 } 657 }
292 } else { 658 } else {
293 # read side becomes idle 659 # read side becomes idle
294 delete $self->{rw}; 660 delete $self->{_rw};
295 return; 661 return;
296 } 662 }
297 } 663 }
298 664
299 if ($self->{eof}) { 665 if ($self->{_eof}) {
300 $self->_shutdown; 666 $self->_shutdown;
301 $self->{on_eof}($self); 667 $self->{on_eof}($self)
668 if $self->{on_eof};
302 } 669 }
303} 670}
304 671
305=item $handle->on_read ($cb) 672=item $handle->on_read ($cb)
306 673
312 679
313sub on_read { 680sub on_read {
314 my ($self, $cb) = @_; 681 my ($self, $cb) = @_;
315 682
316 $self->{on_read} = $cb; 683 $self->{on_read} = $cb;
317
318 unless ($self->{rw} || $self->{eof}) {
319 Scalar::Util::weaken $self;
320
321 $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
324 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
331 } elsif (defined $len) {
332 $self->{eof} = 1;
333 delete $self->{rw};
334
335 } elsif ($! != EAGAIN && $! != EINTR) {
336 return $self->error;
337 }
338
339 $self->_drain_rbuf;
340 });
341 }
342} 684}
343 685
344=item $handle->rbuf 686=item $handle->rbuf
345 687
346Returns the read buffer (as a modifiable lvalue). 688Returns the read buffer (as a modifiable lvalue).
365Append the given callback to the end of the queue (C<push_read>) or 707Append the given callback to the end of the queue (C<push_read>) or
366prepend it (C<unshift_read>). 708prepend it (C<unshift_read>).
367 709
368The callback is called each time some additional read data arrives. 710The callback is called each time some additional read data arrives.
369 711
370It must check wether enough data is in the read buffer already. 712It must check whether enough data is in the read buffer already.
371 713
372If not enough data is available, it must return the empty list or a false 714If not enough data is available, it must return the empty list or a false
373value, in which case it will be called repeatedly until enough data is 715value, in which case it will be called repeatedly until enough data is
374available (or an error condition is detected). 716available (or an error condition is detected).
375 717
377interested in (which can be none at all) and return a true value. After returning 719interested in (which can be none at all) and return a true value. After returning
378true, it will be removed from the queue. 720true, it will be removed from the queue.
379 721
380=cut 722=cut
381 723
724our %RH;
725
726sub register_read_type($$) {
727 $RH{$_[0]} = $_[1];
728}
729
382sub push_read { 730sub push_read {
383 my ($self, $cb) = @_; 731 my $self = shift;
732 my $cb = pop;
384 733
734 if (@_) {
735 my $type = shift;
736
737 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
738 ->($self, $cb, @_);
739 }
740
385 push @{ $self->{queue} }, $cb; 741 push @{ $self->{_queue} }, $cb;
386 $self->_drain_rbuf; 742 $self->_drain_rbuf;
387} 743}
388 744
389sub unshift_read { 745sub unshift_read {
390 my ($self, $cb) = @_; 746 my $self = shift;
747 my $cb = pop;
391 748
749 if (@_) {
750 my $type = shift;
751
752 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
753 ->($self, $cb, @_);
754 }
755
756
392 push @{ $self->{queue} }, $cb; 757 unshift @{ $self->{_queue} }, $cb;
393 $self->_drain_rbuf; 758 $self->_drain_rbuf;
394} 759}
395 760
396=item $handle->push_read_chunk ($len, $cb->($self, $data)) 761=item $handle->push_read (type => @args, $cb)
397 762
398=item $handle->unshift_read_chunk ($len, $cb->($self, $data)) 763=item $handle->unshift_read (type => @args, $cb)
399 764
400Append the given callback to the end of the queue (C<push_read_chunk>) or 765Instead of providing a callback that parses the data itself you can chose
401prepend it (C<unshift_read_chunk>). 766between a number of predefined parsing formats, for chunks of data, lines
767etc.
402 768
403The callback will be called only once C<$len> bytes have been read, and 769Predefined types are (if you have ideas for additional types, feel free to
404these C<$len> bytes will be passed to the callback. 770drop by and tell us):
405 771
406=cut 772=over 4
407 773
408sub _read_chunk($$) { 774=item chunk => $octets, $cb->($handle, $data)
409 my ($len, $cb) = @_; 775
776Invoke the callback only once C<$octets> bytes have been read. Pass the
777data read to the callback. The callback will never be called with less
778data.
779
780Example: read 2 bytes.
781
782 $handle->push_read (chunk => 2, sub {
783 warn "yay ", unpack "H*", $_[1];
784 });
785
786=cut
787
788register_read_type chunk => sub {
789 my ($self, $cb, $len) = @_;
410 790
411 sub { 791 sub {
412 $len <= length $_[0]{rbuf} or return; 792 $len <= length $_[0]{rbuf} or return;
413 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 793 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
414 1 794 1
415 } 795 }
416} 796};
417 797
798# compatibility with older API
418sub push_read_chunk { 799sub push_read_chunk {
419 my ($self, $len, $cb) = @_; 800 $_[0]->push_read (chunk => $_[1], $_[2]);
420
421 $self->push_read (_read_chunk $len, $cb);
422} 801}
423
424 802
425sub unshift_read_chunk { 803sub unshift_read_chunk {
426 my ($self, $len, $cb) = @_; 804 $_[0]->unshift_read (chunk => $_[1], $_[2]);
427
428 $self->unshift_read (_read_chunk $len, $cb);
429} 805}
430 806
431=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol)) 807=item line => [$eol, ]$cb->($handle, $line, $eol)
432
433=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
434
435Append the given callback to the end of the queue (C<push_read_line>) or
436prepend it (C<unshift_read_line>).
437 808
438The callback will be called only once a full line (including the end of 809The callback will be called only once a full line (including the end of
439line marker, C<$eol>) has been read. This line (excluding the end of line 810line marker, C<$eol>) has been read. This line (excluding the end of line
440marker) will be passed to the callback as second argument (C<$line>), and 811marker) will be passed to the callback as second argument (C<$line>), and
441the end of line marker as the third argument (C<$eol>). 812the end of line marker as the third argument (C<$eol>).
452Partial lines at the end of the stream will never be returned, as they are 823Partial lines at the end of the stream will never be returned, as they are
453not marked by the end of line marker. 824not marked by the end of line marker.
454 825
455=cut 826=cut
456 827
457sub _read_line($$) { 828register_read_type line => sub {
458 my $cb = pop; 829 my ($self, $cb, $eol) = @_;
459 my $eol = @_ ? shift : qr|(\015?\012)|;
460 my $pos;
461 830
831 $eol = qr|(\015?\012)| if @_ < 3;
462 $eol = qr|(\Q$eol\E)| unless ref $eol; 832 $eol = quotemeta $eol unless ref $eol;
463 $eol = qr|^(.*?)($eol)|; 833 $eol = qr|^(.*?)($eol)|s;
464 834
465 sub { 835 sub {
466 $_[0]{rbuf} =~ s/$eol// or return; 836 $_[0]{rbuf} =~ s/$eol// or return;
467 837
468 $cb->($1, $2); 838 $cb->($_[0], $1, $2);
469 1 839 1
470 } 840 }
471} 841};
472 842
843# compatibility with older API
473sub push_read_line { 844sub push_read_line {
474 my $self = shift; 845 my $self = shift;
475
476 $self->push_read (&_read_line); 846 $self->push_read (line => @_);
477} 847}
478 848
479sub unshift_read_line { 849sub unshift_read_line {
480 my $self = shift; 850 my $self = shift;
481
482 $self->unshift_read (&_read_line); 851 $self->unshift_read (line => @_);
483} 852}
853
854=item netstring => $cb->($handle, $string)
855
856A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
857
858Throws an error with C<$!> set to EBADMSG on format violations.
859
860=cut
861
862register_read_type netstring => sub {
863 my ($self, $cb) = @_;
864
865 sub {
866 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
867 if ($_[0]{rbuf} =~ /[^0-9]/) {
868 $! = &Errno::EBADMSG;
869 $self->error;
870 }
871 return;
872 }
873
874 my $len = $1;
875
876 $self->unshift_read (chunk => $len, sub {
877 my $string = $_[1];
878 $_[0]->unshift_read (chunk => 1, sub {
879 if ($_[1] eq ",") {
880 $cb->($_[0], $string);
881 } else {
882 $! = &Errno::EBADMSG;
883 $self->error;
884 }
885 });
886 });
887
888 1
889 }
890};
891
892=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
893
894Makes a regex match against the regex object C<$accept> and returns
895everything up to and including the match.
896
897Example: read a single line terminated by '\n'.
898
899 $handle->push_read (regex => qr<\n>, sub { ... });
900
901If C<$reject> is given and not undef, then it determines when the data is
902to be rejected: it is matched against the data when the C<$accept> regex
903does not match and generates an C<EBADMSG> error when it matches. This is
904useful to quickly reject wrong data (to avoid waiting for a timeout or a
905receive buffer overflow).
906
907Example: expect a single decimal number followed by whitespace, reject
908anything else (not the use of an anchor).
909
910 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
911
912If C<$skip> is given and not C<undef>, then it will be matched against
913the receive buffer when neither C<$accept> nor C<$reject> match,
914and everything preceding and including the match will be accepted
915unconditionally. This is useful to skip large amounts of data that you
916know cannot be matched, so that the C<$accept> or C<$reject> regex do not
917have to start matching from the beginning. This is purely an optimisation
918and is usually worth only when you expect more than a few kilobytes.
919
920Example: expect a http header, which ends at C<\015\012\015\012>. Since we
921expect the header to be very large (it isn't in practise, but...), we use
922a skip regex to skip initial portions. The skip regex is tricky in that
923it only accepts something not ending in either \015 or \012, as these are
924required for the accept regex.
925
926 $handle->push_read (regex =>
927 qr<\015\012\015\012>,
928 undef, # no reject
929 qr<^.*[^\015\012]>,
930 sub { ... });
931
932=cut
933
934register_read_type regex => sub {
935 my ($self, $cb, $accept, $reject, $skip) = @_;
936
937 my $data;
938 my $rbuf = \$self->{rbuf};
939
940 sub {
941 # accept
942 if ($$rbuf =~ $accept) {
943 $data .= substr $$rbuf, 0, $+[0], "";
944 $cb->($self, $data);
945 return 1;
946 }
947
948 # reject
949 if ($reject && $$rbuf =~ $reject) {
950 $! = &Errno::EBADMSG;
951 $self->error;
952 }
953
954 # skip
955 if ($skip && $$rbuf =~ $skip) {
956 $data .= substr $$rbuf, 0, $+[0], "";
957 }
958
959 ()
960 }
961};
962
963=item json => $cb->($handle, $hash_or_arrayref)
964
965Reads a JSON object or array, decodes it and passes it to the callback.
966
967If a C<json> object was passed to the constructor, then that will be used
968for the final decode, otherwise it will create a JSON coder expecting UTF-8.
969
970This read type uses the incremental parser available with JSON version
9712.09 (and JSON::XS version 2.2) and above. You have to provide a
972dependency on your own: this module will load the JSON module, but
973AnyEvent does not depend on it itself.
974
975Since JSON texts are fully self-delimiting, the C<json> read and write
976types are an ideal simple RPC protocol: just exchange JSON datagrams. See
977the C<json> write type description, above, for an actual example.
978
979=cut
980
981register_read_type json => sub {
982 my ($self, $cb, $accept, $reject, $skip) = @_;
983
984 require JSON;
985
986 my $data;
987 my $rbuf = \$self->{rbuf};
988
989 my $json = $self->{json} ||= JSON->new->utf8;
990
991 sub {
992 my $ref = $json->incr_parse ($self->{rbuf});
993
994 if ($ref) {
995 $self->{rbuf} = $json->incr_text;
996 $json->incr_text = "";
997 $cb->($self, $ref);
998
999 1
1000 } else {
1001 $self->{rbuf} = "";
1002 ()
1003 }
1004 }
1005};
484 1006
485=back 1007=back
486 1008
1009=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1010
1011This function (not method) lets you add your own types to C<push_read>.
1012
1013Whenever the given C<type> is used, C<push_read> will invoke the code
1014reference with the handle object, the callback and the remaining
1015arguments.
1016
1017The code reference is supposed to return a callback (usually a closure)
1018that works as a plain read callback (see C<< ->push_read ($cb) >>).
1019
1020It should invoke the passed callback when it is done reading (remember to
1021pass C<$handle> as first argument as all other callbacks do that).
1022
1023Note that this is a function, and all types registered this way will be
1024global, so try to use unique names.
1025
1026For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
1027search for C<register_read_type>)).
1028
1029=item $handle->stop_read
1030
1031=item $handle->start_read
1032
1033In rare cases you actually do not want to read anything from the
1034socket. In this case you can call C<stop_read>. Neither C<on_read> no
1035any queued callbacks will be executed then. To start reading again, call
1036C<start_read>.
1037
1038=cut
1039
1040sub stop_read {
1041 my ($self) = @_;
1042
1043 delete $self->{_rw};
1044}
1045
1046sub start_read {
1047 my ($self) = @_;
1048
1049 unless ($self->{_rw} || $self->{_eof}) {
1050 Scalar::Util::weaken $self;
1051
1052 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1053 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
1054 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1055
1056 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now;
1058
1059 $self->{filter_r}
1060 ? $self->{filter_r}->($self, $rbuf)
1061 : $self->_drain_rbuf;
1062
1063 } elsif (defined $len) {
1064 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
1067 $self->{_eof} = 1;
1068 $self->_drain_rbuf;
1069
1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1071 return $self->error;
1072 }
1073 });
1074 }
1075}
1076
1077sub _dotls {
1078 my ($self) = @_;
1079
1080 if (length $self->{_tls_wbuf}) {
1081 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1082 substr $self->{_tls_wbuf}, 0, $len, "";
1083 }
1084 }
1085
1086 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1087 $self->{wbuf} .= $buf;
1088 $self->_drain_wbuf;
1089 }
1090
1091 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
1092 $self->{rbuf} .= $buf;
1093 $self->_drain_rbuf;
1094 }
1095
1096 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1097
1098 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1099 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1100 $self->error;
1101 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1102 $! = &Errno::EIO;
1103 $self->error;
1104 }
1105
1106 # all others are fine for our purposes
1107 }
1108}
1109
1110=item $handle->starttls ($tls[, $tls_ctx])
1111
1112Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1113object is created, you can also do that at a later time by calling
1114C<starttls>.
1115
1116The first argument is the same as the C<tls> constructor argument (either
1117C<"connect">, C<"accept"> or an existing Net::SSLeay object).
1118
1119The second argument is the optional C<Net::SSLeay::CTX> object that is
1120used when AnyEvent::Handle has to create its own TLS connection object.
1121
1122The TLS connection object will end up in C<< $handle->{tls} >> after this
1123call and can be used or changed to your liking. Note that the handshake
1124might have already started when this function returns.
1125
1126=cut
1127
1128# TODO: maybe document...
1129sub starttls {
1130 my ($self, $ssl, $ctx) = @_;
1131
1132 $self->stoptls;
1133
1134 if ($ssl eq "accept") {
1135 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
1136 Net::SSLeay::set_accept_state ($ssl);
1137 } elsif ($ssl eq "connect") {
1138 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
1139 Net::SSLeay::set_connect_state ($ssl);
1140 }
1141
1142 $self->{tls} = $ssl;
1143
1144 # basically, this is deep magic (because SSL_read should have the same issues)
1145 # but the openssl maintainers basically said: "trust us, it just works".
1146 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
1147 # and mismaintained ssleay-module doesn't even offer them).
1148 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1149 Net::SSLeay::CTX_set_mode ($self->{tls},
1150 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1151 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1152
1153 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1154 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1155
1156 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1157
1158 $self->{filter_w} = sub {
1159 $_[0]{_tls_wbuf} .= ${$_[1]};
1160 &_dotls;
1161 };
1162 $self->{filter_r} = sub {
1163 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1164 &_dotls;
1165 };
1166}
1167
1168=item $handle->stoptls
1169
1170Destroys the SSL connection, if any. Partial read or write data will be
1171lost.
1172
1173=cut
1174
1175sub stoptls {
1176 my ($self) = @_;
1177
1178 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
1179
1180 delete $self->{_rbio};
1181 delete $self->{_wbio};
1182 delete $self->{_tls_wbuf};
1183 delete $self->{filter_r};
1184 delete $self->{filter_w};
1185}
1186
1187sub DESTROY {
1188 my $self = shift;
1189
1190 $self->stoptls;
1191}
1192
1193=item AnyEvent::Handle::TLS_CTX
1194
1195This function creates and returns the Net::SSLeay::CTX object used by
1196default for TLS mode.
1197
1198The context is created like this:
1199
1200 Net::SSLeay::load_error_strings;
1201 Net::SSLeay::SSLeay_add_ssl_algorithms;
1202 Net::SSLeay::randomize;
1203
1204 my $CTX = Net::SSLeay::CTX_new;
1205
1206 Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
1207
1208=cut
1209
1210our $TLS_CTX;
1211
1212sub TLS_CTX() {
1213 $TLS_CTX || do {
1214 require Net::SSLeay;
1215
1216 Net::SSLeay::load_error_strings ();
1217 Net::SSLeay::SSLeay_add_ssl_algorithms ();
1218 Net::SSLeay::randomize ();
1219
1220 $TLS_CTX = Net::SSLeay::CTX_new ();
1221
1222 Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
1223
1224 $TLS_CTX
1225 }
1226}
1227
1228=back
1229
1230=head1 SUBCLASSING AnyEvent::Handle
1231
1232In many cases, you might want to subclass AnyEvent::Handle.
1233
1234To make this easier, a given version of AnyEvent::Handle uses these
1235conventions:
1236
1237=over 4
1238
1239=item * all constructor arguments become object members.
1240
1241At least initially, when you pass a C<tls>-argument to the constructor it
1242will end up in C<< $handle->{tls} >>. Those members might be changes or
1243mutated later on (for example C<tls> will hold the TLS connection object).
1244
1245=item * other object member names are prefixed with an C<_>.
1246
1247All object members not explicitly documented (internal use) are prefixed
1248with an underscore character, so the remaining non-C<_>-namespace is free
1249for use for subclasses.
1250
1251=item * all members not documented here and not prefixed with an underscore
1252are free to use in subclasses.
1253
1254Of course, new versions of AnyEvent::Handle may introduce more "public"
1255member variables, but thats just life, at least it is documented.
1256
1257=back
1258
487=head1 AUTHOR 1259=head1 AUTHOR
488 1260
489Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 1261Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
490 1262
491=cut 1263=cut

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines