Dogcows Code - chaz/p5-Plack-App-Proxy-WebSocket/blob - lib/Plack/App/Proxy/WebSocket.pm

   1 package Plack::App::Proxy::WebSocket;
   2 # ABSTRACT: proxy HTTP and WebSocket connections
   3
   4 use warnings FATAL => 'all';
   5 use strict;
   6
   7 use AnyEvent::Handle;
   8 use AnyEvent::Socket;
   9 use HTTP::Request;
  10 use HTTP::Response;
  11 use Plack::Request;
  12 use URI;
  13
  14 use parent 'Plack::App::Proxy';
  15
  16 =head1 SYNOPSIS
  17
  18     use Plack::App::Proxy::WebSocket;
  19     use Plack::Builder;
  20
  21     builder {
  22         mount "/socket.io" => Plack::App::Proxy::WebSocket->new(
  23             remote               => "http://localhost:9000/socket.io",
  24             preserve_host_header => 1,
  25         )->to_app;
  26     };
  27
  28 =head1 DESCRIPTION
  29
  30 This is a subclass of L<Plack::App::Proxy> that adds support for proxying
  31 WebSocket connections.  It has no extra dependencies or configuration options
  32 beyond what L<Plack::App::Proxy> requires or provides, so it may be an easy
  33 drop-in replacement.  Read the documentation of that module for advanced usage
  34 not covered by the L<SYNOPSIS>.
  35
  36 This subclass necessarily requires extra L<PSGI> server features in order to
  37 work.  The server must support C<psgi.streaming> and C<psgix.io>.  It is also
  38 highly recommended to choose a C<psgi.nonblocking> server, though that isn't
  39 strictly required; performance may suffer greatly without it.  L<Twiggy> is an
  40 excellent choice for this application.
  41
  42 This module is B<EXPERIMENTAL>.  I use it in development and it works
  43 swimmingly for me, but it is completely untested in production scenarios.
  44
  45 =head1 CAVEATS
  46
  47 Some servers (e.g. L<Starman>) ignore the C<Connection> HTTP response header
  48 and use their own values, but WebSocket clients expect the value of that
  49 header to be C<Upgrade>.  This module cannot work on such servers.  Your best
  50 bet is to use a non-blocking server like L<Twiggy> that doesn't mess with the
  51 C<Connection> header.
  52
  53 =cut
  54
  55 sub call {
  56     my ($self, $env) = @_;
  57     my $req = Plack::Request->new($env);
  58
  59     # detect the websocket handshake or just proxy as usual
  60     lc($req->header('Upgrade') || "") eq 'websocket' or return $self->SUPER::call($env);
  61
  62     $env->{'psgi.streaming'} or die "Plack server support for psgi.streaming is required";
  63     my $client_fh = $env->{'psgix.io'} or die "Plack server support for the psgix.io extension is required";
  64
  65     my $url = $self->build_url_from_env($env) or return [502, [], ["Bad Gateway"]];
  66     my $uri = URI->new($url);
  67
  68     sub {
  69         my $res = shift;
  70
  71         # set up an event loop if the server is blocking
  72         my $cv;
  73         unless ($env->{'psgi.nonblocking'}) {
  74             $env->{'psgi.errors'}->print("Plack server support for psgi.nonblocking is highly recommended.\n");
  75             $cv = AE::cv;
  76         }
  77
  78         tcp_connect $uri->host, $uri->port, sub {
  79             my $server_fh = shift;
  80
  81             # return 502 if connection to server fails
  82             unless ($server_fh) {
  83                 $res->([502, [], ["Bad Gateway"]]);
  84                 $cv->send if $cv;
  85                 return;
  86             }
  87
  88             my $client = AnyEvent::Handle->new(fh => $client_fh);
  89             my $server = AnyEvent::Handle->new(fh => $server_fh);
  90
  91             # forward request from the client, modifying the host and origin
  92             my $headers = $req->headers->clone;
  93             my $host = $uri->host_port;
  94             $headers->header(Host => $host, Origin => "http://$host");
  95             my $hs = HTTP::Request->new('GET', $uri->path, $headers);
  96             $hs->protocol($req->protocol);
  97             $server->push_write($hs->as_string);
  98
  99             my $buffer = "";
 100             my $writer;
 101
 102             # buffer the exchange between the client and server
 103             $client->on_read(sub {
 104                 my $hdl = shift;
 105                 my $buf = delete $hdl->{rbuf};
 106                 $server->push_write($buf);
 107             });
 108             $server->on_read(sub {
 109                 my $hdl = shift;
 110                 my $buf = delete $hdl->{rbuf};
 111
 112                 if ($writer) {
 113                     $writer->write($buf);
 114                     return;
 115                 }
 116
 117                 if (($buffer .= $buf) =~ s/^(.+\r?\n\r?\n)//s) {
 118                     my $http = HTTP::Response->parse($1);
 119                     my @headers;
 120                     $http->headers->remove_header('Status');
 121                     $http->headers->scan(sub { push @headers, @_ });
 122                     $writer = $res->([$http->code, [@headers]]);
 123                     $writer->write($buffer) if $buffer;
 124                     $buffer = undef;
 125                 }
 126             });
 127
 128             # shut down the sockets and exit the loop if an error occurs
 129             $client->on_error(sub {
 130                 $client->destroy;
 131                 $server->push_shutdown;
 132                 $cv->send if $cv;
 133                 $writer->close if $writer;
 134             });
 135             $server->on_error(sub {
 136                 $server->destroy;
 137                 # get the client handle's attention
 138                 $writer->close;
 139             });
 140         };
 141
 142         $cv->recv if $cv;
 143     };
 144 };
 145
 146 1;