File Coverage

blib/lib/Net/Server/NonBlocking.pm

Criterion	Covered	Total	%
statement	33	283	11.6
branch	0	132	0.0
condition	0	17	0.0
subroutine	11	34	32.3
pod	13	18	72.2
total	57	484	11.7

package Net::Server::NonBlocking;

use 5.000503;

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

$VERSION = '0.048';

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 = IO::Select->new();
my %idle;
my %timer;
my %map_all;
my %map_specific;
my %map_server;
my %map_client;
my %alive;

tie %ready, 'Tie::RefHash';

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

    my $self={};
    $self->{pidfile}=exists $hash->{pidfile} ? $hash->{pidfile} : '/tmp/anonymous_server';

    bless $self,$class;
}

sub add {
    my $self=shift;
    my $hash=shift;

    die("server_name is required") if not exists $hash->{server_name};

    if (not exists $hash->{local_port}) {
	$self->{listen}->{$hash->{server_name}}->{delimiter}=
	    exists $hash->{delimiter} ? $hash->{delimiter} : "\0";
	$self->{listen}->{$hash->{server_name}}->{string_format}=
	    exists $hash->{string_format} ? $hash->{string_format} : '.*?';
	$self->{listen}->{$hash->{server_name}}->{timeout}=
	    exists $hash->{timeout} ? $hash->{timeout} : 300;
	$self->{listen}->{$hash->{server_name}}->{on_disconnected}=
	    exists $hash->{on_disconnected} ? $hash->{on_disconnected} : sub {};
	$self->{listen}->{$hash->{server_name}}->{on_recv_msg}=
	    exists $hash->{on_recv_msg} ? $hash->{on_recv_msg} : sub {};
	$self->{listen}->{$hash->{server_name}}->{read_buffer} =
	    exists $hash->{read_buffer} ? $hash->{read_buffer} : \&read_buffer;
	$self->{listen}->{$hash->{server_name}}->{on_disconnected_param}=
	    exists $hash->{on_disconnected_param} ? $hash->{on_disconnected_param} : [];
	$self->{listen}->{$hash->{server_name}}->{on_recv_msg_param}=
	    exists $hash->{on_recv_msg_param} ? $hash->{on_recv_msg_param} : [];

	return undef;
    } else {
	my $server;

	if (exists $hash->{local_address}) {
	    $server = IO::Socket::INET->new(
					    LocalAddr => $hash->{local_address},
					    LocalPort => $hash->{local_port},
					    Listen    => 50,
					    Proto	=> 'tcp',
					    Reuse	=> 1,
					    Blocking => 0)
		or die "Can't make server socket -- $@\n";
	} else {
	    $server = IO::Socket::INET->new(
					    LocalPort => $hash->{local_port},
					    Listen    => 50,
					    Proto	=> 'tcp',
					    Reuse	=> 1,
					    Blocking => 0)
		or die "Can't make server socket -- $@\n";
	}
	$self->nonblock($server);

	$self->{listen}->{$hash->{server_name}}->{socket}=$server;
	$self->{listen}->{$hash->{server_name}}->{local_address}=$hash->{local_address} || "localhost";
	$self->{listen}->{$hash->{server_name}}->{local_port}=$hash->{local_port};
	$self->{listen}->{$hash->{server_name}}->{delimiter}=
	    exists $hash->{delimiter} ? $hash->{delimiter} : "\0";
	$self->{listen}->{$hash->{server_name}}->{string_format}=
	    exists $hash->{string_format} ? $hash->{string_format} : '.*?';
	$self->{listen}->{$hash->{server_name}}->{timeout}=
	    exists $hash->{timeout} ? $hash->{timeout} : 300;
	$self->{listen}->{$hash->{server_name}}->{on_connected}=
	    exists $hash->{on_connected} ? $hash->{on_connected} : sub {};
	$self->{listen}->{$hash->{server_name}}->{on_disconnected}=
	    exists $hash->{on_disconnected} ? $hash->{on_disconnected} : sub {};
	$self->{listen}->{$hash->{server_name}}->{on_recv_msg}=
	    exists $hash->{on_recv_msg} ? $hash->{on_recv_msg} : sub {};
	$self->{listen}->{$hash->{server_name}}->{read_buffer} =
	    exists $hash->{read_buffer} ? $hash->{read_buffer} : \&read_buffer;
	$self->{listen}->{$hash->{server_name}}->{on_connected_param}=
	    exists $hash->{on_connected_param} ? $hash->{on_connected_param} : [];
	$self->{listen}->{$hash->{server_name}}->{on_disconnected_param}=
	    exists $hash->{on_disconnected_param} ? $hash->{on_disconnected_param} : [];
	$self->{listen}->{$hash->{server_name}}->{on_recv_msg_param}=
	    exists $hash->{on_recv_msg_param} ? $hash->{on_recv_msg_param} : [];

	if (exists $hash->{local_address}) {
	    $map_specific{"$hash->{local_address}:$hash->{local_port}"}=
		$hash->{server_name};
	} else {
	    $map_all{$hash->{local_port}}=
		$hash->{server_name};
	}

	$map_server{$server} = $hash->{server_name};

	return $server;
    }
}

