#!/usr/bin/perl

use strict;
use warnings;

use threads;

use HTTP::Daemon ();
use SOAP::Lite ();
use SOAP::Transport::HTTP ();
use Thread::Semaphore ();

package API::Utils::Daemon;
use base 'SOAP::Transport::HTTP::Server';

sub DESTROY { SOAP::Trace::objects('()') }

# maximum number of concurrent request handling threads
our $MAX_THREADS = 4;

return 1;

###############################################################################

sub new {
    my $self = shift;

    unless(ref $self) {
        my $class = ref($self) || $self;
        my (@params, @methods);
        while(@_) {
            if($class->can($_[0])) {
                push(@methods, shift() => shift);
            }
            else {
                push(@params, shift);
            }
        }
        $self = $class->SUPER::new;

        $self->{'_daemon'} = HTTP::Daemon->new(@params)
            or die $!;

        $self->myuri(URI->new($self->{'_daemon'}->url)->canonical->as_string);
        while(@methods) {
            my ($method, $params) = splice(@methods, 0, 2);
            if(ref $params eq 'ARRAY') {
                $self->$method(@{$params});
            }
            else {
                $self->$method($params);
            }
        }
        $self->{'_sema'} = Thread::Semaphore->new($MAX_THREADS);
        SOAP::Trace::objects('()');
    }

    return $self;
}

###############################################################################

sub handle {
    my $self = shift;
    my $connection = shift;

    printf("[%s] - - - Accepting Connections\n",
        localtime() . "",
        );

    my %threads = ();
    while(my $connection = $self->{'_daemon'}->accept()) {
        $self->{'_sema'}->down();

        my $thr = threads->create(\&handle_one_request,
            $self,
            $connection,
            )->detach();
    }
}

###############################################################################

sub handle_one_request {
    my $self = shift;
    my $connection = shift;

    eval {
        while(my $request = $connection->get_request()) {

            $self->request($request);
            $self->SUPER::handle();
            $connection->send_response($self->response());

            my $action = $request->header("soapaction");
            if(not defined $action) {
                next;
            }
            $action =~ s/^urn://;
            $action =~ s!^/!!;
            $action =~ s!\#!/!g;
            $action =~ s/"//g;
            printf("[%s] %s \"%s %s\" %s\n",
                localtime() . "",
                $connection->peerhost,
                $request->method,
                $action,
                $self->response->code,
                );

            undef $request;
        }
        $connection->close();
        undef $connection;
    };
    $self->{'_sema'}->up();
    threads->exit();
}

###############################################################################

sub max_threads($) {
    my $self = shift;
    my $max_threads = shift;

    # we can't just create a new semaphore because there might be threads
    # already running which will increment the semaphore as they return.
    # as a result this call may block if the new max_threads is smaller
    # than the previous max_threads.

    my $thread_change = $max_threads - $MAX_THREADS;
    if($thread_change > 0) {
        for(1 .. $thread_change) {
            $self->{'_sema'}->up();
        }
    }
    elsif($thread_change < 0) {
        for(1 .. abs($thread_change)) {
            $self->{'_sema'}->down();
        }
    }

    $MAX_THREADS = $max_threads;
    return $self;
}

###############################################################################

