File Coverage

blib/lib/IO/Handle/Record.pm

Criterion	Covered	Total	%
statement	130	184	70.6
branch	62	98	63.2
condition	33	53	62.2
subroutine	21	23	91.3
pod	2	5	40.0
total	248	363	68.3

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
      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;
      undef $I->read_buffer;
      croak "IO::Handle::Record: premature end of file";
    } elsif( $!==EAGAIN ) {
      return;
    } elsif( $!==EINTR ) {
      next;
    } else {
      undef $I->read_buffer;
      croak "IO::Handle::Record: sysread";
    }
  }

  if( $I->can('expect_fds') and
      $I->expect_fds>0 and defined $I->_received_fds ) {
    $I->received_fds=[splice @{$I->_received_fds}, 0, $I->expect_fds];
  }
  my $rc=eval {
    local $Storable::Eval;
    $I->record_opts and $Storable::Eval=$I->record_opts->{receive_CODE};
    Storable::thaw( $I->read_buffer );
  };
  if( $@ ) {
    my $e=$@;
    $e=~s/ at .*//s;
    croak $e;
  }

  undef $I->expected;
  undef $I->read_buffer;

  return @{$rc};
}

sub write_record {
  my $I=shift;

  my $writer=(issock($I)
	      ? sub { sendmsg( $_[0], $_[1], $_[2], (@_>3?$_[3]:0) ); }
	      : sub { syswrite $_[0], $_[1], $_[2], (@_>3?$_[3]:()); });

  my $can_fds_to_send=$I->can('fds_to_send');
  if( @_ ) {
    croak "IO::Handle::Record: busy"
      if( defined $I->write_buffer );
    my $L=($I->record_opts && $I->record_opts->{local_encoding}) ? 'L' : 'N';
    my $msg=eval {
      local $Storable::Deparse;
      local $Storable::forgive_me;
      $I->record_opts and do {
	$Storable::forgive_me=$I->record_opts->{forgive_me};
	$Storable::Deparse=$I->record_opts->{send_CODE};
      };
      local $SIG{__WARN__}=sub {};
      $L eq 'L'
	? Storable::freeze \@_
	: Storable::nfreeze \@_;
    };
    if( $@ ) {
      my $e=$@;
      $e=~s/ at .*//s;
      croak $e;
    }

    if( $can_fds_to_send ) {
      $I->write_buffer=pack( $L.'2', length($msg),
			     (defined $I->fds_to_send
			      ? 0+@{$I->fds_to_send}
			      : 0) ).$msg;
    } else {
      $I->write_buffer=pack( $L.'2', length($msg), 0 ).$msg;
    }
    $I->written=0;
  }

  my $written;

  # if there are file descriptors to send send them first along with the length
  # header only. (work around a bug in the suse 11.1 kernel)
  if( $I->written==0 and
      $can_fds_to_send and
      defined $I->fds_to_send and
      @{$I->fds_to_send} ) {
    while(!defined ($written=$writer->($I, $I->write_buffer, HEADERLENGTH))) {
      if( $!==EINTR ) {
	next;
      } elsif( $!==EAGAIN ) {
	return;
      } else {
	croak "IO::Handle::Record: syswrite";
      }
    }
    $I->written+=$written;
  }

  while( $I->writtenwrite_buffer) and
	 (defined ($written=$writer->($I, $I->write_buffer,
				      length($I->write_buffer)-$I->written,
				      $I->written)) or
	  $!==EINTR) ) {
    if( defined $written ) {
      $I->written+=$written;
    }
  }
  if( $I->written==length($I->write_buffer) ) {
    undef $I->write_buffer;
    undef $I->written;
    return 1;
  } elsif( $!==EAGAIN ) {
    return;
  } else {
    croak "IO::Handle::Record: syswrite";
  }
}

sub read_simple_record {
  my $I=shift;
  local $/;
  my $delim;
  if( $I->record_opts ) {
    $/=$I->record_opts->{record_delimiter} || "\n";
    $delim=$I->record_opts->{field_delimiter} || "\0";
  } else {
    $/="\n";
    $delim="\0";
  }

  my $r=<$I>;
  return unless( defined $r );	# EOF

  chomp $r;
  return split /\Q$delim\E/, $r;
}

sub write_simple_record {
  my $I=shift;
  my $rdelim;
  my $delim;
  if( $I->record_opts ) {
    $rdelim=$I->record_opts->{record_delimiter} || "\n";
    $delim=$I->record_opts->{field_delimiter} || "\0";
  } else {
    $rdelim="\n";
    $delim="\0";
  }

  print( $I join( $delim , @_ ), $rdelim );
  $I->flush;
}

*IO::Handle::write_record=\&write_record;
*IO::Handle::read_record=\&read_record;
*IO::Handle::end_of_input=\&end_of_input;
*IO::Handle::write_simple_record=\&write_simple_record;
*IO::Handle::read_simple_record=\&read_simple_record;
*IO::Handle::record_opts=\&record_opts;
*IO::Handle::expected=\&expected;
*IO::Socket::UNIX::expect_fds=\&expect_fds;
*IO::Handle::read_buffer=\&read_buffer;
*IO::Socket::UNIX::received_fds=\&received_fds;
*IO::Socket::UNIX::_received_fds=\&_received_fds;
*IO::Handle::written=\&written;
*IO::Handle::write_buffer=\&write_buffer;
*IO::Socket::UNIX::fds_to_send=\&fds_to_send;
*IO::Socket::UNIX::peercred=\&peercred;

1;
__END__