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.88 by root, Thu Aug 21 23:48:35 2008 UTC vs.
Revision 1.93 by root, Wed Oct 1 14:49:23 2008 UTC

… …
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
17=cut 17=cut
18 18
19our $VERSION = 4.233; 19our $VERSION = 4.3;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
… …
59treatment of characters applies to this module as well. 59treatment of characters applies to this module as well.
60 60
61All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
62argument. 62argument.
63 63
64=head2 SIGPIPE is not handled by this module
65
66SIGPIPE is not handled by this module, so one of the practical
67requirements of using it is to ignore SIGPIPE (C<$SIG{PIPE} =
68'IGNORE'>). At least, this is highly recommend in a networked program: If
69you use AnyEvent::Handle in a filter program (like sort), exiting on
70SIGPIPE is probably the right thing to do.
71
64=head1 METHODS 72=head1 METHODS
65 73
66=over 4 74=over 4
67 75
68=item B<new (%args)> 76=item B<new (%args)>
… …
232write data and will install a watcher that will write this data to the 240write 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 241socket. No errors will be reported (this mostly matches how the operating
234system treats outstanding data at socket close time). 242system treats outstanding data at socket close time).
235 243
236This will not work for partial TLS data that could not be encoded 244This will not work for partial TLS data that could not be encoded
237yet. This data will be lost. 245yet. This data will be lost. Calling the C<stoptls> method in time might
246help.
238 247
239=item tls => "accept" | "connect" | Net::SSLeay::SSL object 248=item tls => "accept" | "connect" | Net::SSLeay::SSL object
240 249
241When this parameter is given, it enables TLS (SSL) mode, that means 250When this parameter is given, it enables TLS (SSL) mode, that means
242AnyEvent will start a TLS handshake as soon as the conenction has been 251AnyEvent will start a TLS handshake as soon as the conenction has been
… …
273texts. 282texts.
274 283
275Note that you are responsible to depend on the JSON module if you want to 284Note that you are responsible to depend on the JSON module if you want to
276use this functionality, as AnyEvent does not have a dependency itself. 285use this functionality, as AnyEvent does not have a dependency itself.
277 286
278=item filter_r => $cb
279
280=item filter_w => $cb
281
282These exist, but are undocumented at this time. (They are used internally
283by the TLS code).
284
285=back 287=back
286 288
287=cut 289=cut
288 290
289sub new { 291sub new {
… …
318 delete $self->{_tw}; 320 delete $self->{_tw};
319 delete $self->{_rw}; 321 delete $self->{_rw};
320 delete $self->{_ww}; 322 delete $self->{_ww};
321 delete $self->{fh}; 323 delete $self->{fh};
322 324
323 $self->stoptls; 325 &_freetls;
324 326
325 delete $self->{on_read}; 327 delete $self->{on_read};
326 delete $self->{_queue}; 328 delete $self->{_queue};
327} 329}
328 330
… …
487 my ($self, $cb) = @_; 489 my ($self, $cb) = @_;
488 490
489 $self->{on_drain} = $cb; 491 $self->{on_drain} = $cb;
490 492
491 $cb->($self) 493 $cb->($self)
492 if $cb && $self->{low_water_mark} >= length $self->{wbuf}; 494 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
493} 495}
494 496
495=item $handle->push_write ($data) 497=item $handle->push_write ($data)
496 498
497Queues the given scalar to be written. You can push as much data as you 499Queues the given scalar to be written. You can push as much data as you
… …
514 substr $self->{wbuf}, 0, $len, ""; 516 substr $self->{wbuf}, 0, $len, "";
515 517
516 $self->{_activity} = AnyEvent->now; 518 $self->{_activity} = AnyEvent->now;
517 519
518 $self->{on_drain}($self) 520 $self->{on_drain}($self)
519 if $self->{low_water_mark} >= length $self->{wbuf} 521 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
520 && $self->{on_drain}; 522 && $self->{on_drain};
521 523
522 delete $self->{_ww} unless length $self->{wbuf}; 524 delete $self->{_ww} unless length $self->{wbuf};
523 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 525 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
524 $self->_error ($!, 1); 526 $self->_error ($!, 1);
… …
548 550
549 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 551 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
550 ->($self, @_); 552 ->($self, @_);
551 } 553 }
552 554
553 if ($self->{filter_w}) { 555 if ($self->{tls}) {
554 $self->{filter_w}($self, \$_[0]); 556 $self->{_tls_wbuf} .= $_[0];
557 &_dotls ($self);
555 } else { 558 } else {
556 $self->{wbuf} .= $_[0]; 559 $self->{wbuf} .= $_[0];
557 $self->_drain_wbuf; 560 $self->_drain_wbuf;
558 } 561 }
559} 562}
… …
795 798
796 last; # more data might arrive 799 last; # more data might arrive
797 } 800 }
798 } else { 801 } else {
799 # read side becomes idle 802 # read side becomes idle
800 delete $self->{_rw}; 803 delete $self->{_rw} unless $self->{tls};
801 last; 804 last;
802 } 805 }
803 } 806 }
804 807
805 if ($self->{_eof}) { 808 if ($self->{_eof}) {
… …
1263Note that AnyEvent::Handle will automatically C<start_read> for you when 1266Note 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 1267you 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 1268will automatically C<stop_read> for you when neither C<on_read> is set nor
1266there are any read requests in the queue. 1269there are any read requests in the queue.
1267 1270
1271These methods will have no effect when in TLS mode (as TLS doesn't support
1272half-duplex connections).
1273
1268=cut 1274=cut
1269 1275
1270sub stop_read { 1276sub stop_read {
1271 my ($self) = @_; 1277 my ($self) = @_;
1272 1278
1273 delete $self->{_rw}; 1279 delete $self->{_rw} unless $self->{tls};
1274} 1280}
1275 1281
1276sub start_read { 1282sub start_read {
1277 my ($self) = @_; 1283 my ($self) = @_;
1278 1284
1279 unless ($self->{_rw} || $self->{_eof}) { 1285 unless ($self->{_rw} || $self->{_eof}) {
1280 Scalar::Util::weaken $self; 1286 Scalar::Util::weaken $self;
1281 1287
1282 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1288 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1283 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1289 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1284 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1290 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1285 1291
1286 if ($len > 0) { 1292 if ($len > 0) {
1287 $self->{_activity} = AnyEvent->now; 1293 $self->{_activity} = AnyEvent->now;
1288 1294
1289 $self->{filter_r} 1295 if ($self->{tls}) {
1290 ? $self->{filter_r}($self, $rbuf) 1296 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1291 : $self->{_in_drain} || $self->_drain_rbuf; 1297 &_dotls ($self);
1298 } else {
1299 $self->_drain_rbuf unless $self->{_in_drain};
1300 }
1292 1301
1293 } elsif (defined $len) { 1302 } elsif (defined $len) {
1294 delete $self->{_rw}; 1303 delete $self->{_rw};
1295 $self->{_eof} = 1; 1304 $self->{_eof} = 1;
1296 $self->_drain_rbuf unless $self->{_in_drain}; 1305 $self->_drain_rbuf unless $self->{_in_drain};
… …
1311 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1320 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1312 substr $self->{_tls_wbuf}, 0, $len, ""; 1321 substr $self->{_tls_wbuf}, 0, $len, "";
1313 } 1322 }
1314 } 1323 }
1315 1324
1316 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1317 $self->{wbuf} .= $buf;
1318 $self->_drain_wbuf;
1319 }
1320
1321 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { 1325 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1322 if (length $buf) { 1326 unless (length $buf) {
1323 $self->{rbuf} .= $buf;
1324 $self->_drain_rbuf unless $self->{_in_drain};
1325 } else {
1326 # let's treat SSL-eof as we treat normal EOF 1327 # let's treat SSL-eof as we treat normal EOF
1328 delete $self->{_rw};
1327 $self->{_eof} = 1; 1329 $self->{_eof} = 1;
1328 $self->_shutdown; 1330 &_freetls;
1329 return;
1330 } 1331 }
1332
1333 $self->{rbuf} .= $buf;
1334 $self->_drain_rbuf unless $self->{_in_drain};
1335 $self->{tls} or return; # tls session might have gone away in callback
1331 } 1336 }
1332 1337
1333 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1338 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1334 1339
1335 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1340 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
… …
1339 return $self->_error (&Errno::EIO, 1); 1344 return $self->_error (&Errno::EIO, 1);
1340 } 1345 }
1341 1346
1342 # all others are fine for our purposes 1347 # all others are fine for our purposes
1343 } 1348 }
1349
1350 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1351 $self->{wbuf} .= $buf;
1352 $self->_drain_wbuf;
1353 }
1344} 1354}
1345 1355
1346=item $handle->starttls ($tls[, $tls_ctx]) 1356=item $handle->starttls ($tls[, $tls_ctx])
1347 1357
1348Instead of starting TLS negotiation immediately when the AnyEvent::Handle 1358Instead of starting TLS negotiation immediately when the AnyEvent::Handle
… …
1357 1367
1358The TLS connection object will end up in C<< $handle->{tls} >> after this 1368The TLS connection object will end up in C<< $handle->{tls} >> after this
1359call and can be used or changed to your liking. Note that the handshake 1369call and can be used or changed to your liking. Note that the handshake
1360might have already started when this function returns. 1370might have already started when this function returns.
1361 1371
1372If it an error to start a TLS handshake more than once per
1373AnyEvent::Handle object (this is due to bugs in OpenSSL).
1374
1362=cut 1375=cut
1363 1376
1364sub starttls { 1377sub starttls {
1365 my ($self, $ssl, $ctx) = @_; 1378 my ($self, $ssl, $ctx) = @_;
1366 1379
1367 $self->stoptls; 1380 Carp::croak "it is an error to call starttls more than once on an Anyevent::Handle object"
1368 1381 if $self->{tls};
1382
1369 if ($ssl eq "accept") { 1383 if ($ssl eq "accept") {
1370 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1384 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
1371 Net::SSLeay::set_accept_state ($ssl); 1385 Net::SSLeay::set_accept_state ($ssl);
1372 } elsif ($ssl eq "connect") { 1386 } elsif ($ssl eq "connect") {
1373 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1387 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
… …
1382 # and mismaintained ssleay-module doesn't even offer them). 1396 # and mismaintained ssleay-module doesn't even offer them).
1383 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1397 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1384 # 1398 #
1385 # in short: this is a mess. 1399 # in short: this is a mess.
1386 # 1400 #
1387 # note that we do not try to kepe the length constant between writes as we are required to do. 1401 # note that we do not try to keep 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, 1402 # 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. 1403 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
1404 # have identity issues in that area.
1390 Net::SSLeay::CTX_set_mode ($self->{tls}, 1405 Net::SSLeay::CTX_set_mode ($self->{tls},
1391 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1406 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1392 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1407 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1393 1408
1394 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1409 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1395 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1410 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1396 1411
1397 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio}); 1412 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1398 1413
1399 $self->{filter_w} = sub { 1414 &_dotls; # need to trigger the initial handshake
1400 $_[0]{_tls_wbuf} .= ${$_[1]}; 1415 $self->start_read; # make sure we actually do read
1401 &_dotls;
1402 };
1403 $self->{filter_r} = sub {
1404 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1405 &_dotls;
1406 };
1407} 1416}
1408 1417
1409=item $handle->stoptls 1418=item $handle->stoptls
1410 1419
1411Destroys the SSL connection, if any. Partial read or write data will be 1420Shuts down the SSL connection - this makes a proper EOF handshake by
1412lost. 1421sending a close notify to the other side, but since OpenSSL doesn't
1422support non-blocking shut downs, it is not possible to re-use the stream
1423afterwards.
1413 1424
1414=cut 1425=cut
1415 1426
1416sub stoptls { 1427sub stoptls {
1417 my ($self) = @_; 1428 my ($self) = @_;
1418 1429
1430 if ($self->{tls}) {
1431 Net::SSLeay::shutdown $self->{tls};
1432
1433 &_dotls;
1434
1435 # we don't give a shit. no, we do, but we can't. no...
1436 # we, we... have to use openssl :/
1437 &_freetls;
1438 }
1439}
1440
1441sub _freetls {
1442 my ($self) = @_;
1443
1444 return unless $self->{tls};
1445
1419 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 1446 Net::SSLeay::free (delete $self->{tls});
1420 1447
1421 delete $self->{_rbio}; 1448 delete @$self{qw(_rbio _wbio _tls_wbuf)};
1422 delete $self->{_wbio};
1423 delete $self->{_tls_wbuf};
1424 delete $self->{filter_r};
1425 delete $self->{filter_w};
1426} 1449}
1427 1450
1428sub DESTROY { 1451sub DESTROY {
1429 my $self = shift; 1452 my $self = shift;
1430 1453
1431 $self->stoptls; 1454 &_freetls;
1432 1455
1433 my $linger = exists $self->{linger} ? $self->{linger} : 3600; 1456 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1434 1457
1435 if ($linger && length $self->{wbuf}) { 1458 if ($linger && length $self->{wbuf}) {
1436 my $fh = delete $self->{fh}; 1459 my $fh = delete $self->{fh};

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines