File Coverage

blib/lib/Event/RPC/Server.pm

Criterion	Covered	Total	%
statement	238	297	80.1
branch	47	76	61.8
condition	2	4	50.0
subroutine	65	94	69.1
pod	10	77	12.9
total	362	548	66.0

#-----------------------------------------------------------------------
# Copyright (C) 2005-2015 by Jörn Reder .
# All Rights Reserved. See file COPYRIGHT for details.
#
# This module is part of Event::RPC, which is free software; you can
# redistribute it and/or modify it under the same terms as Perl itself.
#-----------------------------------------------------------------------

package Event::RPC::Server;

use Event::RPC;
use Event::RPC::Connection;
use Event::RPC::LogConnection;
use Event::RPC::Message::Negotiate;

use Carp;
use strict;
use utf8;

use IO::Socket;
use Sys::Hostname;

sub get_host                    { shift->{host}                         }
sub get_port                    { shift->{port}                         }
sub get_name                    { shift->{name}                         }
sub get_loop                    { shift->{loop}                         }
sub get_classes                 { shift->{classes}                      }
sub get_singleton_classes       { shift->{singleton_classes}            }
sub get_loaded_classes          { shift->{loaded_classes}               }
sub get_clients_connected       { shift->{clients_connected}            }
sub get_log_clients_connected   { shift->{log_clients_connected}        }
sub get_logging_clients         { shift->{logging_clients}              }
sub get_logger                  { shift->{logger}                       }
sub get_start_log_listener      { shift->{start_log_listener}           }
sub get_objects                 { shift->{objects}                      }
sub get_rpc_socket              { shift->{rpc_socket}                   }
sub get_ssl                     { shift->{ssl}                          }
sub get_ssl_key_file            { shift->{ssl_key_file}                 }
sub get_ssl_cert_file           { shift->{ssl_cert_file}                }
sub get_ssl_passwd_cb           { shift->{ssl_passwd_cb}                }
sub get_ssl_opts                { shift->{ssl_opts}                     }
sub get_auth_required           { shift->{auth_required}                }
sub get_auth_passwd_href        { shift->{auth_passwd_href}             }
sub get_auth_module             { shift->{auth_module}                  }
sub get_listeners_started       { shift->{listeners_started}            }
sub get_connection_hook         { shift->{connection_hook}              }
sub get_load_modules            { shift->{load_modules}                 }
sub get_auto_reload_modules     { shift->{auto_reload_modules}          }
sub get_active_connection       { shift->{active_connection}            }
sub get_message_formats         { shift->{message_formats}              }
sub get_insecure_msg_fmt_ok     { shift->{insecure_msg_fmt_ok}          }

sub set_host                    { shift->{host}                 = $_[1] }
sub set_port                    { shift->{port}                 = $_[1] }
sub set_name                    { shift->{name}                 = $_[1] }
sub set_loop                    { shift->{loop}                 = $_[1] }
sub set_classes                 { shift->{classes}              = $_[1] }
sub set_singleton_classes       { shift->{singleton_classes}    = $_[1] }
sub set_loaded_classes          { shift->{loaded_classes}       = $_[1] }
sub set_clients_connected       { shift->{clients_connected}    = $_[1] }
sub set_log_clients_connected   { shift->{log_clients_connected}= $_[1] }
sub set_logging_clients         { shift->{logging_clients}      = $_[1] }
sub set_logger                  { shift->{logger}               = $_[1] }
sub set_start_log_listener      { shift->{start_log_listener}   = $_[1] }
sub set_objects                 { shift->{objects}              = $_[1] }
sub set_rpc_socket              { shift->{rpc_socket}           = $_[1] }
sub set_ssl                     { shift->{ssl}                  = $_[1] }
sub set_ssl_key_file            { shift->{ssl_key_file}         = $_[1] }
sub set_ssl_cert_file           { shift->{ssl_cert_file}        = $_[1] }
sub set_ssl_passwd_cb           { shift->{ssl_passwd_cb}        = $_[1] }
sub set_ssl_opts                { shift->{ssl_opts}             = $_[1] }
sub set_auth_required           { shift->{auth_required}        = $_[1] }
sub set_auth_passwd_href        { shift->{auth_passwd_href}     = $_[1] }
sub set_auth_module             { shift->{auth_module}          = $_[1] }
sub set_listeners_started       { shift->{listeners_started}    = $_[1] }
sub set_connection_hook         { shift->{connection_hook}      = $_[1] }
sub set_load_modules            { shift->{load_modules}         = $_[1] }
sub set_auto_reload_modules     { shift->{auto_reload_modules}  = $_[1] }
sub set_active_connection       { shift->{active_connection}    = $_[1] }
sub set_message_formats         { shift->{message_formats}      = $_[1] }
sub set_insecure_msg_fmt_ok     { shift->{insecure_msg_fmt_ok}  = $_[1] }

