File Coverage

blib/lib/Festival/Client/Async.pm

Criterion	Covered	Total	%
statement	27	180	15.0
branch	2	58	3.4
condition	3	22	13.6
subroutine	9	27	33.3
pod	2	19	10.5
total	43	306	14.0

# Festival::Client::Async:  Non-blocking interface to a Festival server
#
# Copyright (c) 2000 Cepstral LLC. All rights Reserved.
#
# This module is free software; you can redistribute it and/or modify
# it under the same terms as Perl itself.
#
# Written by David Huggins-Daines

package Festival::Client::Async;
use strict;
use IO::Socket;
use Fcntl;

BEGIN {
    unless (defined &DEBUG) {
	*DEBUG = sub () { 0 }
    }
};

use vars qw($VERSION @ISA @EXPORT_OK);
$VERSION = 0.03_03;
@ISA = qw(Exporter);
@EXPORT_OK = qw(parse_lisp);

sub parse_lisp {
    my $lisp = shift;

    my (@stack, $top);
    $top = [];
    @stack = ($top);
    while ($lisp =~ m{(
		       [()]
		      |
		       "(?:[^"\\]+|\\.)*"
		      |
		       \#<[^>]+>
		      |
		       [^()\s]+
		       )}xg) {
	my $tok = $1;
	if ($tok eq '(') {
	    my $newtop = [];
	    push @$top, $newtop;
	    push @stack, ($top = $newtop);
	} elsif ($tok eq ')') {
	    pop @stack;
	    $top = $stack[-1];
	    die "stack underflow" unless defined $top;
	} else {
	    push @$top, $tok;
	}
    }

    return $top->[0];
}

sub new {
    my $this = shift;
    my $class = ref $this || $this;

    my ($host, $port) = @_;

    my $s = IO::Socket::INET->new(Proto     => 'tcp',
				  PeerAddr  => $host || 'localhost',
				  PeerPort  => $port || 1314)
	or return undef;
    binmode $s;

    my $self = bless {
		      blocked => 0,
		      sock => $s,
		      outbuf => "",
		      outq => {
			       LP => [],
			      },
		      intag => "",
		      inbuf => "",
		      inq => {
			      LP => [],
			      WV => [],
			      OK => [],
			      ER => [],
			     },
		     }, $class;
    $self->unblock;
    return $self;
}

sub fh {
    my $self = shift;
    return $self->{sock};
}

sub block {
    my $self = shift;
    my $flags = 0;
    fcntl $self->{sock}, F_GETFL, $flags
	or die "fcntl(F_GETFL) failed: $!";
    fcntl $self->{sock}, F_SETFL, $flags & ~O_NONBLOCK
	or die "fcntl(F_SETFL) failed: $!";
    $self->{blocked} = 1;
}

sub unblock {
    my $self = shift;
    my $flags = 0;
    fcntl $self->{sock}, F_GETFL, $flags
	or die "fcntl(F_GETFL) failed: $!";
    fcntl $self->{sock}, F_SETFL, $flags | O_NONBLOCK
	or die "fcntl(F_SETFL) failed: $!";
    $self->{blocked} = 0;
}

# Protocol encoding
use constant KEY     => "ft_StUfF_key";
use constant KEYLEN  => length KEY;

sub write_more {
    my $self = shift;

    while (defined(my $expr = shift @{$self->{outq}{LP}})) {
	$self->{outbuf} .= $expr;
    }

    my $count;
    while (defined(my $b = syswrite($self->{sock}, $self->{outbuf}, 4096))) {
	print "wrote $b bytes\n" if DEBUG;
	last if $b == 0;

	$count += $b;
	substr($self->{outbuf}, 0, $b) = "";
	last if $self->{blocked} and $b < 4096;
    }

    return $count;
}