sub bind {
    my $self=shift;
    my $server_name=shift;
    my $client=shift;

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

    $alive{$client}=1;
    $idle{$client}=time;
    $turn_timeout{$client}=-1;

    $map_client{$client}=$server_name;
}

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 $server_name=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->{listen}->{$server_name}->{on_recv_msg}->($self,$client,$request,@{$self->{listen}->{$server_name}->{on_recv_msg_param}});
    }

    delete $ready{$client};
}

sub get_server_socket {
    my $self=shift;
    my $server_name = shift;

    $self->{listen}->{$server_name}->{socket};
}

sub get_server_name {
    my $self=shift;
    my $client=shift;
    #my @caller=caller();

    return $map_server{$client} if exists $map_server{$client};
    return $map_client{$client} if exists $map_client{$client};

    if (exists $map_specific{$client->sockhost().":".$client->sockport()}) {
        return $map_specific{$client->sockhost().":".$client->sockport()};
    } else {
        return $map_all{$client->sockport()};
    }
}

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

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

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

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

#sub flush_input {
#    my $self=shift;
#    my $client=shift;
#    my $server_name=$self->get_server_name($client);
#
#    my $rin='';
#    vec($rin,fileno($client),1)=1;
#
#    select(my $rout=$rin,undef,undef,0);
#
#    if (vec($rout,fileno($client),1)) {
#	my $data = '';
#	my $rv = $client->recv($data, POSIX::BUFSIZ, 0);
#
#	unless (defined($rv) && length $data) {
#	    # This would be the end of file, so close the client
#	    $self->erase_client($server_name,$client);
#	    next;
#	}
#
#	$inbuffer{$client} .= $data;
#	$self->{listen}->{$server_name}->{read_buffer}->($self,\$inbuffer{$client},\$ready{$client},$server_name);
#
#	$idle{$client}=time;
#    }
#
#    $self->handle($server_name,$client);
#}

sub flush_output {
    my $self=shift;
    my $client=shift;
    my $server_name=$self->get_server_name($client);

    return unless length $outbuffer{$client};

    my $rin='';
    vec($rin,fileno($client),1)=1;

    select(undef,my $rout=$rin,undef,0);

    if (vec($rout,fileno($client),1)) {
	while ($outbuffer{$client}) {
	    my $rv;

	    eval{
		$rv = $client->send($outbuffer{$client}, 0);
	    };
	    return if $@;   #the $client is disconnected

	    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

		substr($outbuffer{$client}, 0,$rv,'') if defined $rv;
		delete $outbuffer{$client} unless length $outbuffer{$client};
	    }
	}
    }
}

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

    #print "Idle delete close_client $client\n";

    delete $alive{$client};
    delete $turn_timeout{$client};
    delete $turn_timeout_trigger{$client};
    delete $idle{$client};
    delete $inbuffer{$client};
    delete $outbuffer{$client};
    delete $ready{$client};
    delete $map_client{$client} if exists $map_client{$client};

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

