File Coverage

blib/lib/Plack/App/WebSocket/Connection.pm

Criterion	Covered	Total	%
statement	18	69	26.0
branch	0	14	0.0
condition			n/a
subroutine	6	18	33.3
pod	3	4	75.0
total	27	105	25.7

package Plack::App::WebSocket::Connection;
use strict;
use warnings;
use Carp;
use Scalar::Util qw(weaken refaddr);
use Devel::GlobalDestruction ();
use AnyEvent;

our $VERSION = "0.08";

sub new {
    my ($class, $conn, $responder) = @_;
    my $self = bless {
        connection => $conn,
        responder => $responder,
        handlers => {
            message => [],
            finish  => [],
        },
    }, $class;
    $self->_setup_internal_event_handlers();
    return $self;
}

sub _setup_internal_event_handlers {
    my ($self) = @_;
    weaken $self;
    $self->{connection}->on(each_message => sub {
        return if !defined($self);
        my $strong_self = $self; ## make sure $self is alive during callback execution
        $_->($self, $_[1]->body, $self->_cancel_for('message',$_))
            foreach @{$self->{handlers}{message}};
    });
    $self->{connection}->on(finish => sub {
        return if !defined($self);
        my $strong_self = $self; ## make sure $self is alive during callback execution
        $_->($self) foreach @{$self->{handlers}{finish}};
    });
}

sub _clear_event_handlers {
    my ($self) = @_;
    foreach my $handler_list (values %{$self->{handlers}}) {
        @$handler_list = ();
    }
}

sub _cancel_for {
    my( $self, $event, $handler ) = @_;
    return sub {
        $self->{handlers}{$event} = [
            grep { refaddr($_) != refaddr($handler) }
                 @{ $self->{handlers}{$event} }
        ];
    };
}

sub on {
    my ($self, @handlers) = @_;

    my @cancel;

    while( my( $event, $handler ) = splice @handlers, 0, 2 ) {
        croak "handler for event $event must be a code-ref" if ref($handler) ne "CODE";
        $event = "finish" if $event eq "close";
        my $handler_list = $self->{handlers}{$event};
        croak "Unknown event: $event" if not defined $handler_list;
        push(@$handler_list, $handler);
        push @cancel, $self->_cancel_for($event,$handler);
    }

    return wantarray ? @cancel : $cancel[0];
}

sub send {
    my ($self, $message) = @_;
    $self->{connection}->send($message);
}

sub close {
    my ($self) = @_;
    $self->{connection}->close;
}

our $WAIT_FOR_FLUSHING_SEC = 5;

sub DESTROY {
    my ($self) = @_;
    return if Devel::GlobalDestruction::in_global_destruction;
    $self->_clear_event_handlers();
    my $connection = $self->{connection};
    $connection->close();  ## explicit close because $responder may keep the socket.
    my $responder = $self->{responder};
    my $w; $w = AnyEvent->timer(after => $WAIT_FOR_FLUSHING_SEC, cb => sub {
        $responder->([200, ["Content-Type", "text/plain"], ["WebSocket finished"]]);
        undef $w;
        undef $responder;

        ## Prolong $connection's life as long as $responder. This is
        ## necessary to make sure $connection actively shuts down the
        ## socket. If $connection is destroyed immediately and the
        ## kernel's write buffer is full, $connection may fail to shut
        ## down the socket (because $connection delays the active
        ## shutdown after sending all the buffered data). If that
        ## happens, the socket stays open, which is bad.
        undef $connection;
    });
}

1;

__END__