# QuizRPC - async RPC alike stuff

package QuizRPC;

use strict;
use IO::Socket;
use IO::Handle;
use Config;
use Fcntl qw/F_GETFL F_SETFL/;
use POSIX;
use Tk;
use Tk::IO;
use Storable qw(freeze thaw);
use Exporter;

use vars qw(@ISA @EXPORT_OK);
@ISA = qw(Exporter);
@EXPORT_OK = qw(packargs unpackargs);

use constant DEFAULTTIMEOUT => 4000;

my $NONBLOCK = eval "&$Config{o_nonblock}";

use vars qw($DEBUG);
$DEBUG = 0;

sub server {
    my $class = shift;
    my $widget = shift;
    my $port = shift;
    my $incoming_cb = shift;
    my $remotecmdtable = shift;

    $class = ref $class if ref $class;
    my $self = bless {}, $class;

    $self->{WIDGET} = $widget;
    $self->{LISTEN} = new IO::Socket::INET(
	qw/Listen 5 Reuse 1/, LocalPort => $port)
	or return;
    $self->{ID} = $self->{LISTEN}->sockhost() . ":" .
	$self->{LISTEN}->sockport();
    print STDERR "Listening on $self->{ID}\n" if $DEBUG;
    $self->{INCOMING_CB} = $incoming_cb;
    $self->{REMOTECMDTABLE} = $remotecmdtable;

    $widget->fileevent($self->{LISTEN}, 'readable',
	[ _new_connection => $self ] );
    $self;
}

sub getID {
    my $self = shift;

    $self->{ID};
}

sub connect {
    my $class = shift;
    my $widget = shift;
    my $addr = shift;
    my $remotecmdtable = shift;
    my $errcb = shift;

    $class = ref $class if ref $class;
    my $self = bless {}, $class;

    $self->{WIDGET} = $widget;
    $self->{SOCK} = new IO::Socket::INET(PeerAddr => $addr)
	or return;
    make_nonblocking($self->{SOCK});
    print STDERR "Connected to $addr\n" if $DEBUG;
    $self->{REMOTECMDTABLE} = $remotecmdtable;
    $self->{SERIAL} = 0;
    $self->{WRITEBUF} = '';
    $self->{READBUF} = '';
    $self->{ID} = $addr;
    $self->{CLIENTERRCB} = $errcb;
    $widget->fileevent($self->{SOCK}, 'readable', [ _comm_read => $self ]);
    $self;
}

sub make_nonblocking {
    my $sock = shift;

    return if $^O eq "MSWin32";
    my $fl = fcntl($sock, F_GETFL, 0);
    fcntl($sock, F_SETFL, 0 + ($fl | $NONBLOCK));
    $sock->autoflush(1);
}

sub _new_connection {
    my $server = shift;
    my $self = bless {}, ref $server;

    my($incoming, $addr) = $server->{LISTEN}->accept('Tk::IO');
    make_nonblocking($incoming);
    my($port, $iaddr) = sockaddr_in($addr);
    my $naddr = inet_ntoa($iaddr);
    my $id = "$naddr:$port";
    print STDERR "New connection from $id\n" if $DEBUG;
    $self->{SOCK} = $incoming;
    $self->{SERIAL} = 0;
    $self->{WRITEBUF} = '';
    $self->{READBUF} = '';
    $self->{ID} = $id;
    $self->{REMOTECMDTABLE} = $server->{REMOTECMDTABLE};
    my $widget = $self->{WIDGET} = $server->{WIDGET};
    $self->{CLIENTERRCB} = _call($server->{INCOMING_CB}, $self, $id);
    if ( ! $self->{CLIENTERRCB} ) {
	# connection rejected
	return $self->dead_comm("Connection rejected");
    }
    $widget->fileevent($incoming, 'readable', [ _comm_read => $self ]);
}

sub commcmd {
    my $self = shift;
    my $cmd = shift;

    my $serial = $self->{SERIAL}++ % 1000;
    my $widget = $self->{WIDGET};
    print STDERR "Sending cmd \"$cmd\" with serial $serial\n" if $DEBUG;
    my $timeout = $widget->after(DEFAULTTIMEOUT,
	[ _comm_timeout => $self ]);
    if ( $self->{TIMEOUT} ) {
	$self->{TIMEOUT}->cancel();
	print STDERR "Cancelled TIMEOUT for $self->{ID}\n" if $DEBUG;
    }
    $self->{TIMEOUT} = $timeout;
    print STDERR "Installed TIMEOUT for $self->{ID}\n" if $DEBUG;
    $self->{CALLBACK}{$serial} = shift;
    $self->comm_print(sprintf("C%03d %s\r\n", $serial, $cmd));
}

sub comm_print {
    my($self, $write) = @_;

    $self->{WRITEBUF} .= $write;
    if ( ! $self->{WRITECB} ) {
	my $widget = $self->{WIDGET};
	$self->{WRITECB} = 1;
	$widget->fileevent($self->{SOCK}, 'writable', [ _comm_write => $self ]);
	print STDERR "Installed writeCB for $self->{ID}\n" if $DEBUG;
    }
}

