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.27 by root, Sat May 24 15:26:04 2008 UTC vs.
Revision 1.79 by root, Sun Jul 27 08:37:56 2008 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines