File Coverage

blib/lib/IO/NonBlocking.pm

Criterion	Covered	Total	%
statement	27	196	13.7
branch	0	52	0.0
condition	0	13	0.0
subroutine	9	31	29.0
pod	19	22	86.3
total	55	314	17.5

package IO::NonBlocking;

use strict;
use warnings;
use POSIX;
use IO::Socket;
use IO::Select;
use Socket;
use Fcntl;
use Tie::RefHash;
use vars qw($VERSION);

$VERSION = '1.035';

my @now=localtime(time);
my $cronCounter=$now[0]+60*$now[1]+3600*$now[2]+3600*24*$now[3];
my %buff;

# begin with empty buffers
my %inbuffer  = ();
my %outbuffer = ();
my %ready = ();

my %turn_timeout;
my %turn_timeout_trigger;
my $select;
my %idle;
my %timer;

tie %ready, 'Tie::RefHash';

my $daily=0;

sub new{
	my($proto,@arg)=@_;
	my $class=ref($proto) || $proto;
	my %hash=%{$arg[0]};

	my $self={};

	$self->{serverName}=$hash{'server_name'} || die("server_name is required");
	$self->{port}=$hash{'port'} || die("port is required");
	$self->{delimiter}=defined($hash{delimiter}) ? $hash{delimiter} : "\0";
	$self->{string_format}=defined($hash{string_format}) ? $hash{string_format} : '.*?';
	$self->{timeout}=defined($hash{timeout}) ? $hash{timeout} : 300;
	$self->{piddir}=defined($hash{piddir}) ? $hash{piddir} : '/tmp';

	bless $self,$class;
}

sub start_turn{
	my $self=shift;
	my $client=shift;
	my $time=shift;

	$turn_timeout{$client}=$time;
	$turn_timeout_trigger{$client}=$_[0];
}

sub stop_time {
	my $self=shift;
	my $client=shift;

	$turn_timeout{$client}=-1;
	delete($turn_timeout_trigger{$client});
}

sub close_client {
	my $self=shift;
	my $client=shift;

	delete $turn_timeout{$client};
	delete $turn_timeout_trigger{$client};
	delete $idle{$client};
	delete $inbuffer{$client};
	delete $outbuffer{$client};
	delete $ready{$client};

	$select->remove($client);
	close $client if $client;
}

sub disconnect_client {
	my $self=shift;
	my $client=shift;

	delete $turn_timeout{$client};
	delete $turn_timeout_trigger{$client};
	delete $idle{$client};
	delete $inbuffer{$client};
	delete $outbuffer{$client};
	delete $ready{$client};

	$self->onClientDisconnected($client);

	$select->remove($client);
	close $client if $client;
}

sub start{
	my $self=shift;
	my $current_time=time;

	print "Listening on port ".$self->port."\n";

	my $server = IO::Socket::INET->new(
					LocalPort => $self->port,
					Listen    => 50,
					Proto	=> 'tcp',
					Reuse	=> 1)
				  or die "Can't make server socket: $@\n";

	$self->nonblock($server);
	$select = IO::Select->new($server);

	open(FILE,">".$self->piddir."/".$self->serverName) or die "Cannot write PID file: $!\n";
	print FILE $$;
	close(FILE);

	$self->onServerInit;

	while (1) {
		my $client;
		my $rv;
    		my $data;

		# cron
		my $this_time=time;
		if ($current_time != $this_time) {
		  foreach $client($select->handles) {
		    next if $server == $client;

		    if ($turn_timeout{$client} != -1) {
		      if ($turn_timeout{$client} <= 0) {
			&{$turn_timeout_trigger{$client}}($self);
			delete $turn_timeout_trigger{$client};
			$turn_timeout{$client} = -1;
		      } else {
			--$turn_timeout{$client};
		      }
		    }
		  }

		  $self->onSheddo;
		  $current_time=$this_time;
		}

		#timeout the Idles

		if($cronCounter % ($self->{timeout}+30) == 0){
			foreach $client ($select->handles) {
				next if $server == $client;

				my $this_time=time;

				if( $this_time - $idle{$client} >= $self->{timeout} ){
					$self->disconnect_client($client);
					next;
				}
			}
		}

		# check for new information on the connections we have

		# anything to read or accept?
    		foreach $client ($select->can_read(1)) {

 			if ($client == $server) {
            			# accept a new connection

				$client = $server->accept();
				$select->add($client);
				$self->nonblock($client);

				$self->onClientConnected($client);
				$idle{$client}=time;
				$turn_timeout{$client}=-1;
			} else {
            			# read data

				$data = '';
				$rv   = $client->recv($data, POSIX::BUFSIZ, 0);

				unless (defined($rv) && length $data) {
                			# This would be the end of file, so close the client
					$self->disconnect_client($client);
					next;
				}

				$inbuffer{$client} .= $data;

				# test whether the data in the buffer or the data we
				# just read means there is a complete request waiting
				# to be fulfilled.  If there is, set $ready{$client}
				# to the requests waiting to be fulfilled.
				my $dm=$self->{delimiter};
				my $sf=$self->{string_format};

				while ($inbuffer{$client} =~ s/($sf)$dm//s) {
					push( @{$ready{$client}}, $1 );
				}

				$idle{$client}=time;
			}
		}

		# Any complete requests to process?
		foreach $client (keys %ready) {
			$self->handle($client);
		}

		my @bad_client;

		# Buffers to flush?
		foreach $client ($select->can_write(1)) {
			# Skip this client if we have nothing to say
			next unless exists $outbuffer{$client};

			eval{
				$rv = $client->send($outbuffer{$client}, 0);
			};
			push(@bad_client,$client),next if $@;

			unless (defined $rv) {
            			# Whine, but move on.

				warn "I was told I could write, but I can't.\n";
				next;
			}

			if ( $rv == length $outbuffer{$client} || $! == POSIX::EWOULDBLOCK) {
				substr($outbuffer{$client}, 0, $rv) = '';
				delete $outbuffer{$client} unless length $outbuffer{$client};
			} else {
				# Couldn't write all the data, and it wasn't because
				# it would have blocked.  Shutdown and move on.

				$self->disconnect_client($client);
				next;
			}
		}

		foreach (@bad_client){
			$self->disconnect_client($client);
		}

		# Out of band data?
		foreach $client ($select->has_exception(0)) {
			# arg is timeout
        			# Deal with out-of-band data here, if you want to.
		}
	}

}

sub nonblock {
	my $self=shift;
	my $socket=shift;
	my $flags;

	$flags = fcntl($socket, F_GETFL, 0)
		or die "Can't get flags for socket: $!\n";
	fcntl($socket, F_SETFL, $flags | O_NONBLOCK)
		or die "Can't make socket nonblocking: $!\n";
}

sub handle {
	my $self=shift;
	my $client = shift;
	my $request;

	# requests are in $ready{$client}
	# send output to $outbuffer{$client}

	foreach $request (@{$ready{$client}}) {
        		# $request is the text of the request
        		# put text of reply into $outbuffer{$client}

		$self->onReceiveMessage($client,$request);
	}

	delete $ready{$client};
}

#============= facility functions ============

sub getip {
        my $self=shift;
	$_[0]->sockhost();
}

sub getport {
        my $self=shift;
	$_[0]->sockport();
}

sub piddir{
	my $self=shift;

	return $self->{piddir};
}

sub serverName{
	my $self=shift;

	return $self->{serverName};
}

sub port{
	my $self=shift;

	return $self->{port};
}

sub sendmsg{
        my $self=shift;
        my $client=shift;
        my $msg=shift;

        $outbuffer{$client}.=$msg.$self->{delimiter};
}

sub onServerInit {
}

sub onClientConnected{
	my $self=shift;
	my $client=shift;
}

sub onClientDisconnected{
	my $self=shift;
	my $client=shift;
}

sub onReceiveMessage{
	my $self=shift;
	my $client=shift;
	my $request=shift;
}

sub onSheddo{
	my $self=shift;

	foreach (sort {$a <=> $b} keys %timer) {
	  unless ($cronCounter % $_) {
	    &{$timer{$_}}($self);
	  }
	}

	if ($cronCounter % 4527 == 0) {
	  #Sync time

	  my @now=localtime(time);
	  $cronCounter=$now[0]+60*$now[1]+3600*$now[2]+3600*24*$now[3];
	  return;
	}

	++$cronCounter;
}

sub cron {
        my $self=shift;

	$timer{$_[0]}=$_[1];
}

sub add_socket {
	my $self=shift;
	my $sock=shift;

	$select->add($sock);
	$self->nonblock($sock);

	$idle{$sock}=time;
	$turn_timeout{$sock}=-1;
}

sub select {
	my $self=shift;

	$select;
}

1;
__END__