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.43 by root, Wed May 28 23:57:38 2008 UTC vs.
Revision 1.89 by root, Sat Sep 6 10:54:32 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 qw(WSAEWOULDBLOCK); 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);
12use Time::HiRes qw(time);
13 12
14=head1 NAME 13=head1 NAME
15 14
16AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
17 16
18=cut 17=cut
19 18
20our $VERSION = '0.04'; 19our $VERSION = 4.234;
21 20
22=head1 SYNOPSIS 21=head1 SYNOPSIS
23 22
24 use AnyEvent; 23 use AnyEvent;
25 use AnyEvent::Handle; 24 use AnyEvent::Handle;
50 49
51This 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
52filehandles. For utility functions for doing non-blocking connects and accepts 51filehandles. For utility functions for doing non-blocking connects and accepts
53on sockets see L<AnyEvent::Util>. 52on sockets see L<AnyEvent::Util>.
54 53
54The L<AnyEvent::Intro> tutorial contains some well-documented
55AnyEvent::Handle examples.
56
55In the following, when the documentation refers to of "bytes" then this 57In the following, when the documentation refers to of "bytes" then this
56means characters. As sysread and syswrite are used for all I/O, their 58means characters. As sysread and syswrite are used for all I/O, their
57treatment of characters applies to this module as well. 59treatment of characters applies to this module as well.
58 60
59All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
71 73
72=item fh => $filehandle [MANDATORY] 74=item fh => $filehandle [MANDATORY]
73 75
74The filehandle this L<AnyEvent::Handle> object will operate on. 76The filehandle this L<AnyEvent::Handle> object will operate on.
75 77
76NOTE: The filehandle will be set to non-blocking (using 78NOTE: The filehandle will be set to non-blocking mode (using
77AnyEvent::Util::fh_nonblocking). 79C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in
80that mode.
78 81
79=item on_eof => $cb->($handle) 82=item on_eof => $cb->($handle)
80 83
81Set the callback to be called on EOF. 84Set the callback to be called when an end-of-file condition is detected,
85i.e. in the case of a socket, when the other side has closed the
86connection cleanly.
82 87
88For sockets, this just means that the other side has stopped sending data,
89you can still try to write data, and, in fact, one can return from the eof
90callback and continue writing data, as only the read part has been shut
91down.
92
83While not mandatory, it is highly recommended to set an eof callback, 93While not mandatory, it is I<highly> recommended to set an eof callback,
84otherwise you might end up with a closed socket while you are still 94otherwise you might end up with a closed socket while you are still
85waiting for data. 95waiting for data.
86 96
97If an EOF condition has been detected but no C<on_eof> callback has been
98set, then a fatal error will be raised with C<$!> set to <0>.
99
87=item on_error => $cb->($handle) 100=item on_error => $cb->($handle, $fatal)
88 101
89This is the fatal error callback, that is called when, well, a fatal error 102This is the error callback, which is called when, well, some error
90occurs, such as not being able to resolve the hostname, failure to connect 103occured, such as not being able to resolve the hostname, failure to
91or a read error. 104connect or a read error.
92 105
93The object will not be in a usable state when this callback has been 106Some errors are fatal (which is indicated by C<$fatal> being true). On
94called. 107fatal errors the handle object will be shut down and will not be usable
108(but you are free to look at the current C<< ->rbuf >>). Examples of fatal
109errors are an EOF condition with active (but unsatisifable) read watchers
110(C<EPIPE>) or I/O errors.
111
112Non-fatal errors can be retried by simply returning, but it is recommended
113to simply ignore this parameter and instead abondon the handle object
114when this callback is invoked. Examples of non-fatal errors are timeouts
115C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
95 116
96On callback entrance, the value of C<$!> contains the operating system 117On callback entrance, the value of C<$!> contains the operating system
97error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 118error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
98 119
99The callback should throw an exception. If it returns, then
100AnyEvent::Handle will C<croak> for you.
101
102While not mandatory, it is I<highly> recommended to set this callback, as 120While not mandatory, it is I<highly> recommended to set this callback, as
103you will not be notified of errors otherwise. The default simply calls 121you will not be notified of errors otherwise. The default simply calls
104die. 122C<croak>.
105 123
106=item on_read => $cb->($handle) 124=item on_read => $cb->($handle)
107 125
108This sets the default read callback, which is called when data arrives 126This sets the default read callback, which is called when data arrives
109and no read request is in the queue. 127and no read request is in the queue (unlike read queue callbacks, this
128callback will only be called when at least one octet of data is in the
129read buffer).
110 130
111To access (and remove data from) the read buffer, use the C<< ->rbuf >> 131To access (and remove data from) the read buffer, use the C<< ->rbuf >>
112method or access the C<$handle->{rbuf}> member directly. 132method or access the C<$handle->{rbuf}> member directly.
113 133
114When an EOF condition is detected then AnyEvent::Handle will first try to 134When an EOF condition is detected then AnyEvent::Handle will first try to
121This sets the callback that is called when the write buffer becomes empty 141This sets the callback that is called when the write buffer becomes empty
122(or when the callback is set and the buffer is empty already). 142(or when the callback is set and the buffer is empty already).
123 143
124To append to the write buffer, use the C<< ->push_write >> method. 144To append to the write buffer, use the C<< ->push_write >> method.
125 145
146This callback is useful when you don't want to put all of your write data
147into the queue at once, for example, when you want to write the contents
148of some file to the socket you might not want to read the whole file into
149memory and push it into the queue, but instead only read more data from
150the file when the write queue becomes empty.
151
126=item timeout => $fractional_seconds 152=item timeout => $fractional_seconds
127 153
128If non-zero, then this enables an "inactivity" timeout: whenever this many 154If non-zero, then this enables an "inactivity" timeout: whenever this many
129seconds pass without a successful read or write on the underlying file 155seconds pass without a successful read or write on the underlying file
130handle, the C<on_timeout> callback will be invoked (and if that one is 156handle, the C<on_timeout> callback will be invoked (and if that one is
131missing, an C<ETIMEDOUT> errror will be raised). 157missing, a non-fatal C<ETIMEDOUT> error will be raised).
132 158
133Note that timeout processing is also active when you currently do not have 159Note that timeout processing is also active when you currently do not have
134any outstanding read or write requests: If you plan to keep the connection 160any outstanding read or write requests: If you plan to keep the connection
135idle then you should disable the timout temporarily or ignore the timeout 161idle then you should disable the timout temporarily or ignore the timeout
136in the C<on_timeout> callback. 162in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
163restart the timeout.
137 164
138Zero (the default) disables this timeout. 165Zero (the default) disables this timeout.
139 166
140=item on_timeout => $cb->($handle) 167=item on_timeout => $cb->($handle)
141 168
145 172
146=item rbuf_max => <bytes> 173=item rbuf_max => <bytes>
147 174
148If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 175If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
149when the read buffer ever (strictly) exceeds this size. This is useful to 176when the read buffer ever (strictly) exceeds this size. This is useful to
150avoid denial-of-service attacks. 177avoid some forms of denial-of-service attacks.
151 178
152For example, a server accepting connections from untrusted sources should 179For example, a server accepting connections from untrusted sources should
153be configured to accept only so-and-so much data that it cannot act on 180be configured to accept only so-and-so much data that it cannot act on
154(for example, when expecting a line, an attacker could send an unlimited 181(for example, when expecting a line, an attacker could send an unlimited
155amount of data without a callback ever being called as long as the line 182amount of data without a callback ever being called as long as the line
156isn't finished). 183isn't finished).
157 184
185=item autocork => <boolean>
186
187When disabled (the default), then C<push_write> will try to immediately
188write the data to the handle, if possible. This avoids having to register
189a write watcher and wait for the next event loop iteration, but can
190be inefficient if you write multiple small chunks (on the wire, this
191disadvantage is usually avoided by your kernel's nagle algorithm, see
192C<no_delay>, but this option can save costly syscalls).
193
194When enabled, then writes will always be queued till the next event loop
195iteration. This is efficient when you do many small writes per iteration,
196but less efficient when you do a single write only per iteration (or when
197the write buffer often is full). It also increases write latency.
198
199=item no_delay => <boolean>
200
201When doing small writes on sockets, your operating system kernel might
202wait a bit for more data before actually sending it out. This is called
203the Nagle algorithm, and usually it is beneficial.
204
205In some situations you want as low a delay as possible, which can be
206accomplishd by setting this option to a true value.
207
208The default is your opertaing system's default behaviour (most likely
209enabled), this option explicitly enables or disables it, if possible.
210
158=item read_size => <bytes> 211=item read_size => <bytes>
159 212
160The default read block size (the amount of bytes this module will try to read 213The default read block size (the amount of bytes this module will
161on each [loop iteration). Default: C<4096>. 214try to read during each loop iteration, which affects memory
215requirements). Default: C<8192>.
162 216
163=item low_water_mark => <bytes> 217=item low_water_mark => <bytes>
164 218
165Sets the amount of bytes (default: C<0>) that make up an "empty" write 219Sets the amount of bytes (default: C<0>) that make up an "empty" write
166buffer: If the write reaches this size or gets even samller it is 220buffer: If the write reaches this size or gets even samller it is
167considered empty. 221considered empty.
168 222
223Sometimes it can be beneficial (for performance reasons) to add data to
224the write buffer before it is fully drained, but this is a rare case, as
225the operating system kernel usually buffers data as well, so the default
226is good in almost all cases.
227
228=item linger => <seconds>
229
230If non-zero (default: C<3600>), then the destructor of the
231AnyEvent::Handle object will check whether there is still outstanding
232write data and will install a watcher that will write this data to the
233socket. No errors will be reported (this mostly matches how the operating
234system treats outstanding data at socket close time).
235
236This will not work for partial TLS data that could not be encoded
237yet. This data will be lost.
238
169=item tls => "accept" | "connect" | Net::SSLeay::SSL object 239=item tls => "accept" | "connect" | Net::SSLeay::SSL object
170 240
171When this parameter is given, it enables TLS (SSL) mode, that means it 241When this parameter is given, it enables TLS (SSL) mode, that means
172will start making tls handshake and will transparently encrypt/decrypt 242AnyEvent will start a TLS handshake as soon as the conenction has been
173data. 243established and will transparently encrypt/decrypt data afterwards.
174 244
175TLS mode requires Net::SSLeay to be installed (it will be loaded 245TLS mode requires Net::SSLeay to be installed (it will be loaded
176automatically when you try to create a TLS handle). 246automatically when you try to create a TLS handle): this module doesn't
247have a dependency on that module, so if your module requires it, you have
248to add the dependency yourself.
177 249
178For the TLS server side, use C<accept>, and for the TLS client side of a 250Unlike TCP, TLS has a server and client side: for the TLS server side, use
179connection, use C<connect> mode. 251C<accept>, and for the TLS client side of a connection, use C<connect>
252mode.
180 253
181You can also provide your own TLS connection object, but you have 254You can also provide your own TLS connection object, but you have
182to make sure that you call either C<Net::SSLeay::set_connect_state> 255to make sure that you call either C<Net::SSLeay::set_connect_state>
183or C<Net::SSLeay::set_accept_state> on it before you pass it to 256or C<Net::SSLeay::set_accept_state> on it before you pass it to
184AnyEvent::Handle. 257AnyEvent::Handle.
185 258
186See the C<starttls> method if you need to start TLs negotiation later. 259See the C<< ->starttls >> method for when need to start TLS negotiation later.
187 260
188=item tls_ctx => $ssl_ctx 261=item tls_ctx => $ssl_ctx
189 262
190Use the given Net::SSLeay::CTX object to create the new TLS connection 263Use the given C<Net::SSLeay::CTX> object to create the new TLS connection
191(unless a connection object was specified directly). If this parameter is 264(unless a connection object was specified directly). If this parameter is
192missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 265missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
193 266
194=item json => JSON or JSON::XS object 267=item json => JSON or JSON::XS object
195 268
196This is the json coder object used by the C<json> read and write types. 269This is the json coder object used by the C<json> read and write types.
197 270
198If you don't supply it, then AnyEvent::Handle will create and use a 271If you don't supply it, then AnyEvent::Handle will create and use a
199suitable one, which will write and expect UTF-8 encoded JSON texts. 272suitable one (on demand), which will write and expect UTF-8 encoded JSON
273texts.
200 274
201Note that you are responsible to depend on the JSON module if you want to 275Note that you are responsible to depend on the JSON module if you want to
202use this functionality, as AnyEvent does not have a dependency itself. 276use this functionality, as AnyEvent does not have a dependency itself.
203 277
204=item filter_r => $cb 278=item filter_r => $cb
205 279
206=item filter_w => $cb 280=item filter_w => $cb
207 281
208These exist, but are undocumented at this time. 282These exist, but are undocumented at this time. (They are used internally
283by the TLS code).
209 284
210=back 285=back
211 286
212=cut 287=cut
213 288
223 if ($self->{tls}) { 298 if ($self->{tls}) {
224 require Net::SSLeay; 299 require Net::SSLeay;
225 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 300 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
226 } 301 }
227 302
228# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
229# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
230# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
231 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
232
233 $self->{_activity} = time; 303 $self->{_activity} = AnyEvent->now;
234 $self->_timeout; 304 $self->_timeout;
235 305
306 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
307 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
308
236 $self->start_read; 309 $self->start_read
310 if $self->{on_read};
237 311
238 $self 312 $self
239} 313}
240 314
241sub _shutdown { 315sub _shutdown {
242 my ($self) = @_; 316 my ($self) = @_;
243 317
318 delete $self->{_tw};
244 delete $self->{_rw}; 319 delete $self->{_rw};
245 delete $self->{_ww}; 320 delete $self->{_ww};
246 delete $self->{fh}; 321 delete $self->{fh};
247}
248 322
323 $self->stoptls;
324
325 delete $self->{on_read};
326 delete $self->{_queue};
327}
328
249sub error { 329sub _error {
250 my ($self) = @_; 330 my ($self, $errno, $fatal) = @_;
251 331
252 {
253 local $!;
254 $self->_shutdown; 332 $self->_shutdown
255 } 333 if $fatal;
256 334
257 $self->{on_error}($self) 335 $! = $errno;
336
258 if $self->{on_error}; 337 if ($self->{on_error}) {
259 338 $self->{on_error}($self, $fatal);
339 } else {
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 340 Carp::croak "AnyEvent::Handle uncaught error: $!";
341 }
261} 342}
262 343
263=item $fh = $handle->fh 344=item $fh = $handle->fh
264 345
265This method returns the file handle of the L<AnyEvent::Handle> object. 346This method returns the file handle used to create the L<AnyEvent::Handle> object.
266 347
267=cut 348=cut
268 349
269sub fh { $_[0]{fh} } 350sub fh { $_[0]{fh} }
270 351
288 $_[0]{on_eof} = $_[1]; 369 $_[0]{on_eof} = $_[1];
289} 370}
290 371
291=item $handle->on_timeout ($cb) 372=item $handle->on_timeout ($cb)
292 373
293Replace the current C<on_timeout> callback, or disables the callback 374Replace the current C<on_timeout> callback, or disables the callback (but
294(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor 375not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
295argument. 376argument and method.
296 377
297=cut 378=cut
298 379
299sub on_timeout { 380sub on_timeout {
300 $_[0]{on_timeout} = $_[1]; 381 $_[0]{on_timeout} = $_[1];
382}
383
384=item $handle->autocork ($boolean)
385
386Enables or disables the current autocork behaviour (see C<autocork>
387constructor argument).
388
389=cut
390
391=item $handle->no_delay ($boolean)
392
393Enables or disables the C<no_delay> setting (see constructor argument of
394the same name for details).
395
396=cut
397
398sub no_delay {
399 $_[0]{no_delay} = $_[1];
400
401 eval {
402 local $SIG{__DIE__};
403 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
404 };
301} 405}
302 406
303############################################################################# 407#############################################################################
304 408
305=item $handle->timeout ($seconds) 409=item $handle->timeout ($seconds)
319# also check for time-outs 423# also check for time-outs
320sub _timeout { 424sub _timeout {
321 my ($self) = @_; 425 my ($self) = @_;
322 426
323 if ($self->{timeout}) { 427 if ($self->{timeout}) {
324 my $NOW = time; 428 my $NOW = AnyEvent->now;
325 429
326 # when would the timeout trigger? 430 # when would the timeout trigger?
327 my $after = $self->{_activity} + $self->{timeout} - $NOW; 431 my $after = $self->{_activity} + $self->{timeout} - $NOW;
328
329 warn "next to in $after\n";#d#
330 432
331 # now or in the past already? 433 # now or in the past already?
332 if ($after <= 0) { 434 if ($after <= 0) {
333 $self->{_activity} = $NOW; 435 $self->{_activity} = $NOW;
334 436
335 if ($self->{on_timeout}) { 437 if ($self->{on_timeout}) {
336 $self->{on_timeout}->($self); 438 $self->{on_timeout}($self);
337 } else { 439 } else {
338 $! = Errno::ETIMEDOUT; 440 $self->_error (&Errno::ETIMEDOUT);
339 $self->error;
340 } 441 }
341 442
342 # callbakx could have changed timeout value, optimise 443 # callback could have changed timeout value, optimise
343 return unless $self->{timeout}; 444 return unless $self->{timeout};
344 445
345 # calculate new after 446 # calculate new after
346 $after = $self->{timeout}; 447 $after = $self->{timeout};
347 } 448 }
348 449
349 Scalar::Util::weaken $self; 450 Scalar::Util::weaken $self;
451 return unless $self; # ->error could have destroyed $self
350 452
351 warn "after $after\n";#d#
352 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 453 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
353 delete $self->{_tw}; 454 delete $self->{_tw};
354 $self->_timeout; 455 $self->_timeout;
355 }); 456 });
356 } else { 457 } else {
410 my $len = syswrite $self->{fh}, $self->{wbuf}; 511 my $len = syswrite $self->{fh}, $self->{wbuf};
411 512
412 if ($len >= 0) { 513 if ($len >= 0) {
413 substr $self->{wbuf}, 0, $len, ""; 514 substr $self->{wbuf}, 0, $len, "";
414 515
415 $self->{_activity} = time; 516 $self->{_activity} = AnyEvent->now;
416 517
417 $self->{on_drain}($self) 518 $self->{on_drain}($self)
418 if $self->{low_water_mark} >= length $self->{wbuf} 519 if $self->{low_water_mark} >= length $self->{wbuf}
419 && $self->{on_drain}; 520 && $self->{on_drain};
420 521
421 delete $self->{_ww} unless length $self->{wbuf}; 522 delete $self->{_ww} unless length $self->{wbuf};
422 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 523 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
423 $self->error; 524 $self->_error ($!, 1);
424 } 525 }
425 }; 526 };
426 527
427 # try to write data immediately 528 # try to write data immediately
428 $cb->(); 529 $cb->() unless $self->{autocork};
429 530
430 # if still data left in wbuf, we need to poll 531 # if still data left in wbuf, we need to poll
431 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 532 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
432 if length $self->{wbuf}; 533 if length $self->{wbuf};
433 }; 534 };
448 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 549 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
449 ->($self, @_); 550 ->($self, @_);
450 } 551 }
451 552
452 if ($self->{filter_w}) { 553 if ($self->{filter_w}) {
453 $self->{filter_w}->($self, \$_[0]); 554 $self->{filter_w}($self, \$_[0]);
454 } else { 555 } else {
455 $self->{wbuf} .= $_[0]; 556 $self->{wbuf} .= $_[0];
456 $self->_drain_wbuf; 557 $self->_drain_wbuf;
457 } 558 }
458} 559}
459 560
460=item $handle->push_write (type => @args) 561=item $handle->push_write (type => @args)
461 562
462=item $handle->unshift_write (type => @args)
463
464Instead of formatting your data yourself, you can also let this module do 563Instead of formatting your data yourself, you can also let this module do
465the job by specifying a type and type-specific arguments. 564the job by specifying a type and type-specific arguments.
466 565
467Predefined types are (if you have ideas for additional types, feel free to 566Predefined types are (if you have ideas for additional types, feel free to
468drop by and tell us): 567drop by and tell us):
472=item netstring => $string 571=item netstring => $string
473 572
474Formats the given value as netstring 573Formats the given value as netstring
475(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 574(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
476 575
477=back
478
479=cut 576=cut
480 577
481register_write_type netstring => sub { 578register_write_type netstring => sub {
482 my ($self, $string) = @_; 579 my ($self, $string) = @_;
483 580
484 sprintf "%d:%s,", (length $string), $string 581 sprintf "%d:%s,", (length $string), $string
582};
583
584=item packstring => $format, $data
585
586An octet string prefixed with an encoded length. The encoding C<$format>
587uses the same format as a Perl C<pack> format, but must specify a single
588integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
589optional C<!>, C<< < >> or C<< > >> modifier).
590
591=cut
592
593register_write_type packstring => sub {
594 my ($self, $format, $string) = @_;
595
596 pack "$format/a*", $string
485}; 597};
486 598
487=item json => $array_or_hashref 599=item json => $array_or_hashref
488 600
489Encodes the given hash or array reference into a JSON object. Unless you 601Encodes the given hash or array reference into a JSON object. Unless you
523 635
524 $self->{json} ? $self->{json}->encode ($ref) 636 $self->{json} ? $self->{json}->encode ($ref)
525 : JSON::encode_json ($ref) 637 : JSON::encode_json ($ref)
526}; 638};
527 639
640=item storable => $reference
641
642Freezes the given reference using L<Storable> and writes it to the
643handle. Uses the C<nfreeze> format.
644
645=cut
646
647register_write_type storable => sub {
648 my ($self, $ref) = @_;
649
650 require Storable;
651
652 pack "w/a*", Storable::nfreeze ($ref)
653};
654
655=back
656
528=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 657=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
529 658
530This function (not method) lets you add your own types to C<push_write>. 659This function (not method) lets you add your own types to C<push_write>.
531Whenever the given C<type> is used, C<push_write> will invoke the code 660Whenever the given C<type> is used, C<push_write> will invoke the code
532reference with the handle object and the remaining arguments. 661reference with the handle object and the remaining arguments.
552ways, the "simple" way, using only C<on_read> and the "complex" way, using 681ways, the "simple" way, using only C<on_read> and the "complex" way, using
553a queue. 682a queue.
554 683
555In the simple case, you just install an C<on_read> callback and whenever 684In the simple case, you just install an C<on_read> callback and whenever
556new data arrives, it will be called. You can then remove some data (if 685new data arrives, it will be called. You can then remove some data (if
557enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 686enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
558or not. 687leave the data there if you want to accumulate more (e.g. when only a
688partial message has been received so far).
559 689
560In the more complex case, you want to queue multiple callbacks. In this 690In the more complex case, you want to queue multiple callbacks. In this
561case, AnyEvent::Handle will call the first queued callback each time new 691case, AnyEvent::Handle will call the first queued callback each time new
562data arrives and removes it when it has done its job (see C<push_read>, 692data arrives (also the first time it is queued) and removes it when it has
563below). 693done its job (see C<push_read>, below).
564 694
565This way you can, for example, push three line-reads, followed by reading 695This way you can, for example, push three line-reads, followed by reading
566a chunk of data, and AnyEvent::Handle will execute them in order. 696a chunk of data, and AnyEvent::Handle will execute them in order.
567 697
568Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 698Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
569the specified number of bytes which give an XML datagram. 699the specified number of bytes which give an XML datagram.
570 700
571 # in the default state, expect some header bytes 701 # in the default state, expect some header bytes
572 $handle->on_read (sub { 702 $handle->on_read (sub {
573 # some data is here, now queue the length-header-read (4 octets) 703 # some data is here, now queue the length-header-read (4 octets)
574 shift->unshift_read_chunk (4, sub { 704 shift->unshift_read (chunk => 4, sub {
575 # header arrived, decode 705 # header arrived, decode
576 my $len = unpack "N", $_[1]; 706 my $len = unpack "N", $_[1];
577 707
578 # now read the payload 708 # now read the payload
579 shift->unshift_read_chunk ($len, sub { 709 shift->unshift_read (chunk => $len, sub {
580 my $xml = $_[1]; 710 my $xml = $_[1];
581 # handle xml 711 # handle xml
582 }); 712 });
583 }); 713 });
584 }); 714 });
585 715
586Example 2: Implement a client for a protocol that replies either with 716Example 2: Implement a client for a protocol that replies either with "OK"
587"OK" and another line or "ERROR" for one request, and 64 bytes for the 717and another line or "ERROR" for the first request that is sent, and 64
588second request. Due tot he availability of a full queue, we can just 718bytes for the second request. Due to the availability of a queue, we can
589pipeline sending both requests and manipulate the queue as necessary in 719just pipeline sending both requests and manipulate the queue as necessary
590the callbacks: 720in the callbacks.
591 721
592 # request one 722When the first callback is called and sees an "OK" response, it will
723C<unshift> another line-read. This line-read will be queued I<before> the
72464-byte chunk callback.
725
726 # request one, returns either "OK + extra line" or "ERROR"
593 $handle->push_write ("request 1\015\012"); 727 $handle->push_write ("request 1\015\012");
594 728
595 # we expect "ERROR" or "OK" as response, so push a line read 729 # we expect "ERROR" or "OK" as response, so push a line read
596 $handle->push_read_line (sub { 730 $handle->push_read (line => sub {
597 # if we got an "OK", we have to _prepend_ another line, 731 # if we got an "OK", we have to _prepend_ another line,
598 # so it will be read before the second request reads its 64 bytes 732 # so it will be read before the second request reads its 64 bytes
599 # which are already in the queue when this callback is called 733 # which are already in the queue when this callback is called
600 # we don't do this in case we got an error 734 # we don't do this in case we got an error
601 if ($_[1] eq "OK") { 735 if ($_[1] eq "OK") {
602 $_[0]->unshift_read_line (sub { 736 $_[0]->unshift_read (line => sub {
603 my $response = $_[1]; 737 my $response = $_[1];
604 ... 738 ...
605 }); 739 });
606 } 740 }
607 }); 741 });
608 742
609 # request two 743 # request two, simply returns 64 octets
610 $handle->push_write ("request 2\015\012"); 744 $handle->push_write ("request 2\015\012");
611 745
612 # simply read 64 bytes, always 746 # simply read 64 bytes, always
613 $handle->push_read_chunk (64, sub { 747 $handle->push_read (chunk => 64, sub {
614 my $response = $_[1]; 748 my $response = $_[1];
615 ... 749 ...
616 }); 750 });
617 751
618=over 4 752=over 4
619 753
620=cut 754=cut
621 755
622sub _drain_rbuf { 756sub _drain_rbuf {
623 my ($self) = @_; 757 my ($self) = @_;
758
759 local $self->{_in_drain} = 1;
624 760
625 if ( 761 if (
626 defined $self->{rbuf_max} 762 defined $self->{rbuf_max}
627 && $self->{rbuf_max} < length $self->{rbuf} 763 && $self->{rbuf_max} < length $self->{rbuf}
628 ) { 764 ) {
629 $! = &Errno::ENOSPC; 765 $self->_error (&Errno::ENOSPC, 1), return;
630 $self->error;
631 } 766 }
632 767
633 return if $self->{in_drain}; 768 while () {
634 local $self->{in_drain} = 1;
635
636 while (my $len = length $self->{rbuf}) { 769 my $len = length $self->{rbuf};
637 no strict 'refs'; 770
638 if (my $cb = shift @{ $self->{_queue} }) { 771 if (my $cb = shift @{ $self->{_queue} }) {
639 unless ($cb->($self)) { 772 unless ($cb->($self)) {
640 if ($self->{_eof}) { 773 if ($self->{_eof}) {
641 # no progress can be made (not enough data and no data forthcoming) 774 # no progress can be made (not enough data and no data forthcoming)
642 $! = &Errno::EPIPE; 775 $self->_error (&Errno::EPIPE, 1), return;
643 $self->error;
644 } 776 }
645 777
646 unshift @{ $self->{_queue} }, $cb; 778 unshift @{ $self->{_queue} }, $cb;
647 return; 779 last;
648 } 780 }
649 } elsif ($self->{on_read}) { 781 } elsif ($self->{on_read}) {
782 last unless $len;
783
650 $self->{on_read}($self); 784 $self->{on_read}($self);
651 785
652 if ( 786 if (
653 $self->{_eof} # if no further data will arrive
654 && $len == length $self->{rbuf} # and no data has been consumed 787 $len == length $self->{rbuf} # if no data has been consumed
655 && !@{ $self->{_queue} } # and the queue is still empty 788 && !@{ $self->{_queue} } # and the queue is still empty
656 && $self->{on_read} # and we still want to read data 789 && $self->{on_read} # but we still have on_read
657 ) { 790 ) {
791 # no further data will arrive
658 # then no progress can be made 792 # so no progress can be made
659 $! = &Errno::EPIPE; 793 $self->_error (&Errno::EPIPE, 1), return
660 $self->error; 794 if $self->{_eof};
795
796 last; # more data might arrive
661 } 797 }
662 } else { 798 } else {
663 # read side becomes idle 799 # read side becomes idle
664 delete $self->{_rw}; 800 delete $self->{_rw};
665 return; 801 last;
666 } 802 }
667 } 803 }
668 804
669 if ($self->{_eof}) { 805 if ($self->{_eof}) {
670 $self->_shutdown; 806 if ($self->{on_eof}) {
671 $self->{on_eof}($self) 807 $self->{on_eof}($self)
672 if $self->{on_eof}; 808 } else {
809 $self->_error (0, 1);
810 }
811 }
812
813 # may need to restart read watcher
814 unless ($self->{_rw}) {
815 $self->start_read
816 if $self->{on_read} || @{ $self->{_queue} };
673 } 817 }
674} 818}
675 819
676=item $handle->on_read ($cb) 820=item $handle->on_read ($cb)
677 821
683 827
684sub on_read { 828sub on_read {
685 my ($self, $cb) = @_; 829 my ($self, $cb) = @_;
686 830
687 $self->{on_read} = $cb; 831 $self->{on_read} = $cb;
832 $self->_drain_rbuf if $cb && !$self->{_in_drain};
688} 833}
689 834
690=item $handle->rbuf 835=item $handle->rbuf
691 836
692Returns the read buffer (as a modifiable lvalue). 837Returns the read buffer (as a modifiable lvalue).
741 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 886 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
742 ->($self, $cb, @_); 887 ->($self, $cb, @_);
743 } 888 }
744 889
745 push @{ $self->{_queue} }, $cb; 890 push @{ $self->{_queue} }, $cb;
746 $self->_drain_rbuf; 891 $self->_drain_rbuf unless $self->{_in_drain};
747} 892}
748 893
749sub unshift_read { 894sub unshift_read {
750 my $self = shift; 895 my $self = shift;
751 my $cb = pop; 896 my $cb = pop;
757 ->($self, $cb, @_); 902 ->($self, $cb, @_);
758 } 903 }
759 904
760 905
761 unshift @{ $self->{_queue} }, $cb; 906 unshift @{ $self->{_queue} }, $cb;
762 $self->_drain_rbuf; 907 $self->_drain_rbuf unless $self->{_in_drain};
763} 908}
764 909
765=item $handle->push_read (type => @args, $cb) 910=item $handle->push_read (type => @args, $cb)
766 911
767=item $handle->unshift_read (type => @args, $cb) 912=item $handle->unshift_read (type => @args, $cb)
797 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 942 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
798 1 943 1
799 } 944 }
800}; 945};
801 946
802# compatibility with older API
803sub push_read_chunk {
804 $_[0]->push_read (chunk => $_[1], $_[2]);
805}
806
807sub unshift_read_chunk {
808 $_[0]->unshift_read (chunk => $_[1], $_[2]);
809}
810
811=item line => [$eol, ]$cb->($handle, $line, $eol) 947=item line => [$eol, ]$cb->($handle, $line, $eol)
812 948
813The callback will be called only once a full line (including the end of 949The callback will be called only once a full line (including the end of
814line marker, C<$eol>) has been read. This line (excluding the end of line 950line marker, C<$eol>) has been read. This line (excluding the end of line
815marker) will be passed to the callback as second argument (C<$line>), and 951marker) will be passed to the callback as second argument (C<$line>), and
830=cut 966=cut
831 967
832register_read_type line => sub { 968register_read_type line => sub {
833 my ($self, $cb, $eol) = @_; 969 my ($self, $cb, $eol) = @_;
834 970
835 $eol = qr|(\015?\012)| if @_ < 3; 971 if (@_ < 3) {
836 $eol = quotemeta $eol unless ref $eol; 972 # this is more than twice as fast as the generic code below
837 $eol = qr|^(.*?)($eol)|s;
838
839 sub { 973 sub {
840 $_[0]{rbuf} =~ s/$eol// or return; 974 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
841 975
842 $cb->($_[0], $1, $2); 976 $cb->($_[0], $1, $2);
843 1
844 }
845};
846
847# compatibility with older API
848sub push_read_line {
849 my $self = shift;
850 $self->push_read (line => @_);
851}
852
853sub unshift_read_line {
854 my $self = shift;
855 $self->unshift_read (line => @_);
856}
857
858=item netstring => $cb->($handle, $string)
859
860A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
861
862Throws an error with C<$!> set to EBADMSG on format violations.
863
864=cut
865
866register_read_type netstring => sub {
867 my ($self, $cb) = @_;
868
869 sub {
870 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
871 if ($_[0]{rbuf} =~ /[^0-9]/) {
872 $! = &Errno::EBADMSG;
873 $self->error;
874 } 977 1
875 return;
876 } 978 }
979 } else {
980 $eol = quotemeta $eol unless ref $eol;
981 $eol = qr|^(.*?)($eol)|s;
877 982
878 my $len = $1; 983 sub {
984 $_[0]{rbuf} =~ s/$eol// or return;
879 985
880 $self->unshift_read (chunk => $len, sub { 986 $cb->($_[0], $1, $2);
881 my $string = $_[1];
882 $_[0]->unshift_read (chunk => 1, sub {
883 if ($_[1] eq ",") {
884 $cb->($_[0], $string);
885 } else {
886 $! = &Errno::EBADMSG;
887 $self->error;
888 }
889 }); 987 1
890 }); 988 }
891
892 1
893 } 989 }
894}; 990};
895 991
896=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 992=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
897 993
949 return 1; 1045 return 1;
950 } 1046 }
951 1047
952 # reject 1048 # reject
953 if ($reject && $$rbuf =~ $reject) { 1049 if ($reject && $$rbuf =~ $reject) {
954 $! = &Errno::EBADMSG; 1050 $self->_error (&Errno::EBADMSG);
955 $self->error;
956 } 1051 }
957 1052
958 # skip 1053 # skip
959 if ($skip && $$rbuf =~ $skip) { 1054 if ($skip && $$rbuf =~ $skip) {
960 $data .= substr $$rbuf, 0, $+[0], ""; 1055 $data .= substr $$rbuf, 0, $+[0], "";
962 1057
963 () 1058 ()
964 } 1059 }
965}; 1060};
966 1061
1062=item netstring => $cb->($handle, $string)
1063
1064A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
1065
1066Throws an error with C<$!> set to EBADMSG on format violations.
1067
1068=cut
1069
1070register_read_type netstring => sub {
1071 my ($self, $cb) = @_;
1072
1073 sub {
1074 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1075 if ($_[0]{rbuf} =~ /[^0-9]/) {
1076 $self->_error (&Errno::EBADMSG);
1077 }
1078 return;
1079 }
1080
1081 my $len = $1;
1082
1083 $self->unshift_read (chunk => $len, sub {
1084 my $string = $_[1];
1085 $_[0]->unshift_read (chunk => 1, sub {
1086 if ($_[1] eq ",") {
1087 $cb->($_[0], $string);
1088 } else {
1089 $self->_error (&Errno::EBADMSG);
1090 }
1091 });
1092 });
1093
1094 1
1095 }
1096};
1097
1098=item packstring => $format, $cb->($handle, $string)
1099
1100An octet string prefixed with an encoded length. The encoding C<$format>
1101uses the same format as a Perl C<pack> format, but must specify a single
1102integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1103optional C<!>, C<< < >> or C<< > >> modifier).
1104
1105DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1106
1107Example: read a block of data prefixed by its length in BER-encoded
1108format (very efficient).
1109
1110 $handle->push_read (packstring => "w", sub {
1111 my ($handle, $data) = @_;
1112 });
1113
1114=cut
1115
1116register_read_type packstring => sub {
1117 my ($self, $cb, $format) = @_;
1118
1119 sub {
1120 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1121 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1122 or return;
1123
1124 $format = length pack $format, $len;
1125
1126 # bypass unshift if we already have the remaining chunk
1127 if ($format + $len <= length $_[0]{rbuf}) {
1128 my $data = substr $_[0]{rbuf}, $format, $len;
1129 substr $_[0]{rbuf}, 0, $format + $len, "";
1130 $cb->($_[0], $data);
1131 } else {
1132 # remove prefix
1133 substr $_[0]{rbuf}, 0, $format, "";
1134
1135 # read remaining chunk
1136 $_[0]->unshift_read (chunk => $len, $cb);
1137 }
1138
1139 1
1140 }
1141};
1142
967=item json => $cb->($handle, $hash_or_arrayref) 1143=item json => $cb->($handle, $hash_or_arrayref)
968 1144
969Reads a JSON object or array, decodes it and passes it to the callback. 1145Reads a JSON object or array, decodes it and passes it to the callback.
970 1146
971If a C<json> object was passed to the constructor, then that will be used 1147If a C<json> object was passed to the constructor, then that will be used
981the C<json> write type description, above, for an actual example. 1157the C<json> write type description, above, for an actual example.
982 1158
983=cut 1159=cut
984 1160
985register_read_type json => sub { 1161register_read_type json => sub {
986 my ($self, $cb, $accept, $reject, $skip) = @_; 1162 my ($self, $cb) = @_;
987 1163
988 require JSON; 1164 require JSON;
989 1165
990 my $data; 1166 my $data;
991 my $rbuf = \$self->{rbuf}; 1167 my $rbuf = \$self->{rbuf};
1006 () 1182 ()
1007 } 1183 }
1008 } 1184 }
1009}; 1185};
1010 1186
1187=item storable => $cb->($handle, $ref)
1188
1189Deserialises a L<Storable> frozen representation as written by the
1190C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1191data).
1192
1193Raises C<EBADMSG> error if the data could not be decoded.
1194
1195=cut
1196
1197register_read_type storable => sub {
1198 my ($self, $cb) = @_;
1199
1200 require Storable;
1201
1202 sub {
1203 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1204 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1205 or return;
1206
1207 my $format = length pack "w", $len;
1208
1209 # bypass unshift if we already have the remaining chunk
1210 if ($format + $len <= length $_[0]{rbuf}) {
1211 my $data = substr $_[0]{rbuf}, $format, $len;
1212 substr $_[0]{rbuf}, 0, $format + $len, "";
1213 $cb->($_[0], Storable::thaw ($data));
1214 } else {
1215 # remove prefix
1216 substr $_[0]{rbuf}, 0, $format, "";
1217
1218 # read remaining chunk
1219 $_[0]->unshift_read (chunk => $len, sub {
1220 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1221 $cb->($_[0], $ref);
1222 } else {
1223 $self->_error (&Errno::EBADMSG);
1224 }
1225 });
1226 }
1227
1228 1
1229 }
1230};
1231
1011=back 1232=back
1012 1233
1013=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1234=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1014 1235
1015This function (not method) lets you add your own types to C<push_read>. 1236This function (not method) lets you add your own types to C<push_read>.
1033=item $handle->stop_read 1254=item $handle->stop_read
1034 1255
1035=item $handle->start_read 1256=item $handle->start_read
1036 1257
1037In rare cases you actually do not want to read anything from the 1258In rare cases you actually do not want to read anything from the
1038socket. In this case you can call C<stop_read>. Neither C<on_read> no 1259socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1039any queued callbacks will be executed then. To start reading again, call 1260any queued callbacks will be executed then. To start reading again, call
1040C<start_read>. 1261C<start_read>.
1262
1263Note that AnyEvent::Handle will automatically C<start_read> for you when
1264you change the C<on_read> callback or push/unshift a read callback, and it
1265will automatically C<stop_read> for you when neither C<on_read> is set nor
1266there are any read requests in the queue.
1041 1267
1042=cut 1268=cut
1043 1269
1044sub stop_read { 1270sub stop_read {
1045 my ($self) = @_; 1271 my ($self) = @_;
1056 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1282 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1057 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1283 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
1058 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1284 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1059 1285
1060 if ($len > 0) { 1286 if ($len > 0) {
1061 $self->{_activity} = time; 1287 $self->{_activity} = AnyEvent->now;
1062 1288
1063 $self->{filter_r} 1289 $self->{filter_r}
1064 ? $self->{filter_r}->($self, $rbuf) 1290 ? $self->{filter_r}($self, $rbuf)
1065 : $self->_drain_rbuf; 1291 : $self->{_in_drain} || $self->_drain_rbuf;
1066 1292
1067 } elsif (defined $len) { 1293 } elsif (defined $len) {
1068 delete $self->{_rw}; 1294 delete $self->{_rw};
1069 delete $self->{_ww};
1070 delete $self->{_tw};
1071 $self->{_eof} = 1; 1295 $self->{_eof} = 1;
1072 $self->_drain_rbuf; 1296 $self->_drain_rbuf unless $self->{_in_drain};
1073 1297
1074 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1298 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1075 return $self->error; 1299 return $self->_error ($!, 1);
1076 } 1300 }
1077 }); 1301 });
1078 } 1302 }
1079} 1303}
1080 1304
1081sub _dotls { 1305sub _dotls {
1082 my ($self) = @_; 1306 my ($self) = @_;
1307
1308 my $buf;
1083 1309
1084 if (length $self->{_tls_wbuf}) { 1310 if (length $self->{_tls_wbuf}) {
1085 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1311 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1086 substr $self->{_tls_wbuf}, 0, $len, ""; 1312 substr $self->{_tls_wbuf}, 0, $len, "";
1087 } 1313 }
1088 } 1314 }
1089 1315
1090 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1316 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1091 $self->{wbuf} .= $buf; 1317 $self->{wbuf} .= $buf;
1092 $self->_drain_wbuf; 1318 $self->_drain_wbuf;
1093 } 1319 }
1094 1320
1095 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1321 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1322 if (length $buf) {
1096 $self->{rbuf} .= $buf; 1323 $self->{rbuf} .= $buf;
1097 $self->_drain_rbuf; 1324 $self->_drain_rbuf unless $self->{_in_drain};
1325 } else {
1326 # let's treat SSL-eof as we treat normal EOF
1327 $self->{_eof} = 1;
1328 $self->_shutdown;
1329 return;
1330 }
1098 } 1331 }
1099 1332
1100 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1333 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1101 1334
1102 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1335 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1103 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1336 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1104 $self->error; 1337 return $self->_error ($!, 1);
1105 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1338 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1106 $! = &Errno::EIO; 1339 return $self->_error (&Errno::EIO, 1);
1107 $self->error;
1108 } 1340 }
1109 1341
1110 # all others are fine for our purposes 1342 # all others are fine for our purposes
1111 } 1343 }
1112} 1344}
1127call and can be used or changed to your liking. Note that the handshake 1359call and can be used or changed to your liking. Note that the handshake
1128might have already started when this function returns. 1360might have already started when this function returns.
1129 1361
1130=cut 1362=cut
1131 1363
1132# TODO: maybe document...
1133sub starttls { 1364sub starttls {
1134 my ($self, $ssl, $ctx) = @_; 1365 my ($self, $ssl, $ctx) = @_;
1135 1366
1136 $self->stoptls; 1367 $self->stoptls;
1137 1368
1148 # basically, this is deep magic (because SSL_read should have the same issues) 1379 # basically, this is deep magic (because SSL_read should have the same issues)
1149 # but the openssl maintainers basically said: "trust us, it just works". 1380 # but the openssl maintainers basically said: "trust us, it just works".
1150 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 1381 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
1151 # and mismaintained ssleay-module doesn't even offer them). 1382 # and mismaintained ssleay-module doesn't even offer them).
1152 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1383 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1384 #
1385 # in short: this is a mess.
1386 #
1387 # note that we do not try to kepe the length constant between writes as we are required to do.
1388 # we assume that most (but not all) of this insanity only applies to non-blocking cases,
1389 # and we drive openssl fully in blocking mode here.
1153 Net::SSLeay::CTX_set_mode ($self->{tls}, 1390 Net::SSLeay::CTX_set_mode ($self->{tls},
1154 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1391 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1155 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1392 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1156 1393
1157 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1394 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1190 1427
1191sub DESTROY { 1428sub DESTROY {
1192 my $self = shift; 1429 my $self = shift;
1193 1430
1194 $self->stoptls; 1431 $self->stoptls;
1432
1433 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1434
1435 if ($linger && length $self->{wbuf}) {
1436 my $fh = delete $self->{fh};
1437 my $wbuf = delete $self->{wbuf};
1438
1439 my @linger;
1440
1441 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1442 my $len = syswrite $fh, $wbuf, length $wbuf;
1443
1444 if ($len > 0) {
1445 substr $wbuf, 0, $len, "";
1446 } else {
1447 @linger = (); # end
1448 }
1449 });
1450 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1451 @linger = ();
1452 });
1453 }
1195} 1454}
1196 1455
1197=item AnyEvent::Handle::TLS_CTX 1456=item AnyEvent::Handle::TLS_CTX
1198 1457
1199This function creates and returns the Net::SSLeay::CTX object used by 1458This function creates and returns the Net::SSLeay::CTX object used by
1241=over 4 1500=over 4
1242 1501
1243=item * all constructor arguments become object members. 1502=item * all constructor arguments become object members.
1244 1503
1245At least initially, when you pass a C<tls>-argument to the constructor it 1504At least initially, when you pass a C<tls>-argument to the constructor it
1246will end up in C<< $handle->{tls} >>. Those members might be changes or 1505will end up in C<< $handle->{tls} >>. Those members might be changed or
1247mutated later on (for example C<tls> will hold the TLS connection object). 1506mutated later on (for example C<tls> will hold the TLS connection object).
1248 1507
1249=item * other object member names are prefixed with an C<_>. 1508=item * other object member names are prefixed with an C<_>.
1250 1509
1251All object members not explicitly documented (internal use) are prefixed 1510All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines