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.95 by root, Thu Oct 2 06:42:39 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 {
293 295
294 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 296 $self->{fh} or Carp::croak "mandatory argument fh is missing";
295 297
296 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 298 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
297 299
298 if ($self->{tls}) {
299 require Net::SSLeay;
300 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 300 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
301 } 301 if $self->{tls};
302 302
303 $self->{_activity} = AnyEvent->now; 303 $self->{_activity} = AnyEvent->now;
304 $self->_timeout; 304 $self->_timeout;
305 305
306 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain}; 306 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
318 delete $self->{_tw}; 318 delete $self->{_tw};
319 delete $self->{_rw}; 319 delete $self->{_rw};
320 delete $self->{_ww}; 320 delete $self->{_ww};
321 delete $self->{fh}; 321 delete $self->{fh};
322 322
323 $self->stoptls; 323 &_freetls;
324 324
325 delete $self->{on_read}; 325 delete $self->{on_read};
326 delete $self->{_queue}; 326 delete $self->{_queue};
327} 327}
328 328
487 my ($self, $cb) = @_; 487 my ($self, $cb) = @_;
488 488
489 $self->{on_drain} = $cb; 489 $self->{on_drain} = $cb;
490 490
491 $cb->($self) 491 $cb->($self)
492 if $cb && $self->{low_water_mark} >= length $self->{wbuf}; 492 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
493} 493}
494 494
495=item $handle->push_write ($data) 495=item $handle->push_write ($data)
496 496
497Queues the given scalar to be written. You can push as much data as you 497Queues the given scalar to be written. You can push as much data as you
514 substr $self->{wbuf}, 0, $len, ""; 514 substr $self->{wbuf}, 0, $len, "";
515 515
516 $self->{_activity} = AnyEvent->now; 516 $self->{_activity} = AnyEvent->now;
517 517
518 $self->{on_drain}($self) 518 $self->{on_drain}($self)
519 if $self->{low_water_mark} >= length $self->{wbuf} 519 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
520 && $self->{on_drain}; 520 && $self->{on_drain};
521 521
522 delete $self->{_ww} unless length $self->{wbuf}; 522 delete $self->{_ww} unless length $self->{wbuf};
523 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 523 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
524 $self->_error ($!, 1); 524 $self->_error ($!, 1);
548 548
549 @_ = ($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")
550 ->($self, @_); 550 ->($self, @_);
551 } 551 }
552 552
553 if ($self->{filter_w}) { 553 if ($self->{tls}) {
554 $self->{filter_w}($self, \$_[0]); 554 $self->{_tls_wbuf} .= $_[0];
555 &_dotls ($self);
555 } else { 556 } else {
556 $self->{wbuf} .= $_[0]; 557 $self->{wbuf} .= $_[0];
557 $self->_drain_wbuf; 558 $self->_drain_wbuf;
558 } 559 }
559} 560}
795 796
796 last; # more data might arrive 797 last; # more data might arrive
797 } 798 }
798 } else { 799 } else {
799 # read side becomes idle 800 # read side becomes idle
800 delete $self->{_rw}; 801 delete $self->{_rw} unless $self->{tls};
801 last; 802 last;
802 } 803 }
803 } 804 }
804 805
805 if ($self->{_eof}) { 806 if ($self->{_eof}) {
1263Note that AnyEvent::Handle will automatically C<start_read> for you when 1264Note 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 1265you 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 1266will automatically C<stop_read> for you when neither C<on_read> is set nor
1266there are any read requests in the queue. 1267there are any read requests in the queue.
1267 1268
1269These methods will have no effect when in TLS mode (as TLS doesn't support
1270half-duplex connections).
1271
1268=cut 1272=cut
1269 1273
1270sub stop_read { 1274sub stop_read {
1271 my ($self) = @_; 1275 my ($self) = @_;
1272 1276
1273 delete $self->{_rw}; 1277 delete $self->{_rw} unless $self->{tls};
1274} 1278}
1275 1279
1276sub start_read { 1280sub start_read {
1277 my ($self) = @_; 1281 my ($self) = @_;
1278 1282
1279 unless ($self->{_rw} || $self->{_eof}) { 1283 unless ($self->{_rw} || $self->{_eof}) {
1280 Scalar::Util::weaken $self; 1284 Scalar::Util::weaken $self;
1281 1285
1282 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1286 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1283 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1287 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1284 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1288 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1285 1289
1286 if ($len > 0) { 1290 if ($len > 0) {
1287 $self->{_activity} = AnyEvent->now; 1291 $self->{_activity} = AnyEvent->now;
1288 1292
1289 $self->{filter_r} 1293 if ($self->{tls}) {
1290 ? $self->{filter_r}($self, $rbuf) 1294 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1291 : $self->{_in_drain} || $self->_drain_rbuf; 1295 &_dotls ($self);
1296 } else {
1297 $self->_drain_rbuf unless $self->{_in_drain};
1298 }
1292 1299
1293 } elsif (defined $len) { 1300 } elsif (defined $len) {
1294 delete $self->{_rw}; 1301 delete $self->{_rw};
1295 $self->{_eof} = 1; 1302 $self->{_eof} = 1;
1296 $self->_drain_rbuf unless $self->{_in_drain}; 1303 $self->_drain_rbuf unless $self->{_in_drain};
1311 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1318 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1312 substr $self->{_tls_wbuf}, 0, $len, ""; 1319 substr $self->{_tls_wbuf}, 0, $len, "";
1313 } 1320 }
1314 } 1321 }
1315 1322
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}))) { 1323 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1322 if (length $buf) { 1324 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 1325 # let's treat SSL-eof as we treat normal EOF
1326 delete $self->{_rw};
1327 $self->{_eof} = 1; 1327 $self->{_eof} = 1;
1328 $self->_shutdown; 1328 &_freetls;
1329 return;
1330 } 1329 }
1330
1331 $self->{rbuf} .= $buf;
1332 $self->_drain_rbuf unless $self->{_in_drain};
1333 $self->{tls} or return; # tls session might have gone away in callback
1331 } 1334 }
1332 1335
1333 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1336 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1334 1337
1335 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1338 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1339 return $self->_error (&Errno::EIO, 1); 1342 return $self->_error (&Errno::EIO, 1);
1340 } 1343 }
1341 1344
1342 # all others are fine for our purposes 1345 # all others are fine for our purposes
1343 } 1346 }
1347
1348 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1349 $self->{wbuf} .= $buf;
1350 $self->_drain_wbuf;
1351 }
1344} 1352}
1345 1353
1346=item $handle->starttls ($tls[, $tls_ctx]) 1354=item $handle->starttls ($tls[, $tls_ctx])
1347 1355
1348Instead of starting TLS negotiation immediately when the AnyEvent::Handle 1356Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1357 1365
1358The TLS connection object will end up in C<< $handle->{tls} >> after this 1366The 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 1367call and can be used or changed to your liking. Note that the handshake
1360might have already started when this function returns. 1368might have already started when this function returns.
1361 1369
1370If it an error to start a TLS handshake more than once per
1371AnyEvent::Handle object (this is due to bugs in OpenSSL).
1372
1362=cut 1373=cut
1363 1374
1364sub starttls { 1375sub starttls {
1365 my ($self, $ssl, $ctx) = @_; 1376 my ($self, $ssl, $ctx) = @_;
1366 1377
1367 $self->stoptls; 1378 require Net::SSLeay;
1368 1379
1380 Carp::croak "it is an error to call starttls more than once on an Anyevent::Handle object"
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};
1488 } 1511 }
1489} 1512}
1490 1513
1491=back 1514=back
1492 1515
1516
1517=head1 NONFREQUENTLY ASKED QUESTIONS
1518
1519=over 4
1520
1521=item How do I read data until the other side closes the connection?
1522
1523If you just want to read your data into a perl scalar, the easiest way to achieve this is
1524by setting an C<on_read> callback that does nothing, clearing the C<on_eof> callback
1525and in the C<on_error> callback, the data will be in C<$_[0]{rbuf}>:
1526
1527 $handle->on_read (sub { });
1528 $handle->on_eof (undef);
1529 $handle->on_error (sub {
1530 my $data = delete $_[0]{rbuf};
1531 undef $handle;
1532 });
1533
1534The reason to use C<on_error> is that TCP connections, due to latencies
1535and packets loss, might get closed quite violently with an error, when in
1536fact, all data has been received.
1537
1538It is usually better to use acknowledgements when transfering data,
1539to make sure the other side hasn't just died and you got the data
1540intact. This is also one reason why so many internet protocols have an
1541explicit QUIT command.
1542
1543
1544=item I don't want to destroy the handle too early - how do I wait until all data has been sent?
1545
1546After writing your last bits of data, set the C<on_drain> callback
1547and destroy the handle in there - with the default setting of
1548C<low_water_mark> this will be called precisely when all data has been
1549written to the socket:
1550
1551 $handle->push_write (...);
1552 $handle->on_drain (sub {
1553 warn "all data submitted to the kernel\n";
1554 undef $handle;
1555 });
1556
1557=back
1558
1559
1493=head1 SUBCLASSING AnyEvent::Handle 1560=head1 SUBCLASSING AnyEvent::Handle
1494 1561
1495In many cases, you might want to subclass AnyEvent::Handle. 1562In many cases, you might want to subclass AnyEvent::Handle.
1496 1563
1497To make this easier, a given version of AnyEvent::Handle uses these 1564To make this easier, a given version of AnyEvent::Handle uses these

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines