ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
Revision: 1.36
Committed: Mon May 26 18:26:52 2008 UTC (15 years, 11 months ago) by root
Branch: MAIN
Changes since 1.35: +71 -0 lines
Log Message:
*** empty log message ***

File Contents

# 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 root 1.33 use AnyEvent::Util qw(WSAWOULDBLOCK);
8 root 1.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 root 1.22 AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 elmex 1.1
17     =cut
18    
19 root 1.15 our $VERSION = '0.04';
20 elmex 1.1
21     =head1 SYNOPSIS
22    
23     use AnyEvent;
24     use AnyEvent::Handle;
25    
26     my $cv = AnyEvent->condvar;
27    
28 root 1.31 my $handle =
29 elmex 1.2 AnyEvent::Handle->new (
30     fh => \*STDIN,
31     on_eof => sub {
32     $cv->broadcast;
33     },
34     );
35    
36 root 1.31 # 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;
47 elmex 1.1
48     =head1 DESCRIPTION
49    
50 root 1.8 This module is a helper module to make it easier to do event-based I/O on
51 elmex 1.13 filehandles. For utility functions for doing non-blocking connects and accepts
52     on sockets see L<AnyEvent::Util>.
53 root 1.8
54     In the following, when the documentation refers to of "bytes" then this
55     means characters. As sysread and syswrite are used for all I/O, their
56     treatment of characters applies to this module as well.
57 elmex 1.1
58 root 1.8 All callbacks will be invoked with the handle object as their first
59     argument.
60 elmex 1.1
61     =head1 METHODS
62    
63     =over 4
64    
65     =item B<new (%args)>
66    
67 root 1.8 The constructor supports these arguments (all as key => value pairs).
68 elmex 1.1
69     =over 4
70    
71 root 1.8 =item fh => $filehandle [MANDATORY]
72 elmex 1.1
73     The filehandle this L<AnyEvent::Handle> object will operate on.
74    
75 root 1.8 NOTE: The filehandle will be set to non-blocking (using
76     AnyEvent::Util::fh_nonblocking).
77    
78 root 1.16 =item on_eof => $cb->($self)
79 root 1.10
80     Set the callback to be called on EOF.
81 root 1.8
82 root 1.16 While not mandatory, it is highly recommended to set an eof callback,
83     otherwise you might end up with a closed socket while you are still
84     waiting for data.
85    
86 root 1.10 =item on_error => $cb->($self)
87    
88     This is the fatal error callback, that is called when, well, a fatal error
89 elmex 1.20 occurs, such as not being able to resolve the hostname, failure to connect
90 root 1.10 or a read error.
91 root 1.8
92     The object will not be in a usable state when this callback has been
93     called.
94    
95 root 1.10 On callback entrance, the value of C<$!> contains the operating system
96 root 1.29 error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>).
97 root 1.8
98 root 1.10 While not mandatory, it is I<highly> recommended to set this callback, as
99     you will not be notified of errors otherwise. The default simply calls
100     die.
101 root 1.8
102     =item on_read => $cb->($self)
103    
104     This sets the default read callback, which is called when data arrives
105 root 1.10 and no read request is in the queue.
106 root 1.8
107     To access (and remove data from) the read buffer, use the C<< ->rbuf >>
108 elmex 1.20 method or access the C<$self->{rbuf}> member directly.
109 root 1.8
110     When an EOF condition is detected then AnyEvent::Handle will first try to
111     feed all the remaining data to the queued callbacks and C<on_read> before
112     calling the C<on_eof> callback. If no progress can be made, then a fatal
113     error will be raised (with C<$!> set to C<EPIPE>).
114 elmex 1.1
115 root 1.8 =item on_drain => $cb->()
116 elmex 1.1
117 root 1.8 This sets the callback that is called when the write buffer becomes empty
118     (or when the callback is set and the buffer is empty already).
119 elmex 1.1
120 root 1.8 To append to the write buffer, use the C<< ->push_write >> method.
121 elmex 1.2
122 root 1.8 =item rbuf_max => <bytes>
123 elmex 1.2
124 root 1.8 If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
125     when the read buffer ever (strictly) exceeds this size. This is useful to
126     avoid denial-of-service attacks.
127 elmex 1.2
128 root 1.8 For example, a server accepting connections from untrusted sources should
129     be configured to accept only so-and-so much data that it cannot act on
130     (for example, when expecting a line, an attacker could send an unlimited
131     amount of data without a callback ever being called as long as the line
132     isn't finished).
133 elmex 1.2
134 root 1.8 =item read_size => <bytes>
135 elmex 1.2
136 root 1.8 The default read block size (the amount of bytes this module will try to read
137     on each [loop iteration). Default: C<4096>.
138    
139     =item low_water_mark => <bytes>
140    
141     Sets the amount of bytes (default: C<0>) that make up an "empty" write
142     buffer: If the write reaches this size or gets even samller it is
143     considered empty.
144 elmex 1.2
145 root 1.19 =item tls => "accept" | "connect" | Net::SSLeay::SSL object
146    
147     When this parameter is given, it enables TLS (SSL) mode, that means it
148     will start making tls handshake and will transparently encrypt/decrypt
149     data.
150    
151 root 1.26 TLS mode requires Net::SSLeay to be installed (it will be loaded
152     automatically when you try to create a TLS handle).
153    
154 root 1.19 For the TLS server side, use C<accept>, and for the TLS client side of a
155     connection, use C<connect> mode.
156    
157     You can also provide your own TLS connection object, but you have
158     to make sure that you call either C<Net::SSLeay::set_connect_state>
159     or C<Net::SSLeay::set_accept_state> on it before you pass it to
160     AnyEvent::Handle.
161    
162 root 1.26 See the C<starttls> method if you need to start TLs negotiation later.
163    
164 root 1.19 =item tls_ctx => $ssl_ctx
165    
166     Use the given Net::SSLeay::CTX object to create the new TLS connection
167     (unless a connection object was specified directly). If this parameter is
168     missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
169    
170 elmex 1.1 =back
171    
172     =cut
173    
174     sub new {
175 root 1.8 my $class = shift;
176    
177     my $self = bless { @_ }, $class;
178    
179     $self->{fh} or Carp::croak "mandatory argument fh is missing";
180    
181     AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
182 elmex 1.1
183 root 1.19 if ($self->{tls}) {
184     require Net::SSLeay;
185     $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
186     }
187    
188 root 1.16 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof};
189 root 1.10 $self->on_error (delete $self->{on_error}) if $self->{on_error};
190 root 1.8 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
191     $self->on_read (delete $self->{on_read} ) if $self->{on_read};
192 elmex 1.1
193 root 1.10 $self->start_read;
194    
195 root 1.8 $self
196     }
197 elmex 1.2
198 root 1.8 sub _shutdown {
199     my ($self) = @_;
200 elmex 1.2
201 root 1.8 delete $self->{rw};
202     delete $self->{ww};
203     delete $self->{fh};
204     }
205    
206     sub error {
207     my ($self) = @_;
208    
209     {
210     local $!;
211     $self->_shutdown;
212 elmex 1.1 }
213    
214 root 1.10 if ($self->{on_error}) {
215     $self->{on_error}($self);
216     } else {
217 root 1.29 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
218 root 1.10 }
219 elmex 1.1 }
220    
221 root 1.8 =item $fh = $handle->fh
222 elmex 1.1
223 root 1.22 This method returns the file handle of the L<AnyEvent::Handle> object.
224 elmex 1.1
225     =cut
226    
227     sub fh { $_[0]->{fh} }
228    
229 root 1.8 =item $handle->on_error ($cb)
230 elmex 1.1
231 root 1.8 Replace the current C<on_error> callback (see the C<on_error> constructor argument).
232 elmex 1.1
233 root 1.8 =cut
234    
235     sub on_error {
236     $_[0]{on_error} = $_[1];
237     }
238    
239     =item $handle->on_eof ($cb)
240    
241     Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
242 elmex 1.1
243     =cut
244    
245 root 1.8 sub on_eof {
246     $_[0]{on_eof} = $_[1];
247     }
248    
249 root 1.9 #############################################################################
250    
251     =back
252    
253     =head2 WRITE QUEUE
254    
255     AnyEvent::Handle manages two queues per handle, one for writing and one
256     for reading.
257    
258     The write queue is very simple: you can add data to its end, and
259     AnyEvent::Handle will automatically try to get rid of it for you.
260    
261 elmex 1.20 When data could be written and the write buffer is shorter then the low
262 root 1.9 water mark, the C<on_drain> callback will be invoked.
263    
264     =over 4
265    
266 root 1.8 =item $handle->on_drain ($cb)
267    
268     Sets the C<on_drain> callback or clears it (see the description of
269     C<on_drain> in the constructor).
270    
271     =cut
272    
273     sub on_drain {
274 elmex 1.1 my ($self, $cb) = @_;
275    
276 root 1.8 $self->{on_drain} = $cb;
277    
278     $cb->($self)
279     if $cb && $self->{low_water_mark} >= length $self->{wbuf};
280     }
281    
282     =item $handle->push_write ($data)
283    
284     Queues the given scalar to be written. You can push as much data as you
285     want (only limited by the available memory), as C<AnyEvent::Handle>
286     buffers it independently of the kernel.
287    
288     =cut
289    
290 root 1.17 sub _drain_wbuf {
291     my ($self) = @_;
292 root 1.8
293 root 1.29 if (!$self->{ww} && length $self->{wbuf}) {
294 root 1.35
295 root 1.8 Scalar::Util::weaken $self;
296 root 1.35
297 root 1.8 my $cb = sub {
298     my $len = syswrite $self->{fh}, $self->{wbuf};
299    
300 root 1.29 if ($len >= 0) {
301 root 1.8 substr $self->{wbuf}, 0, $len, "";
302    
303     $self->{on_drain}($self)
304     if $self->{low_water_mark} >= length $self->{wbuf}
305     && $self->{on_drain};
306    
307     delete $self->{ww} unless length $self->{wbuf};
308 root 1.33 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) {
309 root 1.8 $self->error;
310 elmex 1.1 }
311 root 1.8 };
312    
313 root 1.35 # try to write data immediately
314     $cb->();
315 root 1.8
316 root 1.35 # if still data left in wbuf, we need to poll
317     $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
318     if length $self->{wbuf};
319 root 1.8 };
320     }
321    
322 root 1.30 our %WH;
323    
324     sub register_write_type($$) {
325     $WH{$_[0]} = $_[1];
326     }
327    
328 root 1.17 sub push_write {
329     my $self = shift;
330    
331 root 1.29 if (@_ > 1) {
332     my $type = shift;
333    
334     @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
335     ->($self, @_);
336     }
337    
338 root 1.17 if ($self->{filter_w}) {
339 root 1.18 $self->{filter_w}->($self, \$_[0]);
340 root 1.17 } else {
341     $self->{wbuf} .= $_[0];
342     $self->_drain_wbuf;
343     }
344     }
345    
346 root 1.29 =item $handle->push_write (type => @args)
347    
348     =item $handle->unshift_write (type => @args)
349    
350     Instead of formatting your data yourself, you can also let this module do
351     the job by specifying a type and type-specific arguments.
352    
353 root 1.30 Predefined types are (if you have ideas for additional types, feel free to
354     drop by and tell us):
355 root 1.29
356     =over 4
357    
358     =item netstring => $string
359    
360     Formats the given value as netstring
361     (http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
362    
363 root 1.30 =back
364    
365 root 1.29 =cut
366    
367     register_write_type netstring => sub {
368     my ($self, $string) = @_;
369    
370     sprintf "%d:%s,", (length $string), $string
371     };
372    
373 root 1.30 =item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
374    
375     This function (not method) lets you add your own types to C<push_write>.
376     Whenever the given C<type> is used, C<push_write> will invoke the code
377     reference with the handle object and the remaining arguments.
378 root 1.29
379 root 1.30 The code reference is supposed to return a single octet string that will
380     be appended to the write buffer.
381 root 1.29
382 root 1.30 Note that this is a function, and all types registered this way will be
383     global, so try to use unique names.
384 root 1.29
385 root 1.30 =cut
386 root 1.29
387 root 1.8 #############################################################################
388    
389 root 1.9 =back
390    
391     =head2 READ QUEUE
392    
393     AnyEvent::Handle manages two queues per handle, one for writing and one
394     for reading.
395    
396     The read queue is more complex than the write queue. It can be used in two
397     ways, the "simple" way, using only C<on_read> and the "complex" way, using
398     a queue.
399    
400     In the simple case, you just install an C<on_read> callback and whenever
401     new data arrives, it will be called. You can then remove some data (if
402     enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
403     or not.
404    
405     In the more complex case, you want to queue multiple callbacks. In this
406     case, AnyEvent::Handle will call the first queued callback each time new
407     data arrives and removes it when it has done its job (see C<push_read>,
408     below).
409    
410     This way you can, for example, push three line-reads, followed by reading
411     a chunk of data, and AnyEvent::Handle will execute them in order.
412    
413     Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
414     the specified number of bytes which give an XML datagram.
415    
416     # in the default state, expect some header bytes
417     $handle->on_read (sub {
418     # some data is here, now queue the length-header-read (4 octets)
419     shift->unshift_read_chunk (4, sub {
420     # header arrived, decode
421     my $len = unpack "N", $_[1];
422    
423     # now read the payload
424     shift->unshift_read_chunk ($len, sub {
425     my $xml = $_[1];
426     # handle xml
427     });
428     });
429     });
430    
431     Example 2: Implement a client for a protocol that replies either with
432     "OK" and another line or "ERROR" for one request, and 64 bytes for the
433     second request. Due tot he availability of a full queue, we can just
434     pipeline sending both requests and manipulate the queue as necessary in
435     the callbacks:
436    
437     # request one
438     $handle->push_write ("request 1\015\012");
439    
440     # we expect "ERROR" or "OK" as response, so push a line read
441     $handle->push_read_line (sub {
442     # if we got an "OK", we have to _prepend_ another line,
443     # so it will be read before the second request reads its 64 bytes
444     # which are already in the queue when this callback is called
445     # we don't do this in case we got an error
446     if ($_[1] eq "OK") {
447     $_[0]->unshift_read_line (sub {
448     my $response = $_[1];
449     ...
450     });
451     }
452     });
453    
454     # request two
455     $handle->push_write ("request 2\015\012");
456    
457     # simply read 64 bytes, always
458     $handle->push_read_chunk (64, sub {
459     my $response = $_[1];
460     ...
461     });
462    
463     =over 4
464    
465 root 1.10 =cut
466    
467 root 1.8 sub _drain_rbuf {
468     my ($self) = @_;
469 elmex 1.1
470 root 1.17 if (
471     defined $self->{rbuf_max}
472     && $self->{rbuf_max} < length $self->{rbuf}
473     ) {
474     $! = &Errno::ENOSPC; return $self->error;
475     }
476    
477 root 1.11 return if $self->{in_drain};
478 root 1.8 local $self->{in_drain} = 1;
479 elmex 1.1
480 root 1.8 while (my $len = length $self->{rbuf}) {
481     no strict 'refs';
482 root 1.10 if (my $cb = shift @{ $self->{queue} }) {
483 root 1.29 unless ($cb->($self)) {
484 root 1.10 if ($self->{eof}) {
485     # no progress can be made (not enough data and no data forthcoming)
486     $! = &Errno::EPIPE; return $self->error;
487     }
488    
489     unshift @{ $self->{queue} }, $cb;
490 root 1.8 return;
491     }
492     } elsif ($self->{on_read}) {
493     $self->{on_read}($self);
494    
495     if (
496     $self->{eof} # if no further data will arrive
497     && $len == length $self->{rbuf} # and no data has been consumed
498     && !@{ $self->{queue} } # and the queue is still empty
499     && $self->{on_read} # and we still want to read data
500     ) {
501     # then no progress can be made
502     $! = &Errno::EPIPE; return $self->error;
503 elmex 1.1 }
504 root 1.8 } else {
505     # read side becomes idle
506     delete $self->{rw};
507     return;
508     }
509     }
510    
511     if ($self->{eof}) {
512     $self->_shutdown;
513 root 1.16 $self->{on_eof}($self)
514     if $self->{on_eof};
515 root 1.8 }
516 elmex 1.1 }
517    
518 root 1.8 =item $handle->on_read ($cb)
519 elmex 1.1
520 root 1.8 This replaces the currently set C<on_read> callback, or clears it (when
521     the new callback is C<undef>). See the description of C<on_read> in the
522     constructor.
523 elmex 1.1
524 root 1.8 =cut
525    
526     sub on_read {
527     my ($self, $cb) = @_;
528 elmex 1.1
529 root 1.8 $self->{on_read} = $cb;
530 elmex 1.1 }
531    
532 root 1.8 =item $handle->rbuf
533    
534     Returns the read buffer (as a modifiable lvalue).
535 elmex 1.1
536 root 1.8 You can access the read buffer directly as the C<< ->{rbuf} >> member, if
537     you want.
538 elmex 1.1
539 root 1.8 NOTE: The read buffer should only be used or modified if the C<on_read>,
540     C<push_read> or C<unshift_read> methods are used. The other read methods
541     automatically manage the read buffer.
542 elmex 1.1
543     =cut
544    
545 elmex 1.2 sub rbuf : lvalue {
546 root 1.8 $_[0]{rbuf}
547 elmex 1.2 }
548 elmex 1.1
549 root 1.8 =item $handle->push_read ($cb)
550    
551     =item $handle->unshift_read ($cb)
552    
553     Append the given callback to the end of the queue (C<push_read>) or
554     prepend it (C<unshift_read>).
555    
556     The callback is called each time some additional read data arrives.
557 elmex 1.1
558 elmex 1.20 It must check whether enough data is in the read buffer already.
559 elmex 1.1
560 root 1.8 If not enough data is available, it must return the empty list or a false
561     value, in which case it will be called repeatedly until enough data is
562     available (or an error condition is detected).
563    
564     If enough data was available, then the callback must remove all data it is
565     interested in (which can be none at all) and return a true value. After returning
566     true, it will be removed from the queue.
567 elmex 1.1
568     =cut
569    
570 root 1.30 our %RH;
571    
572     sub register_read_type($$) {
573     $RH{$_[0]} = $_[1];
574     }
575    
576 root 1.8 sub push_read {
577 root 1.28 my $self = shift;
578     my $cb = pop;
579    
580     if (@_) {
581     my $type = shift;
582    
583     $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
584     ->($self, $cb, @_);
585     }
586 elmex 1.1
587 root 1.8 push @{ $self->{queue} }, $cb;
588     $self->_drain_rbuf;
589 elmex 1.1 }
590    
591 root 1.8 sub unshift_read {
592 root 1.28 my $self = shift;
593     my $cb = pop;
594    
595     if (@_) {
596     my $type = shift;
597    
598     $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
599     ->($self, $cb, @_);
600     }
601    
602 root 1.8
603 root 1.28 unshift @{ $self->{queue} }, $cb;
604 root 1.8 $self->_drain_rbuf;
605     }
606 elmex 1.1
607 root 1.28 =item $handle->push_read (type => @args, $cb)
608 elmex 1.1
609 root 1.28 =item $handle->unshift_read (type => @args, $cb)
610 elmex 1.1
611 root 1.28 Instead of providing a callback that parses the data itself you can chose
612     between a number of predefined parsing formats, for chunks of data, lines
613     etc.
614 elmex 1.1
615 root 1.30 Predefined types are (if you have ideas for additional types, feel free to
616     drop by and tell us):
617 root 1.28
618     =over 4
619    
620     =item chunk => $octets, $cb->($self, $data)
621    
622     Invoke the callback only once C<$octets> bytes have been read. Pass the
623     data read to the callback. The callback will never be called with less
624     data.
625    
626     Example: read 2 bytes.
627    
628     $handle->push_read (chunk => 2, sub {
629     warn "yay ", unpack "H*", $_[1];
630     });
631 elmex 1.1
632     =cut
633    
634 root 1.28 register_read_type chunk => sub {
635     my ($self, $cb, $len) = @_;
636 elmex 1.1
637 root 1.8 sub {
638     $len <= length $_[0]{rbuf} or return;
639 elmex 1.12 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
640 root 1.8 1
641     }
642 root 1.28 };
643 root 1.8
644 root 1.28 # compatibility with older API
645 root 1.8 sub push_read_chunk {
646 root 1.28 $_[0]->push_read (chunk => $_[1], $_[2]);
647 root 1.8 }
648 elmex 1.1
649 root 1.8 sub unshift_read_chunk {
650 root 1.28 $_[0]->unshift_read (chunk => $_[1], $_[2]);
651 elmex 1.1 }
652    
653 root 1.28 =item line => [$eol, ]$cb->($self, $line, $eol)
654 elmex 1.1
655 root 1.8 The callback will be called only once a full line (including the end of
656     line marker, C<$eol>) has been read. This line (excluding the end of line
657     marker) will be passed to the callback as second argument (C<$line>), and
658     the end of line marker as the third argument (C<$eol>).
659 elmex 1.1
660 root 1.8 The end of line marker, C<$eol>, can be either a string, in which case it
661     will be interpreted as a fixed record end marker, or it can be a regex
662     object (e.g. created by C<qr>), in which case it is interpreted as a
663     regular expression.
664 elmex 1.1
665 root 1.8 The end of line marker argument C<$eol> is optional, if it is missing (NOT
666     undef), then C<qr|\015?\012|> is used (which is good for most internet
667     protocols).
668 elmex 1.1
669 root 1.8 Partial lines at the end of the stream will never be returned, as they are
670     not marked by the end of line marker.
671 elmex 1.1
672 root 1.8 =cut
673 elmex 1.1
674 root 1.28 register_read_type line => sub {
675     my ($self, $cb, $eol) = @_;
676 elmex 1.1
677 root 1.28 $eol = qr|(\015?\012)| if @_ < 3;
678 root 1.14 $eol = quotemeta $eol unless ref $eol;
679     $eol = qr|^(.*?)($eol)|s;
680 elmex 1.1
681 root 1.8 sub {
682     $_[0]{rbuf} =~ s/$eol// or return;
683 elmex 1.1
684 elmex 1.12 $cb->($_[0], $1, $2);
685 root 1.8 1
686     }
687 root 1.28 };
688 elmex 1.1
689 root 1.28 # compatibility with older API
690 root 1.8 sub push_read_line {
691 root 1.28 my $self = shift;
692     $self->push_read (line => @_);
693 root 1.10 }
694    
695     sub unshift_read_line {
696 root 1.28 my $self = shift;
697     $self->unshift_read (line => @_);
698 root 1.10 }
699    
700 root 1.29 =item netstring => $cb->($string)
701    
702     A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
703    
704     Throws an error with C<$!> set to EBADMSG on format violations.
705    
706     =cut
707    
708     register_read_type netstring => sub {
709     my ($self, $cb) = @_;
710    
711     sub {
712     unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
713     if ($_[0]{rbuf} =~ /[^0-9]/) {
714     $! = &Errno::EBADMSG;
715     $self->error;
716     }
717     return;
718     }
719    
720     my $len = $1;
721    
722     $self->unshift_read (chunk => $len, sub {
723     my $string = $_[1];
724     $_[0]->unshift_read (chunk => 1, sub {
725     if ($_[1] eq ",") {
726     $cb->($_[0], $string);
727     } else {
728     $! = &Errno::EBADMSG;
729     $self->error;
730     }
731     });
732     });
733    
734     1
735     }
736     };
737    
738 root 1.36 =item regex => $accept[, $reject[, $skip], $cb->($data)
739    
740     Makes a regex match against the regex object C<$accept> and returns
741     everything up to and including the match.
742    
743     Example: read a single line terminated by '\n'.
744    
745     $handle->push_read (regex => qr<\n>, sub { ... });
746    
747     If C<$reject> is given and not undef, then it determines when the data is
748     to be rejected: it is matched against the data when the C<$accept> regex
749     does not match and generates an C<EBADMSG> error when it matches. This is
750     useful to quickly reject wrong data (to avoid waiting for a timeout or a
751     receive buffer overflow).
752    
753     Example: expect a single decimal number followed by whitespace, reject
754     anything else (not the use of an anchor).
755    
756     $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
757    
758     If C<$skip> is given and not C<undef>, then it will be matched against
759     the receive buffer when neither C<$accept> nor C<$reject> match,
760     and everything preceding and including the match will be accepted
761     unconditionally. This is useful to skip large amounts of data that you
762     know cannot be matched, so that the C<$accept> or C<$reject> regex do not
763     have to start matching from the beginning. This is purely an optimisation
764     and is usually worth only when you expect more than a few kilobytes.
765    
766     Example: expect a http header, which ends at C<\015\012\015\012>. Since we
767     expect the header to be very large (it isn't in practise, but...), we use
768     a skip regex to skip initial portions. The skip regex is tricky in that
769     it only accepts something not ending in either \015 or \012, as these are
770     required for the accept regex.
771    
772     $handle->push_read (regex =>
773     qr<\015\012\015\012>,
774     undef, # no reject
775     qr<^.*[^\015\012]>,
776     sub { ... });
777    
778     =cut
779    
780     register_read_type regex => sub {
781     my ($self, $cb, $accept, $reject, $skip) = @_;
782    
783     my $data;
784     my $rbuf = \$self->{rbuf};
785    
786     sub {
787     # accept
788     if ($$rbuf =~ $accept) {
789     $data .= substr $$rbuf, 0, $+[0], "";
790     $cb->($self, $data);
791     return 1;
792     }
793    
794     # reject
795     if ($reject && $$rbuf =~ $reject) {
796     $! = &Errno::EBADMSG;
797     $self->error;
798     }
799    
800     # skip
801     if ($skip && $$rbuf =~ $skip) {
802     $data .= substr $$rbuf, 0, $+[0], "";
803     }
804    
805     ()
806     }
807     };
808    
809 root 1.28 =back
810    
811 root 1.30 =item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
812    
813     This function (not method) lets you add your own types to C<push_read>.
814    
815     Whenever the given C<type> is used, C<push_read> will invoke the code
816     reference with the handle object, the callback and the remaining
817     arguments.
818    
819     The code reference is supposed to return a callback (usually a closure)
820     that works as a plain read callback (see C<< ->push_read ($cb) >>).
821    
822     It should invoke the passed callback when it is done reading (remember to
823     pass C<$self> as first argument as all other callbacks do that).
824    
825     Note that this is a function, and all types registered this way will be
826     global, so try to use unique names.
827    
828     For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
829     search for C<register_read_type>)).
830    
831 root 1.10 =item $handle->stop_read
832    
833     =item $handle->start_read
834    
835 root 1.18 In rare cases you actually do not want to read anything from the
836 root 1.10 socket. In this case you can call C<stop_read>. Neither C<on_read> no
837 root 1.22 any queued callbacks will be executed then. To start reading again, call
838 root 1.10 C<start_read>.
839    
840     =cut
841    
842     sub stop_read {
843     my ($self) = @_;
844 elmex 1.1
845 root 1.10 delete $self->{rw};
846 root 1.8 }
847 elmex 1.1
848 root 1.10 sub start_read {
849     my ($self) = @_;
850    
851     unless ($self->{rw} || $self->{eof}) {
852     Scalar::Util::weaken $self;
853    
854     $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
855 root 1.17 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
856     my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
857 root 1.10
858     if ($len > 0) {
859 root 1.17 $self->{filter_r}
860 root 1.18 ? $self->{filter_r}->($self, $rbuf)
861 root 1.17 : $self->_drain_rbuf;
862 root 1.10
863     } elsif (defined $len) {
864 root 1.17 delete $self->{rw};
865 root 1.10 $self->{eof} = 1;
866 root 1.17 $self->_drain_rbuf;
867 root 1.10
868 root 1.33 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
869 root 1.10 return $self->error;
870     }
871     });
872     }
873 elmex 1.1 }
874    
875 root 1.19 sub _dotls {
876     my ($self) = @_;
877    
878     if (length $self->{tls_wbuf}) {
879 root 1.22 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) {
880     substr $self->{tls_wbuf}, 0, $len, "";
881     }
882 root 1.19 }
883    
884     if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) {
885     $self->{wbuf} .= $buf;
886     $self->_drain_wbuf;
887     }
888    
889 root 1.23 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
890     $self->{rbuf} .= $buf;
891     $self->_drain_rbuf;
892     }
893    
894 root 1.24 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
895    
896     if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
897 root 1.23 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
898     $self->error;
899     } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
900     $! = &Errno::EIO;
901     $self->error;
902 root 1.19 }
903 root 1.23
904     # all others are fine for our purposes
905 root 1.19 }
906     }
907    
908 root 1.25 =item $handle->starttls ($tls[, $tls_ctx])
909    
910     Instead of starting TLS negotiation immediately when the AnyEvent::Handle
911     object is created, you can also do that at a later time by calling
912     C<starttls>.
913    
914     The first argument is the same as the C<tls> constructor argument (either
915     C<"connect">, C<"accept"> or an existing Net::SSLeay object).
916    
917     The second argument is the optional C<Net::SSLeay::CTX> object that is
918     used when AnyEvent::Handle has to create its own TLS connection object.
919    
920     =cut
921    
922 root 1.19 # TODO: maybe document...
923     sub starttls {
924     my ($self, $ssl, $ctx) = @_;
925    
926 root 1.25 $self->stoptls;
927    
928 root 1.19 if ($ssl eq "accept") {
929     $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
930     Net::SSLeay::set_accept_state ($ssl);
931     } elsif ($ssl eq "connect") {
932     $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
933     Net::SSLeay::set_connect_state ($ssl);
934     }
935    
936     $self->{tls} = $ssl;
937    
938 root 1.21 # basically, this is deep magic (because SSL_read should have the same issues)
939     # but the openssl maintainers basically said: "trust us, it just works".
940     # (unfortunately, we have to hardcode constants because the abysmally misdesigned
941     # and mismaintained ssleay-module doesn't even offer them).
942 root 1.27 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
943 root 1.21 Net::SSLeay::CTX_set_mode ($self->{tls},
944 root 1.34 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
945     | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
946 root 1.21
947 root 1.19 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
948     $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
949    
950     Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});
951    
952     $self->{filter_w} = sub {
953     $_[0]{tls_wbuf} .= ${$_[1]};
954     &_dotls;
955     };
956     $self->{filter_r} = sub {
957     Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]});
958     &_dotls;
959     };
960     }
961    
962 root 1.25 =item $handle->stoptls
963    
964     Destroys the SSL connection, if any. Partial read or write data will be
965     lost.
966    
967     =cut
968    
969     sub stoptls {
970     my ($self) = @_;
971    
972     Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
973     delete $self->{tls_rbio};
974     delete $self->{tls_wbio};
975     delete $self->{tls_wbuf};
976     delete $self->{filter_r};
977     delete $self->{filter_w};
978     }
979    
980 root 1.19 sub DESTROY {
981     my $self = shift;
982    
983 root 1.25 $self->stoptls;
984 root 1.19 }
985    
986     =item AnyEvent::Handle::TLS_CTX
987    
988     This function creates and returns the Net::SSLeay::CTX object used by
989     default for TLS mode.
990    
991     The context is created like this:
992    
993     Net::SSLeay::load_error_strings;
994     Net::SSLeay::SSLeay_add_ssl_algorithms;
995     Net::SSLeay::randomize;
996    
997     my $CTX = Net::SSLeay::CTX_new;
998    
999     Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
1000    
1001     =cut
1002    
1003     our $TLS_CTX;
1004    
1005     sub TLS_CTX() {
1006     $TLS_CTX || do {
1007     require Net::SSLeay;
1008    
1009     Net::SSLeay::load_error_strings ();
1010     Net::SSLeay::SSLeay_add_ssl_algorithms ();
1011     Net::SSLeay::randomize ();
1012    
1013     $TLS_CTX = Net::SSLeay::CTX_new ();
1014    
1015     Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
1016    
1017     $TLS_CTX
1018     }
1019     }
1020    
1021 elmex 1.1 =back
1022    
1023     =head1 AUTHOR
1024    
1025 root 1.8 Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
1026 elmex 1.1
1027     =cut
1028    
1029     1; # End of AnyEvent::Handle