POE::Component::Server::Bayeux - Bayeux/cometd server implementation in POE

  use POE qw(Component::Server::Bayeux);

  # Create the server, listening on port 8080
  my $server = POE::Component::Server::Bayeux->spawn(
      Port  => 8080,
      Alias => 'bayeux_server',
  );

  # Create a local client, a reply-bot
  POE::Session->create(
      inline_states => {
          _start => sub {
              my ($kernel, $heap) = @_[KERNEL, HEAP];
              $kernel->alias_set('test_local_client');

              # Subscribe to /chat/demo, assigning a state for events
              $kernel->post('bayeux_server', 'subscribe', {
                  channel => '/chat/demo',
                  client_id => $heap->{client_id},
                  args => {
                      state => 'subscribe_response',
                  },
              });
          },
          subscribe_response => sub {
              my ($kernel, $heap, $message) = @_[KERNEL, HEAP, ARG0];

              # Don't auto-reply to my own messages
              return if $message->{clientId} eq $heap->{client_id};

              # Auto-reply to every message posted
              $kernel->post('bayeux_server', 'publish', {
                  channel => $message->{channel},
                  client_id => $heap->{client_id},
                  data => {
                      user => 'Autobot',
                      chat => "I got your message, ".($message->{data}{user} || 'anon'),
                  },
              });
          },
      },
      heap => {
          client_id => 'test_local_client',
      },
  );

  $poe_kernel->run();

  $kernel->post('bayeux_server', 'subscribe', {
      channel => '/chat/demo',
      client_id => 'local_client',
      args => {
          state => 'subscribe_events',
      },
  });

package POE::Component::Server::Bayeux;

use strict;
use warnings;

use POE qw(
    Component::Server::HTTP
    Component::Server::Bayeux::Client
    Component::Server::Bayeux::Request
);
use HTTP::Status; # for RC_OK
use Params::Validate qw(CODEREF HASHREF validate validate_with);
use POE::Component::Server::Bayeux::Utilities qw(:all);

# Logger modules
use base qw(Class::Accessor);
__PACKAGE__->mk_accessors(qw(logger session));
use Log::Log4perl qw(get_logger :levels);
use Log::Log4perl::Appender;
use Log::Log4perl::Layout;

# Basic HTTP server modules
use URI;

## Class globals ###

our $VERSION = '0.04';
our $protocol_version = '1.0';
our $supported_connection_types = [ 'long-polling' ];

## Class locals ###

our %file_types = (
    'application/javascript' => [ qr/\.js$/i ],
    'text/html'              => [ qr/\.html?$/i ],
    'text/css'               => [ qr/\.css$/i ],
    'image/png'              => [ qr/\.png$/i ],
    'image/jpeg'             => [ qr/\.jpe?g$/i ],
    'image/gif'              => [ qr/\.gif$/i ],
);

{
    package POE::Component::Server::Bayeux::Logger;

    # Performing a dump of a data structure is a useful thing for debuging,
    # but in live code you may not need it.  This is one way of accomplishing
    # this.
    
    use strict;
    use warnings;
    use Carp;

    __PACKAGE__->mk_wrapped_levels(qw(trace debug info warn error fatal));

    sub new {
        my ($class, %self) = @_;

        croak "Pass 'logger'" unless $self{logger};

        if (! $self{dumper}) {
            require JSON::XS;
            my $json = JSON::XS->new();
            $json->indent(1);
            $json->ascii(1);
            $json->space_after(1);
            $self{dumper} = sub {
                $json->encode(@_);
            };
        }

        return bless \%self, $class;
    }

    sub mk_wrapped_levels {
        my ($class, @levels) = @_;

        no strict 'refs';
        foreach my $level (@levels) {
            *{"${class}::$level"} = sub {
                my ($self, $message, @extra) = @_;
                my $logger = $self->{logger};

                # Make sure before creating a dump of the ref's that I have to
                my $is_method = 'is_' . $level;
                return unless $logger->$is_method;

                my $total = $message;
                if (@extra) {
                    $total .= "\n" unless $total =~ m/\n$/s;
                    $total .= $self->{dumper}->(@extra);
                }

                $logger->$level($total);
            };
        }
    }
}

## Class methods ###

sub spawn {
    my $class = shift;
    my %args = validate(@_, {
        Port           => { default => '80' },
        Alias          => { default => 'bayeux' },
        AnonPublish    => { default => 0 },
        Debug          => { default => 0 },
        LogFile        => { default => '' },
        # Client must not go 2 minutes without having an outstanding connect
        ConnectTimeout => { default => 2 * 60 },
        DocumentRoot   => { default => '/var/www' },
        DirectoryIndex => { default => [ 'index.html' ] },
        TypeExpires    => { default => {} },
        PostHandle     => { default => undef, type => CODEREF },
        Services       => { default => {} },
        MessageACL     => { default => sub {}, type => CODEREF },
        Callback       => { default => sub {}, type => CODEREF },
        ContentHandler => { default => {}, type => HASHREF },
        ClientMaxConnections => { default => 10 },
        Logger         => 0,
    });

    my $logger = $args{Logger};

    # Setup logger
    if (! $logger) {
        $logger = Log::Log4perl->get_logger('bayeux_server');
        my $logger_layout = Log::Log4perl::Layout::PatternLayout->new("[\%d] \%p: \%m\%n");
        $logger->level($args{Debug} ? $DEBUG : $INFO);

        my $stdout_appender = Log::Log4perl::Appender->new(
            'Log::Log4perl::Appender::Screen',
            name => 'screenlog',
            stderr => 0,
        );
        $stdout_appender->layout($logger_layout);

        $logger->add_appender($stdout_appender);

        if ($args{LogFile}) {
            my $file_appender = Log::Log4perl::Appender->new(
                'Log::Log4perl::Appender::File',
                name => 'filelog',
                filename => $args{LogFile},
            );
            $file_appender->layout( $logger_layout );

            $logger->add_appender($file_appender);
        }
    }

    # Wrap the Log4perl logger in my own class
    $logger = POE::Component::Server::Bayeux::Logger->new(
        logger => $logger,
    );

    # Create HTTP server
    my $http_aliases = POE::Component::Server::HTTP->new(
        Port => $args{Port},
        ContentHandler => {
            '/cometd' => sub {
                $poe_kernel->call( $args{Alias}, 'handle_cometd', @_ );
            },
            '/' => sub {
                $poe_kernel->call( $args{Alias}, 'handle_generic', @_ );
            },
            %{ $args{ContentHandler} },
        },
    );

    my $self = bless { %args, logger => $logger }, $class;

    # Create manager session
    $self->{session} = POE::Session->create(
        inline_states => {
            _start => \&manager_start,
            _stop  => \&manager_stop,
            shutdown => \&manager_shutdown,

            handle_cometd    => \&handle_cometd,
            handle_generic   => \&http_server_generic,
            delay_request    => \&delay_request,
            complete_request => \&complete_request,
            check_timeouts   => \&check_timeouts,
            
            subscribe        => \&subscribe,
            unsubscribe      => \&unsubscribe,
            publish          => \&publish,

            client_push       => \&client_push,
            client_connect    => \&client_connect,
            client_disconnect => \&client_disconnect,

            delay_sub         => \&delay_sub,
            delay_sub_cb      => \&delay_sub_cb,
        },
        heap => {
            args => \%args,
            manager => $args{Alias},
            clients => {
            #   example_client_id => {
            #       subscriptions => {
            #           '/chat/demo/not_real' => 1,
            #       },
            #   },
            },
            requests => {
            #   example_request_id => 1,
            },
            requests_by_ip => {},
            logger => $logger,
            http_aliases => $http_aliases,
        },
        ($ENV{POE_DEBUG} ? (
        options => { trace => 1, debug => 1 },
        ) : ()),
    );

    return $self;
}

###### POE States ######################

sub manager_start {
    my ($kernel, $heap) = @_[KERNEL, HEAP];

    $kernel->alias_set( $heap->{manager} );

    $kernel->delay('check_timeouts', 30);

    $heap->{logger}->info("Bayeux server started.  Connect to port $$heap{args}{Port}");

    if ($ENV{POE_DEBUG}) {
        $kernel->alias_resolve($heap->{http_aliases}{httpd})->option( trace => 1, debug => 1 );
        $kernel->alias_resolve($heap->{http_aliases}{tcp})->option( trace => 1, debug => 1 );
    }
}

sub manager_stop {
    my ($kernel, $heap) = @_[KERNEL, HEAP];
}

sub manager_shutdown {
    my ($kernel, $heap) = @_[KERNEL, HEAP];

    $heap->{logger}->info("Shutting down");

    while (my $request = values %{ $heap->{requests} }) {
        $request->complete();
    }

    $kernel->alarm_remove_all();
    $kernel->alias_set( $heap->{manager} );

    $kernel->call( $heap->{http_aliases}{httpd}, 'shutdown' );
    $kernel->call( $heap->{http_aliases}{tcp}, 'shutdown' );
}

sub http_server_generic {
    my ($kernel, $heap, $request, $response) = @_[KERNEL, HEAP, ARG0, ARG1];

    my $uri = URI->new($request->uri);
    my $path = $heap->{args}{DocumentRoot} . '/' . $uri->path;

    # Attempt to find a directory index
    if (-d $path) {
        $path .= '/' unless $path =~ m{/$};
        foreach my $index_name (@{ $heap->{args}{DirectoryIndex} }) {
            next unless -f $path . $index_name;
            $path .= $index_name;
            last;
        }

    }
    if (-d $path) {
        $response->code(RC_OK);
        $response->content("Directory listing denied");
    }
    elsif (-f $path) {
        $response->code(RC_OK);
        open my $in, '<', $path;
        if (! $in) {
            $response->content("Unable to open '$path': $!");
            return RC_OK;
        }

        # Find a file type
        my $type;
        foreach my $possible_type (keys %file_types) {
            next unless grep { $path =~ $_ } @{ $file_types{$possible_type} };
            $type = $possible_type;
            last;
        }
        $type ||= 'text/plain';
        $response->content_type($type);

        if (my $whence = $heap->{args}{TypeExpires}{$type}) {
            $response->expires( time() + $whence );
        }

        my $content;
        {
            local $/ = undef;
            $content = <$in>;
        }
        close $in;
        $response->content($content);

        my $ip = $request->header('X-Forwarded-For') || $request->{connection}{remote_ip};
        $heap->{logger}->info(sprintf 'Serving %s %s %s', $ip, $uri->path, $response->content_type);
    }
    else {
        $response->code(RC_NOT_FOUND);
        $response->content("Path '".$uri->path."' not found");
    }

    if ($heap->{args}{PostHandle}) {
        $heap->{args}{PostHandle}($request, $response);
    }

    # Ensure no KeepAlive
    $request->header(Connection => 'close');

    return RC_OK;
}

## Remote clients, long-polling ###

sub handle_cometd {
    my ($kernel, $heap, $request, $response) = @_[KERNEL, HEAP, ARG0, ARG1];

    # Deny based upon ClientMaxConnections restrictions

    my $ip = $request->header('X-Forwarded-For') || $request->{connection}{remote_ip};
    if (! $ip) {
        $ip = '0.0.0.0';
        $heap->{logger}->error("No IP found for cometd request");
    }

    $heap->{requests_by_ip}{$ip} ||= {};
    my @request_ids = keys %{ $heap->{requests_by_ip}{$ip} };
    if (int @request_ids > $heap->{args}{ClientMaxConnections}) {
        $heap->{logger}->info("Denying $ip; too many connections (".int(@request_ids).")");

        $response->code(RC_SERVICE_UNAVAILABLE);
        $response->header( 'Content-Type' => "text/json; charset=utf-8" );
        $response->content( '{ "error": "Too many connections from your IP", "successful": false }' );
        return RC_OK;
    }
    else {
        #$heap->{logger}->info("IP $ip has " . int(@request_ids) . " connections");
    }

    # Proceed with processing

    #$heap->{logger}->debug("Handling new cometd request");

    #$heap->{logger}->debug($request->as_string);

    my $bayeux_request = POE::Component::Server::Bayeux::Request->new(
        request => $request,
        response => $response,
        server_heap => $heap,
        ip => $ip,
    );
    $bayeux_request->handle();

    if ($bayeux_request->is_complete) {
        $heap->{logger}->debug("Immediate remote response:", $bayeux_request->json_response);
        return RC_OK;
    }
    else {
        $heap->{requests}{ $bayeux_request->id } = $bayeux_request;
        $heap->{requests_by_ip}{$ip}{ $bayeux_request->id } = $bayeux_request;
        return RC_WAIT;
    }
}

sub delay_request {
    my ($kernel, $heap, $request_id, $delay) = @_[KERNEL, HEAP, ARG0, ARG1];

    $heap->{logger}->debug("Delaying $delay to process request $request_id");
    $kernel->delay_add('complete_request', $delay, $request_id);
}

sub complete_request {
    my ($kernel, $heap, $request_id) = @_[KERNEL, HEAP, ARG0];

    return unless defined $heap->{requests}{$request_id};
    my $request = delete $heap->{requests}{$request_id};

    my $ip = $request->ip;
    if ($heap->{requests_by_ip}{$ip}) {
        delete $heap->{requests_by_ip}{$ip}{$request_id};
        if (! keys %{ $heap->{requests_by_ip}{$ip} }) {
            delete $heap->{requests_by_ip}{$ip};
        }
    }
    else {
        $heap->{logger}->error("Couldn't find requests by ip ".($ip || 'undef'));
    }

    eval {
        $request->complete();
    };
    if ($@) {
        $heap->{logger}->error("Couldn't complete request $request_id ($@) - mayhap the client went away?");
    }
    else {
        $heap->{logger}->debug("Delayed remote response to request $request_id from $ip:", $request->json_response);
    }
}

sub check_timeouts {
    my ($kernel, $heap) = @_[KERNEL, HEAP];

    # Setup my next call time
    $kernel->delay('check_timeouts', 30);

    foreach my $client_id (keys %{ $heap->{clients} }) {
        my $client = POE::Component::Server::Bayeux::Client->new(
            id => $client_id,
            server_heap => $heap,
        );
        $client->check_timeout();
        if ($client->is_error) {
            $heap->{logger}->info("Found timeed out client $client_id in check_timeouts()");
        }
    }
}

sub delay_sub {
    my ($kernel, $heap, $delay_name, $delay_sec, $sub) = @_[KERNEL, HEAP, ARG0 .. $#_];

    if (my $existing = $heap->{delay_sub}{$delay_name}) {
        return;
    }
    $kernel->delay_add('delay_sub_cb', $delay_sec, $delay_name);
    $heap->{delay_sub}{$delay_name} = $sub;
}
sub delay_sub_cb {
    my ($kernel, $heap, $delay_name) = @_[KERNEL, HEAP, ARG0];

    my $sub = delete $heap->{delay_sub}{$delay_name};
    &$sub();
}

## Client agnostic, no auth performed ###

sub subscribe {
    my ($kernel, $heap, $args) = @_[KERNEL, HEAP, ARG0];

    my @args = %$args;
    my %args;
    eval {
        %args = validate(@args, {
            channel => 1,
            client_id => 1,
            args => { default => {} },
        });
    };
    if ($@) {
        $heap->{logger}->error("subscribe() invalid call: $@");
        return;
    }

    # If subscribe() was called by another POE session
    if ($_[SESSION] != $_[SENDER]) {
        # Create a client, thereby storing the session in the client heap
        my $client = POE::Component::Server::Bayeux::Client->new(
            id => $args{client_id},
            session => $_[SENDER],
            server_heap => $heap,
        );
    }

    $args{args}{subscribed} = time;
    $heap->{clients}{ $args->{client_id} }{subscriptions}{ $args->{channel} } = $args{args};

    $heap->{args}{Callback}->({
        event => 'subscribe',
        client_id => $args->{client_id},
        channel => $args->{channel},
    });
}

sub unsubscribe {
    my ($kernel, $heap, $sender, $args) = @_[KERNEL, HEAP, SENDER, ARG0];

    my @args = %$args;
    my %args;
    eval {
        %args = validate(@args, {
            channel => 1,
            client_id => 1,
        });
    };
    if ($@) {
        $heap->{logger}->error("unsubscribe() invalid call: $@");
        return;
    }

    my $client_heap = $heap->{clients}{ $args->{client_id} };
    return unless $client_heap;
    return unless $client_heap->{subscriptions}{ $args->{channel} };
    delete $client_heap->{subscriptions}{ $args->{channel} };

    $heap->{args}{Callback}->({
        event => 'unsubscribe',
        client_id => $args->{client_id},
        channel => $args->{channel},
    });
}

sub publish {
    my ($kernel, $heap, $sender, $args) = @_[KERNEL, HEAP, SENDER, ARG0];

    my @args = %$args;
    my %args;
    eval {
        %args = validate(@args, {
            channel => 1,
            client_id => 0,
            data => 1,
            id => 0,
            ext => 0,
            timestamp => 0,
        });
    };
    if ($@) {
        $heap->{logger}->error("publish() invalid call: $@");
        return;
    }

    # Check each subscription, getting list of who to send this to

    my %send_to_clients;
    CLIENT:
    foreach my $client_id (keys %{ $heap->{clients} }) {
        my $client_heap = $heap->{clients}{$client_id};
        next unless $client_heap->{subscriptions};
        foreach my $subscribed (keys %{ $client_heap->{subscriptions} }) {
            next unless channel_match($args{channel}, $subscribed);
            my $subscription_args = $client_heap->{subscriptions}{$subscribed};
            $send_to_clients{ $client_id } = $subscription_args;
            next CLIENT;
        }
    }

    $heap->{args}{Callback}->({
        event => 'publish',
        %args,
    });

    my @send_to_clients = keys %send_to_clients;
    if (! @send_to_clients) {
        $heap->{logger}->debug("publish('$args{channel}') had no Bayeux subscribers");
        return;
    }

    # Construct deliver packet

    my %deliver = (
        map { $_ => $args{$_} }
        grep { defined $args{$_} }
        qw(channel data id ext timestamp)
    );
    $deliver{clientId} = $args{client_id} if defined $args{client_id};

    foreach my $client_id (@send_to_clients) {
        my $client = POE::Component::Server::Bayeux::Client->new(
            id => $client_id,
            server_heap => $heap,
        );
        next if ! $client || $client->is_error;
        $client->send_message(\%deliver, $send_to_clients{$client_id});
    }
}

sub client_push {
    my ($kernel, $heap, $sender, $args) = @_[KERNEL, HEAP, SENDER, ARG0];
    
    # Validate args
    my @args = %$args;
    my %args;
    eval {
        %args = validate_with(
            params => \@args,
            spec => {
                channel => 1,
                client_id => 1,
            },
            allow_extra => 1,
        );
    };
    if ($@) {
        $heap->{logger}->error("client_push() invalid call: $@");
        return;
    }

    # Construct the packet

    my %deliver = (%args);
    $deliver{clientId} = delete $deliver{client_id};

    # Find the client and push the packet

    my $client = POE::Component::Server::Bayeux::Client->new(
        id => $args{client_id},
        server_heap => $heap,
    );
    if (! $client) {
        $heap->{logger}->error("client_push() failed: no client found from $args{client_id}");
        return;
    }
    if ($client->is_error) {
        $heap->{logger}->debug("client_push() failed: client $args{client_id} in error state (".$client->is_error.")");
        return;
    }
    $client->send_message(\%deliver);

    $heap->{args}{Callback}->({
        event => 'client_push',
        %args,
    });
}

sub client_connect {
    my ($kernel, $heap, $sender, $args) = @_[KERNEL, HEAP, SENDER, ARG0];
    
    my @args = %$args;
    my %args;
    eval {
        %args = validate(@args, {
            client_id => 1,
            ip => 0,
            session => 0,
        });
    };
    if ($@) {
        $heap->{logger}->error("client_connect() invalid call: $@");
        return;
    }

    # Nothing to do here; the Client class adds it to my $heap->{clients}

    $heap->{args}{Callback}->({
        event => 'client_connect',
        %args,
    });
}

sub client_disconnect {
    my ($kernel, $heap, $sender, $args) = @_[KERNEL, HEAP, SENDER, ARG0];
    
    my @args = %$args;
    my %args;
    eval {
        %args = validate(@args, {
            client_id => 1,
        });
    };
    if ($@) {
        $heap->{logger}->error("client_disconnect() invalid call: $@");
        return;
    }

    my $client_heap = $heap->{clients}{ $args{client_id} };
    return unless $client_heap;

    foreach my $channel (keys %{ $client_heap->{subscriptions} }) {
        # Do a call since this needs to happen right now
        $kernel->call($_[SESSION], 'unsubscribe', {
            client_id => $args{client_id},
            channel => $channel,
        });
    }

    delete $heap->{clients}{ $args{client_id} };

    $heap->{args}{Callback}->({
        event => 'client_disconnect',
        %args,
    });
}

1;