package MooseX::JobControl::Engine;
use strict;
use Moose;
use MooseX::AttributeHelpers;
use MooseX::JobControl::Types;

with qw(MooseX::Workers);

has job_queue => (
    isa     => 'JobQueue',
    is      => 'rw',
    coerce  => 1,
    default => sub {
        require MooseX::JobControl::JobQueue::Array;
        MooseX::JobControl::JobQueue::Array->new();
    },

    handles => [
        qw(
          add_job
          remove_last_job
          next_job
          add_next_job
          has_jobs
          jobs
          )
    ]
);

has process_list => (
    metaclass  => 'Collection::Hash',
    isa        => 'HashRef',
    is         => 'ro',
    auto_deref => 1,
    default    => sub { {} },
    provides   => {
        set    => 'add_process',
        get    => 'get_process',
        empty  => 'has_process',
        count  => 'num_processes',
        delete => 'remove_process',
    }
);

has observers => (
    metaclass  => 'Collection::Array',
    isa        => 'ArrayRef',
    is         => 'ro',
    auto_deref => 1,
    default    => sub {
        require MooseX::JobControl::Logger;
        [ MooseX::JobControl::Logger->new(), ];
    },
    provides => {
        push    => 'add_observer',
        pop     => 'remove_last_observer',
        shift   => 'remove_next_observer',
        unshift => 'add_next_observer',
        count   => 'has_observers',
    }
);

sub notify_observers {
    my ( $self, $event ) = splice @_, 0, 2;
    for my $o ( $self->observers ) {
        $o->can($event)
          ? $o->$event(@_)
          : $o->notify( $event => @_ );
    }
}

before 'run' => sub { (shift)->notify_observers( 'starting_job_queue' => @_ ) };

after 'run' => sub { (shift)->notify_observers( 'finished_job_queue' => @_ ) };

before 'add_process' =>
  sub { (shift)->notify_observers( 'adding_process' => @_ ) };

after 'add_process' =>
  sub { (shift)->notify_observers( 'added_process' => @_ ) };

before 'remove_process' =>
  sub { (shift)->notify_observers( 'removing_process' => @_ ) };

after 'remove_process' =>
  sub { (shift)->notify_observers( 'removed_process' => @_ ) };

before 'add_job' => sub { (shift)->notify_observers( 'adding_job' => @_ ) };
before 'add_job' => sub { (shift)->notify_observers( 'added_job'  => @_ ) };

sub run {
    my ($self) = @_;

    while ( $self->has_jobs ) {
        sleep(1) while $self->check_worker_threashold;
        my $job = $self->next_job;
        $self->add_process( $self->spawn($job) );    # spawn returns PID, ID
        POE::Kernel->run();
    }
}

sub worker_manager_start {
    my $self = shift;
    $self->notify_observers( 'started_worker_manager' => @_ );
}

sub worker_manager_stop {
    my $self = shift;
    $self->notify_observers( 'started_worker_manager' => @_ );
}

sub max_workers_reached {
    my $self = shift;
    $self->notify_observers( 'started_worker_manager' => @_ );
}

sub worker_stdout {
    my $self = shift;
    my $pid = $self->get_process( $_[1] ) || '???';
    $self->notify_observers( 'worker_stdout' => ( @_, $pid ) );
}

sub worker_stderr {
    my $self = shift;
    my $pid = $self->get_process( $_[1] ) || '???';
    $self->notify_observers( 'worker_stderr' => ( @_, $pid ) );
}

sub worker_error {
    my $self = shift;
    my $pid = $self->get_process( $_[3] ) || '???';
    $self->notify_observers( 'worker_error' => ( @_, $pid ) );
}

sub worker_done {
    my $self = shift;
    $self->notify_observers(
        'worker_done' => @_,
        $self->get_process( $_[0] )
    );

}

sub worker_started {
    my $self = shift;
    $self->notify_observers(
        'worker_started' => @_,
        $self->get_process( $_[1] )
    );

}

sub sig_child {
    my ( $self, $chld, $pid ) = @_;

# This won't scale worth shit, but will be replaced by a PID object coming out of MX::Workers
    my %list = reverse $self->process_list;
    $self->notify_observers( 'sig_child' => $pid => $list{$pid} );

    $self->remove_process( $list{$pid} );
}

no Moose;  # unimport Moose's keywords so they won't accidentally become methods
1;         # Magic true value required at end of module
__END__