sub read_more {
    my $self = shift;
    my $fh = $self->{sock};

    my $count = 0;
    my $burf = sysread $fh, my($rbuf), 4096;
    print "read $burf bytes\n" if DEBUG;
    $self->{inbuf} .= $rbuf;

 CHUNK:
    while (length($self->{inbuf}) > 0) {
	# In the middle of a tag?
	if ($self->{intag}) {
	    # Look for the stuff key
	    if ((my $i = index($self->{inbuf}, KEY)) != $[-1) {
		if (substr($self->{inbuf}, $i+KEYLEN, 1) eq 'X') {
		    # If there's an X at the end, it's literal
		    substr($self->{inbuf}, $i+KEYLEN, 1) = "";
		} else {
		    # Otherwise, we've got a complete waveform/expr/whatever
		    push @{$self->{inq}{$self->{intag}}},
			substr($self->{inbuf}, 0, $i);
		    print "queued $i bytes of $self->{intag}\n" if DEBUG;
		    substr($self->{inbuf}, 0, $i+KEYLEN) = "";
		    $self->{intag} = "";
		    $count += $i;
		}
	    } else {
		# Maybe we got *part* of the stuff key at the end of
		# this block.  Stranger things have happened.
		my $leftover = "";
	    PARTIAL:
		for my $sub (1..KEYLEN-1) {
		    my $foo = \substr($self->{inbuf}, -$sub);
		    my $bar = substr(KEY, 0, $sub);
		    if ($$foo eq $bar) {
			$$foo = "";
			$leftover = $bar;
			last PARTIAL;
		    }
		}

		# In any case we don't have any more data
		push @{$self->{inq}{$self->{intag}}}, $self->{inbuf};
		print "queued ", length($self->{inbuf}), " bytes of $self->{intag}\n"
		    if DEBUG;
		$count += length($self->{inbuf});
		$self->{inbuf} = $leftover;

		# But don't keep looping if we left some stuff in there!
		last CHUNK if $leftover;
	    }
	} else {
	    if ($self->{inbuf} =~ s/^(WV|LP|ER|OK)\n//) {
		print "got tag $1\n" if DEBUG;
		$count += length($1);
		# We got a tag, so a new type of data is coming
		if ($1 eq 'OK') {
		    push @{$self->{inq}{OK}}, time;
		} elsif ($1 eq 'ER') {
		    push @{$self->{inq}{ER}}, time;
		} else {
		    $self->{intag} = $1;
		}
	    } else {
		# Should not actually be fatal, it's always possible
		# we just got the middle of a tag.
		last CHUNK;
	    }
	}
    }

    return $count;
}

sub server_eval_sync {
    my ($self, $lisp, $actions) = @_;
    $self->block;
    $self->server_eval($lisp);

    unless ($self->write_more) {
	$self->unblock;
	return undef;
    }
    while ($self->read_more) {
	while (defined(my $wav = $self->dequeue_wave)) {
	    $actions->{WV}->($wav) if exists $actions->{WV};
	}
	while (defined(my $lisp = $self->dequeue_lisp)) {
	    $actions->{LP}->($lisp) if exists $actions->{LP};
	}
	if (defined($self->dequeue_error)) {
	    $self->unblock;
	    return undef;
	}
	if (defined($self->dequeue_ok)) {
	    last;
	}
    }
    $self->unblock;
    return 1;
}

# Don't mix this with async operations :(
sub server_eval_sync_old {
    my ($self, $lisp, $actions) = @_;
    my $fh = $self->{sock};
    $self->block;

    local $|=1;

    my ($rbuf, $rest, $tag);
    print $fh $lisp;
    while (defined($rbuf = $rest) or defined($rbuf = <$fh>)) {
	undef $rest;
	if ($rbuf =~ s/^(WV|LP|ER|OK)\n$//s) {
	    $tag = $1;
	    last if $tag eq 'OK' or $tag eq 'ER';
	}

	if ((my $i = index($rbuf, KEY)) != $[-1) {
	    if (substr($rbuf, $i+KEYLEN, 1) eq 'X') {
		substr($rbuf, $i+KEYLEN, 1) = "";
	    } else {
		$rest = substr($rbuf, $i+KEYLEN);
		substr($rbuf, $i) = "";
	    }
	}

	if (defined $tag and exists $actions->{$tag}) {
	    $actions->{$tag}->($rbuf);
	}
    }

    $self->unblock;
    return defined($tag) && ($tag eq 'OK');
}

sub server_eval {
    my $self = shift;
    push @{$self->{outq}{LP}}, @_;
}

sub write_pending {
    my $self = shift;
    return @{$self->{outq}{LP}};
}

sub wave_pending {
    my $self = shift;
    return @{$self->{inq}{WV}};
}

sub lisp_pending {
    my $self = shift;
    return @{$self->{inq}{LP}};
}

sub ok_pending {
    my $self = shift;
    return @{$self->{inq}{OK}};
}

sub error_pending {
    my $self = shift;
    return @{$self->{inq}{ER}};
}

sub dequeue_wave {
    my $self = shift;
    shift @{$self->{inq}{WV}};
}

sub dequeue_lisp {
    my $self = shift;
    shift @{$self->{inq}{LP}};
}

sub dequeue_ok {
    my $self = shift;
    shift @{$self->{inq}{OK}};
}

sub dequeue_error {
    my $self = shift;
    shift @{$self->{inq}{ER}};
}

1;
__END__