sub erase_client {
    my $self=shift;
    my $server_name=shift;
    my $client=shift;

    delete $alive{$client};
    delete $turn_timeout{$client};
    delete $turn_timeout_trigger{$client};
    delete $idle{$client};
    delete $inbuffer{$client};
    delete $outbuffer{$client};
    delete $ready{$client};
    delete $map_client{$client} if exists $map_client{$client};

    $self->{listen}->{$server_name}->{on_disconnected}->($self,$client,@{$self->{listen}->{$server_name}->{on_disconnected_param}});

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

sub enqueue {
    my $self=shift;
    my $client=shift;
    my $data=shift;

    return unless $client and $data;

    $outbuffer{$client}.=$data;
}

sub read_buffer {
    my $self=shift;
    my $raw_input=shift;
    my $cooked_input=shift;
    my $server_name=shift;

    my $dm=$self->{listen}->{$server_name}->{delimiter};
    my $sf=$self->{listen}->{$server_name}->{string_format};

    while ($$raw_input =~ s/($sf)$dm//s) {
	push( @{$$cooked_input}, $1 );
    }
}

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

    foreach (keys %{$self->{listen}}) {
	next unless $self->{listen}->{$_}->{local_port};
	warn "Listen on ".$self->{listen}->{$_}->{local_address}.":".
	    $self->{listen}->{$_}->{local_port}."\n";
	$select->add($self->{listen}->{$_}->{socket});
    }

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

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

	# cron
	my $this_time=time;
	if ($current_time != $this_time) {
	    foreach $client($select->handles) {
		next if exists $map_server{$client};
		next unless exists $alive{$client};

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

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

	#timeout the Idles

	foreach $client ($select->handles) {
	    next if exists $map_server{$client};
	    next unless exists $alive{$client};

	    my $server_name=$self->get_server_name($client);

	    my $this_time=time;
	    if( $this_time - $idle{$client} >= $self->{listen}->{$server_name}->{timeout} ){
		$self->erase_client($server_name,$client);
		next;
	    }
	}
	# check for new information on the connections we have

	# anything to read or accept?
	foreach $client ($select->can_read(1)) {
	    if (exists $map_server{$client}) {
		my $server_name=$self->get_server_name($client);
		# accept a new connection
		$client = $self->{listen}->{$server_name}->{socket}->accept();
		unless ($client) {
		    warn "Accepting new socket error: $!\n";
		    next;
		}

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

		$alive{$client}=1;
		$self->{listen}->{$server_name}->{on_connected}->($self,$client,@{$self->{listen}->{$server_name}->{on_connected_param}});
		$idle{$client}=time;
		$turn_timeout{$client}=-1;
	    } else {
		next unless exists $alive{$client};
		my $server_name=$self->get_server_name($client);
		# 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->erase_client($server_name,$client);
		    next;
		}

		$inbuffer{$client} .= $data;
		$self->{listen}->{$server_name}->{read_buffer}->($self,\$inbuffer{$client},\$ready{$client},$server_name);

		$idle{$client}=time;
	    }
	}

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

	my @bad_client;

	# Buffers to flush?
	foreach $client ($select->can_write(1)) {
	    next unless exists $alive{$client};

	    my $server_name=$self->get_server_name($client);

	    # 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

		substr($outbuffer{$client}, 0,$rv,'') if defined $rv;
		delete $outbuffer{$client} unless length $outbuffer{$client};
		next;
	    }
	}

	foreach $client (@bad_client){
	    my $server_name=$self->get_server_name($client);
	    $self->erase_client($server_name,$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 onSheddo{
    my $self=shift;

    foreach (sort {$a <=> $b} keys %timer) {
	unless ($cronCounter % $_) {
	    my $count=@{$timer{$_}};
	    &{$timer{$_}->[0]}($self,@{$timer{$_}}[1..($count-1)]);
	}
    }

    ++$cronCounter;
}

sub cron {
    my $self=shift;
    my $sec=shift;
    my $sub=shift;

    $timer{$sec}=[$sub,@_];
}

sub select {
    my $self=shift;

    $select;
}

1;

__END__