File Coverage

blib/lib/Plack/App/WebSocket.pm

Criterion	Covered	Total	%
statement	27	74	36.4
branch	0	22	0.0
condition	0	11	0.0
subroutine	9	14	64.2
pod	2	2	100.0
total	38	123	30.8

package Plack::App::WebSocket;
use strict;
use warnings;
use parent qw(Plack::Component);
use Carp;
use Plack::Response;
use AnyEvent::WebSocket::Server;
use Try::Tiny;
use Plack::App::WebSocket::Connection;
use Scalar::Util qw(blessed);

our $VERSION = "0.05";

my $ERROR_ENV = "plack.app.websocket.error";

sub new {
    my ($class, @args) = @_;
    my $self = $class->SUPER::new(@args);
    croak "on_establish param is mandatory" if not defined $self->{on_establish};
    croak "on_establish param must be a code-ref" if ref($self->{on_establish}) ne "CODE";
    $self->{on_error} ||= \&_default_on_error;
    croak "on_error param must be a code-ref" if ref($self->{on_error}) ne "CODE";

    if(!defined($self->{websocket_server})) {
        $self->{websocket_server} = AnyEvent::WebSocket::Server->new();
    }
    if(blessed($self->{websocket_server}) && !$self->{websocket_server}->isa("AnyEvent::WebSocket::Server")) {
        croak "websocket_server param must be a AnyEvent::WebSocket::Server";
    }

    return $self;
}

sub _default_on_error {
    my ($env) = @_;
    my $res = Plack::Response->new;
    $res->content_type("text/plain");
    if(!defined($env->{$ERROR_ENV})) {
        $res->status(500);
        $res->body("Unknown error");
    }elsif($env->{$ERROR_ENV} eq "not supported by the PSGI server") {
        $res->status(500);
        $res->body("The server does not support WebSocket.");
    }elsif($env->{$ERROR_ENV} eq "invalid request") {
        $res->status(400);
        $res->body("The request is invalid for a WebSocket request.");
    }else {
        $res->status(500);
        $res->body("Unknown error: $env->{$ERROR_ENV}");
    }
    $res->content_length(length($res->body));
    return $res->finalize;
}

sub _respond_via {
    my ($responder, $psgi_res) = @_;
    if(ref($psgi_res) eq "CODE") {
        $psgi_res->($responder);
    }else {
        $responder->($psgi_res);
    }
}

sub call {
    my ($self, $env) = @_;
    if(!$env->{"psgi.streaming"} || !$env->{"psgi.nonblocking"} || !$env->{"psgix.io"}) {
        $env->{$ERROR_ENV} = "not supported by the PSGI server";
        return $self->{on_error}->($env);
    }
    my $cv_conn = $self->{websocket_server}->establish_psgi($env, $env->{"psgix.io"});
    return sub {
        my $responder = shift;
        $cv_conn->cb(sub {
            my ($cv_conn) = @_;
            my ($conn, $error) = try {
                (scalar($cv_conn->recv), undef);
            }catch {
                (undef, $_[0]);
            };
            if(!$conn) {
                $env->{$ERROR_ENV} = "invalid request";
                $env->{"plack.app.websocket.error.handshake"} = $error;
                _respond_via($responder, $self->{on_error}->($env));
                return;
            }
            $self->{on_establish}->(Plack::App::WebSocket::Connection->new($conn, $responder), $env);
        });
    };
}

1;

__END__