view release on metacpan or  search on metacpan

NonBlock.pm  view on Meta::CPAN

	
	if (!defined($Res))
		{
		&{$EOF}($Nest, $SRec, 'recv() fatal error');
		return;
		};

	if (!length($Buf))
		{
		&{$EOF}($Nest, $SRec, 'EOF');
		return;
		};

	$SRec->{'Input'}->[0]->{'Data'} .= $Buf;

	$SRec->{'ATime'}    = $ATime;
	$SRec->{'BytesIn'} += length($Buf);

	return length($Buf);
	};

my $RecvUDP = sub($$$)
	{
	my ($Nest, $SRec, $ATime) = @_;

	($SRec->{'Socket'} && $Nest->{'Pool'}{$SRec})
		or  &{$Die}("$SRec: bad socket");

	my $BufAvail = $SRec->{'BuffSize'} - &{$BuffSize}($SRec, 'Input');
	my $Received = 0;

	my $Sel = IO::Select->new($SRec->{'Socket'});
	while($Sel->can_read(0) && ($BufAvail > $Received))
		{
		my $Buf = '';
		my $Res = $SRec->{'Socket'}->recv($Buf, $SRec->{'BuffSize'});
		
		if (!defined($Res))
			{
			&{$EOF}($Nest, $SRec, 'recv() fatal error');
			return;
			}
		
	        (length($Buf) || !$SRec->{'DiscEmpty'})
	        	or next;

		$Received += (length($Buf) + 20);
		my $tmpHash = {'Data' => $Buf};
		&{$UpdatePeer}($tmpHash, $SRec->{'Socket'});
		push(@{$SRec->{'Input'}}, $tmpHash);
		};

	$Received
		and $SRec->{'ATime'} = $ATime;

	$SRec->{'BytesIn'} += $Received;

	return $Received;
	};

sub IO($$)
	{
	my ($Nest, $ErrArray) = @_;

	my $Result = '0 but true';

	$ErrArray and @{$ErrArray} = ();

	$Nest->{'ErrArray'} = $ErrArray;

	my $CurTime = time();

	my $SRec = undef;

	foreach $SRec (values(%{$Nest->{'Pool'}}))
		{ &{$Cleanup}($Nest, $SRec); };

	my $Socket = undef;

	my @SockArray = $Nest->{'Select'}->can_read($Nest->{'SelectT'});
	foreach $Socket (@SockArray)
		{
		$SRec  = $Nest->{'S2Rec'}{$Socket};
	
		if ($SRec->{'EOF'} || $SRec->{'Close'} ||
		    (&{$BuffSize}($SRec, 'Input') >= $SRec->{'BuffSize'}))
			{ next; };
	
		if ($SRec->{'Accept'} && $SRec->{'TCP'})
			{
			$Result++;
			&{$Accept}($Nest, $SRec)
				and &{$ThrowMsg}(undef, $Nest->{'debug'}, "$SRec: incoming connection accepted")
				or  &{$ThrowMsg}($Nest, ($^W || $Nest->{'debug'}), "$SRec: Can not accept incoming connection: $@");
			$SRec->{'ATime'} = $CurTime;
			next;
			};
	
	        
		my ($Res) = &{$SRec->{'TCP'} ? $RecvTCP : $RecvUDP}($Nest, $SRec, $CurTime)
			or next;
		
		&{$ThrowMsg}(undef, $Nest->{'debug'}, "$SRec: recv $Res bytes");
	  	
	  	$Result++;
		};

	my $Continue = 1;
	while ($Continue)
		{
		$Continue = 0;
		my $Socket = undef;

		@SockArray = $Nest->{'Select'}->can_write($Nest->{'SelectT'});
		foreach $Socket (@SockArray)
			{
			$SRec  = $Nest->{'S2Rec'}{$Socket};

			my $OutRec  = $SRec->{'Output'}->[0];

			(defined($OutRec) && !$SRec->{'EOF'})