my $INSTANCE;
sub instance { $INSTANCE }

sub get_max_packet_size {
    return Event::RPC::Message->get_max_packet_size;
}

sub set_max_packet_size {
    my $class = shift;
    my ($value) = @_;
    Event::RPC::Message->set_max_packet_size($value);
}

sub new {
    my $class = shift;
    my %par = @_;
    my  ($host, $port, $classes, $name, $logger, $start_log_listener) =
    @par{'host','port','classes','name','logger','start_log_listener'};
    my  ($ssl, $ssl_key_file, $ssl_cert_file, $ssl_passwd_cb, $ssl_opts) =
    @par{'ssl','ssl_key_file','ssl_cert_file','ssl_passwd_cb','ssl_opts'};
    my  ($auth_required, $auth_passwd_href, $auth_module, $loop) =
    @par{'auth_required','auth_passwd_href','auth_module','loop'};
    my  ($connection_hook, $auto_reload_modules) =
    @par{'connection_hook','auto_reload_modules'};
    my  ($load_modules, $message_formats, $insecure_msg_fmt_ok) =
    @par{'load_modules','message_formats','insecure_msg_fmt_ok'};

    $name ||= "Event-RPC-Server";
    $insecure_msg_fmt_ok = 1 unless defined $insecure_msg_fmt_ok;

    #-- for backwards compatibility 'load_modules' defaults to 1
    if ( !exists $par{load_modules} ) {
        $load_modules = 1;
    }

    if ( not $loop ) {
        foreach my $impl ( qw/AnyEvent Event Glib/ ) {
            $loop = "Event::RPC::Loop::$impl";
            eval "use $loop";
            if ( $@ ) {
                $loop = undef;
            }
            else {
                $loop = $loop->new;
                last;
            }
        }
        die "It seems no supported event loop module is installed"
            unless $loop;
    }

    my $self = bless {
        host                    => $host,
        port                    => $port,
        name                    => $name,
        classes                 => $classes,
        singleton_classes       => {},
        logger                  => $logger,
        start_log_listener      => $start_log_listener,
        loop                    => $loop,

        ssl                     => $ssl,
        ssl_key_file            => $ssl_key_file,
        ssl_cert_file           => $ssl_cert_file,
        ssl_passwd_cb           => $ssl_passwd_cb,
        ssl_opts                => $ssl_opts,

        auth_required           => $auth_required,
        auth_passwd_href        => $auth_passwd_href,
        auth_module             => $auth_module,

        load_modules            => $load_modules,
        auto_reload_modules     => $auto_reload_modules,
        connection_hook         => $connection_hook,

        message_formats         => $message_formats,
        insecure_msg_fmt_ok     => $insecure_msg_fmt_ok,

        rpc_socket              => undef,
        loaded_classes          => {},
        objects                 => {},
        logging_clients         => {},
        clients_connected       => 0,
        listeners_started       => 0,
        log_clients_connected   => 0,
        active_connection       => undef,
    }, $class;

    $INSTANCE = $self;

    $self->log ($self->get_name." started");

    return $self;
}

sub DESTROY {
    my $self = shift;

    my $rpc_socket = $self->get_rpc_socket;
    close ($rpc_socket) if $rpc_socket;

    1;
}

sub probe_message_formats {
    my $class = shift;
    my ($user_supplied_formats, $insecure_msg_fmt_ok) = @_;

    my $order_lref      = Event::RPC::Message::Negotiate->message_format_order;
    my $modules_by_name = Event::RPC::Message::Negotiate->known_message_formats;

    my %probe_formats;
    if ( $user_supplied_formats ) {
        @probe_formats{@{$user_supplied_formats}} =
            (1) x @{$user_supplied_formats};
    }
    else {
        %probe_formats = %{$modules_by_name};
    }

    #-- By default we probe all supported formats, but
    #-- not Storable. User has to activate this explicitely.
    if ( not $insecure_msg_fmt_ok ) {
        delete $probe_formats{STOR};
    }

    Event::RPC::Message::Negotiate->set_storable_fallback_ok($insecure_msg_fmt_ok);

    my @supported_formats;
    foreach my $name ( @{$order_lref} ) {
        next unless $probe_formats{$name};

        my $module = $modules_by_name->{$name};
        eval "use $module";

        push @supported_formats, $name unless $@;
    }

    return \@supported_formats;
}

