view release on metacpan or  search on metacpan

eg/llng-pubsub-server  view on Meta::CPAN

    my $log  = "$ip - - $date $msg -\n";
    if ( $accessLog eq 'STDERR' ) {
        print STDERR $log;
    }
    else {
        print $log;
    }
    debug "Request: $msg";
}

if ( $accessLog and $accessLog ne '-' and $accessLog ne 'STDERR' ) {
    debug "Access logs will be written into $accessLog\n";
    open STDOUT, '>>', $accessLog or die "Unable to write logs: $!";
    if ($daemon) {
        open STDERR, '>&', fileno(STDOUT) or die "Can't dup STDERR: $!";
        STDERR->autoflush(1);
    }
}

if ( $cert xor $key ) {
    die '--cert and --key must be used together';
}

unless ($token) {
    warning "No token given, this means that anybody can use this server"
      unless $quiet;
}

# 2. Create server

my @servers;
my $selector = IO::Select->new;
$addr = [0] unless @$addr;

foreach my $a (@$addr) {
    my $server;
    my $p;

    # IPv6 addresses with port
    if ( $a =~ s/^\s*\[($RE{net}{IPv6})\]:(\d+)\s*$/$1/ ) {
        $p = $2;
    }
    elsif ( $a =~ s/^\s*($RE{net}{IPv4}):(\d+)\s*$/$1/ ) {
        $p = $2;
    }
    elsif ( $a and $a !~ s/^\s*\[?($RE{net}{IPv4}|$RE{net}{IPv6})\]?\s*/$1/ ) {
        die qq'Malformed IP address "$a"';
    }
    else {
        $p = $port || ( $ENV{PUBSUB_PORT} || ( $cert ? 8443 : 8080 ) );
    }

    my %args = (
        ( $a ? ( LocalAddr => $a ) : () ),
        LocalPort => $p,
        Listen    => $maxWaitingConn,
        ReuseAddr => 1,
        ReusePort => 1,
    );
    if ($cert) {
        require IO::Socket::SSL;
        $server = IO::Socket::SSL->new(
            %args,
            SSL_cert_file => $cert,
            SSL_key_file  => $key,
            SSL_server    => 1,
        );
    }
    else {
        require IO::Socket::IP;
        $server = IO::Socket::IP->new( %args, Proto => 'tcp', );
    }

    die "Cannot create socket ("
      . ( $a ? "address: $a, " : '' )
      . "port: $p): $!"
      unless $server;

    debug 'Listening on ' . ( $a ? "$a:" : 'port ' ) . $p;
    $selector->add($server);
    push @servers, $server;
}

if ($group) {
    my $grp = getgrnam($group) or die "Can't change gid to $group";
    setgid($grp)               or die "setgid: $!";
}

if ($user) {
    my $uid = getpwnam($user) or die "Can't change uid to $user";
    setuid($uid)              or die "setuid: $!";
}

my $json = JSON->new->utf8;

# 3. Manage requests

my %clients;

while (1) {
    for my $sock ( $selector->can_read(0.1) ) {
        my ($server) = grep { $sock eq $_ } @servers;
        if ($server) {
            my $client = $server->accept or next;
            $client->autoflush(1);
            $selector->add($client);
        }
        else {
            my $fileno = fileno($sock);
            my $buf    = '';
            my $n      = sysread( $sock, $buf, 4096 );
            if ( !defined $n || $n == 0 ) {
                cleanupClient($sock);
                next;
            }

            # Known clients are subscribers, we can discard
            if ( exists $clients{$fileno} ) {
                my $frame = $clients{$fileno}->{frame};
                $frame->append($buf);
                while ( my $msg = $frame->next ) {