sub _comm_write {
    my $self = shift;

    if ( length $self->{WRITEBUF} ) {
	print STDERR "In comm_write for $self->{ID}. Writing...\n" if $DEBUG;
	my $n = syswrite($self->{SOCK}, $self->{WRITEBUF}, length $self->{WRITEBUF});
	print STDERR "Wrote $n bytes to $self->{ID}\n" if $DEBUG;
	substr($self->{WRITEBUF}, 0, $n) = "";
	return if length $self->{WRITEBUF};
    }
    else {
	print STDERR "Nothing to write for $self->{ID}\n" if $DEBUG;
    }
    if ( $self->{WRITECB} ) {
	my $widget = $self->{WIDGET};
	$widget->fileevent($self->{SOCK}, 'writable', '');
	$self->{WRITECB} = undef;
	print STDERR "Removed WriteCB for $self->{ID}\n" if $DEBUG;
    }
}

sub _comm_read {
    my $self = shift;

    if ( $self->{TIMEOUT} ) {
	print STDERR "Cancelled TIMEOUT for $self->{ID}\n" if $DEBUG;
	$self->{TIMEOUT}->cancel();
	$self->{TIMEOUT} = undef;
    }
    print STDERR "In _comm_read for $self->{ID}\n" if $DEBUG;
    my $n = sysread($self->{SOCK}, $self->{READBUF}, 65536,
	length $self->{READBUF});
    print STDERR "Read " . ($n || "no") . " bytes\n" if $DEBUG;
    if ( ! $n ) {
	my $err = $!;
	$err = "EOF" if !defined($n) || !$err;
	return $self->dead_comm($err);
    }
    while ( $self->{READBUF} =~ s/^(.*\S)\s*\n// ) {
	last unless $self->_comm_line($1);
    }
}

# XXX: timeouts are not optimal, we need a timeout per outstanding command
# in the ideal case.
sub _comm_timeout {
    my $self = shift;

    return unless $self->{TIMEOUT}; # already cancelled
    $self->dead_comm("Timeout");
}

sub _comm_line {
    my($self, $line) = @_;

    print STDERR "In comm_line for $self->{ID} line $line\n" if $DEBUG;
    unless ( $line =~ /^([CR])(\d{3}) ([A-Z]+)(?:\s(.*))?$/ ) {
	return $self->dead_comm("Invalid response");
    }
    my($cr, $serial, $cmd, $args) = ($1, 0+$2, $3, $4);
    if ( $cr eq "R" ) {
	exists $self->{CALLBACK}{$serial}
	    or return $self->dead_comm("Unbalanced R$serial");
	my $cb = $self->{CALLBACK}{$serial};
	delete $self->{CALLBACK}{$serial};
	_call($cb, $cmd, $args);
	return 1;
    }
    elsif ( exists $self->{REMOTECMDTABLE}{$cmd} ) {
	my $reply = _call([ $self->{REMOTECMDTABLE}{$cmd}, $self ], $args);
	$self->comm_print(sprintf("R%03d %s\r\n", $serial, $reply));
	return 1;
    }
    else {
	return $self->dead_comm("Invalid remote $cmd");
    }
}

sub _call {
    my $cb = shift;
    return unless defined $cb;
    my($cv, @args);
    if ( ref($cb) && ref($cb) eq "ARRAY" ) {
	$cv = shift @$cb;
	@args = @$cb;
    }
    else {
	$cv = $cb;
    }
    if ( ref($cv) && ref($cv) eq "CODE" ) {
	$cv->(@args, @_);
    }
    elsif ( ref($cv) && @args ) {
	my $meth = shift @args;
	$cv->$meth(@args, @_);
    }
    elsif ( @args && ref $args[0] ) {
	my $obj = shift @args;
	$obj->$cv(@args, @_);
    }
    else {
	die "Huh? $cv @args @_\n";
    }
}

sub dead_comm {
    my($self, $reason) = @_;

    my $widget = $self->{WIDGET};
    print STDERR "Error from $self->{ID}: $reason\n" if $DEBUG;
    if ( $self->{TIMEOUT} ) {
	print STDERR "Cancelled TIMEOUT for $self->{ID}\n" if $DEBUG;
	$self->{TIMEOUT}->cancel();
	$self->{TIMEOUT} = undef;
    }
    $widget->fileevent($self->{SOCK}, 'readable', '');
    $widget->fileevent($self->{SOCK}, 'writable', '') if $self->{WRITECB};
    shutdown($self->{SOCK}, 2);
    close $self->{SOCK};
    _call($self->{CLIENTERRCB}, $reason);
    0;
}

sub packargs {
    wantarray ? 
	map { unpack("H*", freeze($_)) } @_ :
	unpack("H*", freeze(shift));
}

sub unpackargs {
    wantarray ? 
	map { thaw(pack("H*", $_)) } @_ :
	thaw(pack("H*", shift));
}

1;
