File Coverage

blib/lib/MCE/Channel.pm

Criterion	Covered	Total	%
statement	65	71	91.5
branch	15	32	46.8
condition	10	21	47.6
subroutine	19	20	95.0
pod	2	2	100.0
total	111	146	76.0

###############################################################################
## ----------------------------------------------------------------------------
## Queue-like and two-way communication capability.
##
###############################################################################

package MCE::Channel;

use strict;
use warnings;

no warnings qw( uninitialized once );

our $VERSION = '1.889';

## no critic (BuiltinFunctions::ProhibitStringyEval)
## no critic (TestingAndDebugging::ProhibitNoStrict)

use if $^O eq 'MSWin32', 'threads';
use if $^O eq 'MSWin32', 'threads::shared';

use Carp ();

$Carp::Internal{ (__PACKAGE__) }++;

my ( $freeze, $thaw );

BEGIN {
   if ( $] ge '5.008008' && ! $INC{'PDL.pm'} ) {
      local $@;
      eval 'use Sereal::Encoder 3.015; use Sereal::Decoder 3.015;';
      if ( ! $@ ) {
         my $encoder_ver = int( Sereal::Encoder->VERSION() );
         my $decoder_ver = int( Sereal::Decoder->VERSION() );
         if ( $encoder_ver - $decoder_ver == 0 ) {
            $freeze = \&Sereal::Encoder::encode_sereal;
            $thaw   = \&Sereal::Decoder::decode_sereal;
         }
      }
   }

   if ( ! defined $freeze ) {
      require Storable;
      $freeze = \&Storable::freeze;
      $thaw   = \&Storable::thaw;
   }
}

use MCE::Util ();

my $tid = $INC{'threads.pm'} ? threads->tid() : 0;

sub new {
   my ( $class, %argv ) = @_;
   my $impl = defined( $argv{impl} ) ? ucfirst( lc $argv{impl} ) : 'Mutex';

   # Replace 'fast' with 'Fast' in the implementation value.
   $impl =~ s/fast/Fast/;

   $impl = 'Threads'     if ( $impl eq 'Mutex' && $^O eq 'MSWin32' );
   $impl = 'ThreadsFast' if ( $impl eq 'MutexFast' && $^O eq 'MSWin32' );
   $impl = 'Mutex'       if ( $impl eq 'Threads' && $^O eq 'cygwin' );
   $impl = 'MutexFast'   if ( $impl eq 'ThreadsFast' && $^O eq 'cygwin' );

   eval "require MCE::Channel::$impl; 1;" ||
      Carp::croak("Could not load Channel implementation '$impl': $@");

   my $pkg = 'MCE::Channel::'.$impl;
   no strict 'refs';

   $pkg->new(%argv);
}

sub CLONE {
   $tid = threads->tid if $INC{'threads.pm'};
}

sub DESTROY {
   my ( $pid, $self ) = ( $tid ? $$ .'.'. $tid : $$, @_ );

   if ( $self->{'init_pid'} && $self->{'init_pid'} eq $pid ) {
      MCE::Util::_destroy_socks($self, qw(c_sock c2_sock p_sock p2_sock));
      delete($self->{c_mutex}), delete($self->{p_mutex});
   }

   return;
}

sub impl {
   $_[0]->{'impl'} || 'Not defined';
}

sub _get_freeze { $freeze; }
sub _get_thaw   { $thaw;   }

sub _ended {
   warn "WARNING: ($_[0]) called on a channel that has been 'end'ed\n";

   return;
}

sub _read {
   my $bytes = MCE::Util::_sysread( $_[0], $_[1], my $len = $_[2] );
   my $read  = $bytes;

   while ( $bytes && $read != $len ) {
      $bytes = MCE::Util::_sysread( $_[0], $_[1], $len - $read, length($_[1]) );
      $read += $bytes if $bytes;
   }

   return;
}

sub _pid {
   $tid ? $$ .'.'. $tid : $$;
}

1;

__END__