sub setup_listeners {
    my $self = shift;

    #-- Listener options
    my $host      = $self->get_host;
    my $port      = $self->get_port;
    my @LocalHost = $host ? ( LocalHost => $host ) : ();
    $host ||= "*";

    #-- get event loop manager
    my $loop = $self->get_loop;

    #-- setup rpc listener
    my $rpc_socket;
    if ( $self->get_ssl ) {
        eval { require IO::Socket::SSL };
        croak "SSL requested, but IO::Socket::SSL not installed" if $@;
        croak "ssl_key_file not set"  unless $self->get_ssl_key_file;
        croak "ssl_cert_file not set" unless $self->get_ssl_cert_file;

        my $ssl_opts = $self->get_ssl_opts;

        $rpc_socket = IO::Socket::SSL->new (
            Listen          => SOMAXCONN,
            @LocalHost,
            LocalPort       => $port,
            Proto           => 'tcp',
            ReuseAddr       => 1,
            SSL_key_file    => $self->get_ssl_key_file,
            SSL_cert_file   => $self->get_ssl_cert_file,
            SSL_passwd_cb   => $self->get_ssl_passwd_cb,
            ($ssl_opts?%{$ssl_opts}:()),
        ) or die "can't start SSL RPC listener: $IO::Socket::SSL::ERROR";
    }
    elsif ($host eq "unix/") {
        require IO::Socket::UNIX;
        unlink $port;
        $rpc_socket = IO::Socket::UNIX->new (
            Type      => IO::Socket::UNIX::SOCK_STREAM(),
            Listen    => SOMAXCONN,
            Local     => $port,
        ) or die "can't start Unix Domain RPC listener at $port: $!";
    }
    else {
        require IO::Socket::INET;
        $rpc_socket = IO::Socket::INET->new (
            Listen    => SOMAXCONN,
            @LocalHost,
            LocalPort => $port,
            Proto     => 'tcp',
            ReuseAddr => 1,
        ) or die "can't start TCP RPC listener: $!";
    }

    $self->set_rpc_socket($rpc_socket);

    $loop->add_io_watcher (
        fh      => $rpc_socket,
        poll    => 'r',
        cb      => sub { $self->accept_new_client($rpc_socket); 1 },
        desc    => "rpc listener port $port",
    );

    if ( $self->get_ssl ) {
        $self->log ("Started SSL RPC listener on port $host:$port");
    } else {
        $self->log ("Started RPC listener on $host:$port");
    }

    # setup log listener
    if ( $self->get_start_log_listener ) {
        my $log_socket;
        if ($host eq "unix/") {
            $port =~ s/\.sock//;
            $port .= ".log.sock";
            unlink $port;
            $log_socket = IO::Socket::UNIX->new (
                Type      => IO::Socket::UNIX::SOCK_STREAM(),
                Listen    => SOMAXCONN,
                Local     => $port,
            ) or die "can't start Unix Domain log listener at $port: $!";
        }
        else {
            $log_socket = IO::Socket::INET->new (
                Listen    => SOMAXCONN,
                LocalPort => $port + 1,
                @LocalHost,
                Proto     => 'tcp',
                ReuseAddr => 1,
            ) or die "can't start log listener: $!";
        }

        $loop->add_io_watcher (
            fh      => $log_socket,
            poll    => 'r',
            cb      => sub { $self->accept_new_log_client($log_socket); 1 },
            desc    => "log listener port ".($port+1),
        );

        $self->log ("Started log listener on $host:".($port+1));
    }

    $self->determine_singletons;

    $self->set_listeners_started(1);

    1;
}

sub setup_auth_module {
    my $self = shift;

    #-- Exit if no auth is required or setup already
    return if not $self->get_auth_required;
    return if     $self->get_auth_module;

    #-- Default to Event::RPC::AuthPasswdHash
    require Event::RPC::AuthPasswdHash;

    #-- Setup an instance
    my $passwd_href = $self->get_auth_passwd_href;
    my $auth_module = Event::RPC::AuthPasswdHash->new ($passwd_href);
    $self->set_auth_module($auth_module);

    1;
}

sub prepare {
    my $self = shift;

    $self->setup_listeners
        unless $self->get_listeners_started;

    $self->setup_auth_module;

    #-- Filter unsupported message formats
    $self->set_message_formats(
        $self->probe_message_formats(
            $self->get_message_formats,
            $self->get_insecure_msg_fmt_ok
        )
    );

    1;
}

sub start {
    my $self = shift;

    #-- Prepare server for startup
    $self->prepare;

    my $loop = $self->get_loop;

    $self->log ("Enter main loop using ".ref($loop));

    $loop->enter;

    $self->log ("Server stopped");

    1;
}

sub stop {
    my $self = shift;

    $self->get_loop->leave;

    1;
}

sub determine_singletons {
    my $self = shift;

    my $classes = $self->get_classes;
    my $singleton_classes = $self->get_singleton_classes;

    foreach my $class ( keys %{$classes} ) {
        foreach my $method ( keys %{$classes->{$class}} ) {
            # check for singleton returner
            if ( $classes->{$class}->{$method} eq '_singleton' ) {
                # change to constructor
                $classes->{$class}->{$method} = '_constructor';
                # track that this class is a singleton
                $singleton_classes->{$class} = 1;
                last;
            }
        }
    }

    1;
}

sub accept_new_client {
    my $self = shift;
    my ($rpc_socket) = @_;

    my $client_socket = $rpc_socket->accept or return;

    Event::RPC::Connection->new ($self, $client_socket);

    $self->set_clients_connected ( 1 + $self->get_clients_connected );

    1;
}

sub accept_new_log_client {
    my $self = shift;
    my ($log_socket) = @_;

    my $client_socket = $log_socket->accept or return;

    my $log_client =
        Event::RPC::LogConnection->new($self, $client_socket);

    $self->set_log_clients_connected ( 1 + $self->get_log_clients_connected );
    $self->get_logging_clients->{$log_client->get_cid} = $log_client;
    $self->get_logger->add_fh($client_socket)
        if $self->get_logger;

    $self->log(2, "New log client connected");

    1;
}

sub load_class {
    my $self = shift;
    my ($class) = @_;

    Event::RPC::Connection->new ($self)->load_class($class);

    return $class;
}

sub log {
    my $self = shift;
    my $logger = $self->get_logger;
    return unless $logger;
    $logger->log(@_);
    1;
}

sub remove_object {
    my $self = shift;
    my ($object) = @_;

    my $objects = $self->get_objects;

    if ( not $objects->{"$object"} ) {
        warn "Object $object not registered";
        return;
    }

    delete $objects->{"$object"};

    $self->log(5, "Object '$object' removed");

    1;
}

sub register_object {
    my $self = shift;
    my ($object, $class) = @_;

    my $objects = $self->get_objects;

    my $refcount;
    if ( $objects->{"$object"} ) {
        $refcount = ++$objects->{"$object"}->{refcount};
    } else {
        $refcount = 1;
        $objects->{"$object"} = {
            object   => $object,
            class    => $class,
            refcount => 1,
        };
    }

    $self->log(5, "Object '$object' registered. Refcount=$refcount");

    1;
}

sub deregister_object {
    my $self = shift;
    my ($object) = @_;

    my $objects = $self->get_objects;

    if ( not $objects->{"$object"} ) {
        warn "Object $object not registered";
        return;
    }

    my $refcount = --$objects->{"$object"}->{refcount};

    my ($class) = split(/=/, $object);
    if ( $self->get_singleton_classes->{$class} ) {
        # never deregister singletons
        $self->log(4, "Skip deregistration of singleton '$object'");
        return;
    }

    $self->log(5, "Object '$object' deregistered. Refcount=$refcount");

    $self->remove_object($object) if $refcount == 0;

    1;
}

sub print_object_register {
    my $self = shift;

    print "-"x70,"\n";

    my $objects = $self->get_objects;
    foreach my $oid ( sort keys %{$objects} ) {
        print "$oid\t$objects->{$oid}->{refcount}\n";
    }

    1;
}

1;

__END__