view release on metacpan or  search on metacpan

lib/IO/Handle/Record.pm  view on Meta::CPAN

package IO::Handle::Record;

use 5.008008;
use strict;
use warnings;
use Storable;
use Class::Member::GLOB qw/record_opts
			   read_buffer expected expect_fds received_fds
			   end_of_input _received_fds
			   write_buffer fds_to_send written/;
use Errno qw/EAGAIN EINTR/;
use Carp;
my $have_inet6;
BEGIN {
  eval {
    require Socket6;
    $have_inet6=1;
  };
};
use Socket;
require XSLoader;

our $VERSION = '0.15';
XSLoader::load('IO::Handle::Record', $VERSION);

use constant {
  HEADERLENGTH=>8,		# 2 unsigned long
};

# this is called from the XS stuff in recvmsg
sub open_fd {
  my ($fd, $flags)=@_;
  use Fcntl qw/O_APPEND O_RDONLY O_WRONLY O_RDWR O_ACCMODE/;
  use POSIX ();
  use IO::Handle ();

  if( ($flags & O_ACCMODE) == O_RDONLY ) {
    $flags='<';
  } elsif( ($flags & O_ACCMODE) == O_WRONLY ) {
    if( $flags & O_APPEND ) {
      $flags='>>';
    } else {
      $flags='>';
    }
  } elsif( ($flags & O_ACCMODE) == O_RDWR ) {
    if( $flags & O_APPEND ) {
      $flags='+>>';
    } else {
      $flags='+>';
    }
  } else {
    POSIX::close($fd);
    return undef;
  }

  my $obj=bless IO::Handle->new_from_fd($fd, $flags),
                IO::Handle::Record::typeof($fd);

  if( ref($obj)=~/Socket/ ) {
    ${*$obj}{io_socket_domain}=socket_family($fd);
    ${*$obj}{io_socket_type}=socket_type($fd);

    if($obj->sockdomain==AF_INET or
       ($have_inet6 and $obj->sockdomain==&Socket6::AF_INET6) ) {
      if($obj->socktype==SOCK_STREAM) {
	${*$obj}{io_socket_proto}=&Socket::IPPROTO_TCP;
      } elsif($obj->socktype==SOCK_DGRAM) {
	${*$obj}{io_socket_proto}=&Socket::IPPROTO_UDP;
      } elsif($obj->socktype==SOCK_RAW) {
	${*$obj}{io_socket_proto}=&Socket::IPPROTO_ICMP;
      }
    }
  }

  return $obj;
}

sub read_record {
  my $I=shift;

  my $reader=(issock($I)
	      ? sub { recvmsg( $_[0], $_[1], $_[2], (@_>3?$_[3]:0) ); }
	      : sub { sysread $_[0], $_[1], $_[2], (@_>3?$_[3]:()); });

  unless( defined $I->expected ) {
    undef $I->end_of_input;
    undef $I->received_fds if( $I->can('received_fds') );
    $I->read_buffer='' unless( defined $I->read_buffer );
    my $buflen=length($I->read_buffer);
    while( $buflen<HEADERLENGTH ) {
      my $len=$reader->( $I, $I->read_buffer, HEADERLENGTH-$buflen, $buflen );
      if( defined($len) && $len==0 ) { # EOF
	undef $I->read_buffer;
	$I->end_of_input=1;
	return;
      } elsif( !defined($len) && $!==EAGAIN ) {
	return;			# non blocking file handle
      } elsif( !defined($len) && $!==EINTR ) {
	next;			# interrupted
      } elsif( !$len ) {	# ERROR
	$len=length $I->read_buffer;
	undef $I->read_buffer;
	croak "IO::Handle::Record: sysread";
      }
      $buflen+=$len;
    }
    my $L=($I->record_opts && $I->record_opts->{local_encoding}) ? 'L' : 'N';
    if( $I->can('expect_fds') ) {
      ($I->expected, $I->expect_fds)=unpack $L.'2', $I->read_buffer;
    } else {
      ($I->expected)=unpack $L.'2', $I->read_buffer;
    }
    $I->read_buffer='';
  }

  my $wanted=$I->expected;
  my $buflen=length($I->read_buffer);
  while( $buflen<$wanted ) {
    my $len=$reader->( $I, $I->read_buffer, $wanted-$buflen, $buflen );

    if( defined $len and $len>0 ) {
      $buflen+=$len;
    } elsif( defined $len ) {	# EOF
      $len=length $I->read_buffer;