File Coverage

blib/lib/Email/Sender/Server/Manager.pm

Criterion	Covered	Total	%
statement	33	144	22.9
branch	0	36	0.0
condition	0	8	0.0
subroutine	11	17	64.7
pod	4	5	80.0
total	48	210	22.8

# ABSTRACT: Email Server Manager

package Email::Sender::Server::Manager;

use Moo;
use utf8;

with 'Email::Sender::Server::Base';

use Carp 'confess';
use Data::Dumper 'Dumper';
use File::Copy 'move';
use File::Path 'mkpath';
use File::Spec::Functions 'splitdir', 'splitpath';
use File::Slurp 'write_file', 'read_file';
use Class::Date;

use Email::Sender::Server::Message;
use Email::Sender::Server::Worker;

$Data::Dumper::Useperl = 1;

our $VERSION = '1.000001'; # VERSION



has spawn => (
    is      => 'rw',
    default => 3
);


has workers => (
    is      => 'rw',
    default => sub {[
        # list of workers process IDs
    ]}
);


has workspace => (
    is      => 'rw',
    default => sub {
        shift->directory('queued');
    }
);


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

    my $queue = $self->directory('queued');

    confess "Couldn't find or access (write-to) the server's queue $queue"
        unless -d $queue && -w $queue;

    return $self;
}


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

    # re-queue imcompleted work orders

    opendir my $directory, $self->directory('worker');

    my @workers = grep { !/^\./ } readdir $directory;

    foreach my $worker (@workers) {
        my @filelist = @{ $self->message_filelist('worker', $worker) };

        foreach my $filepath (@filelist) {
            my $filename = $self->message_filename($filepath);

            move $filepath, $self->filepath('queued', $filename);

            unlink $filepath;
        }

        rmdir $self->directory('worker', $worker);
    }

    closedir $directory;

    # remove shutdown flag

    my $killer = $self->filepath('shutdown') ;

    unlink $killer if -e $killer;

    return $self;
}


sub create_config {
    my ($self) = @_;
    my $cfg = $self->filepath("config");

    unless (-e $cfg) {
        # write config file template
        write_file $cfg, {binmode => ':utf8'}, join "\n\n",
        '# see the Emailesque module for configuration options',
        'use utf8;',
        Dumper {
            message => {
                to          => undef,
                from        => undef,
                subject     => undef,
                cc          => undef,
                bcc         => undef,
                reply_to    => undef,
                message     => undef,
                type        => 'text'
            },
            transport => {
                Sendmail => {
                    sendmail => do {
                        my $path = `which sendmail`;
                           $path =~ s/[\r\n]//g;
                           $path || '/usr/sbin/sendmail';
                      }
                  }
              }
        };
    }

    return $self;
}


sub create_work {
    my ($self, %input) = @_;
    my $messenger = Email::Sender::Server::Message->new;

    while (my($name, $value) = each(%input)) {
        $messenger->$name($value);
    }

    my $message = $messenger->to_hash;

    if ($message) {
        my $filename = do {
            my $n = $message->{created}; $n =~ s/\W//g;
            my $x = join "-", $n, $$;
            my $i = do {
                my @chars = ('a'..'z','0'..'9');
                join '',
                    $chars[rand @chars],
                    $chars[rand @chars],
                    $chars[rand @chars],
                    $chars[rand @chars]
            };
            "$x" . "-" . $i . ".msg"
        };

        my $filepath = $self->filepath('queued', $filename);

        my $pid = fork();

        if ($pid == 0) {
            write_file $filepath, {binmode => ':utf8'}, join "\n\n",
            '# see the Emailesque module for configuration options',
            'use utf8;', Dumper $message;

            exit; # zombies will self-destruct
        }

        return $filepath if $pid;
    }

    return;
}


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

    # delegate to workers (minions)

    my $i = $self->spawn || 1;

    for(1..$i) {
        my $pid = fork;

        if ($pid == 0) {
            my $worker = Email::Sender::Server::Worker->new(id => $$);

            while (1) {
                foreach my $filepath (@{ $worker->message_filelist }) {
                    my $data = do $filepath;

                    my $next_filepath;

                    my $message = $worker->process_message($data);

                    if ($message && ref($message->result) =~ /success/i) {
                        # move message to passed

                        my $filename = $self->message_filename($filepath);

                        # segment storage in attempt to avoid filesystem
                        # directory size error

                        my @directory = ('passed');

                        push @directory, ($filename =~ /\W?(\d{4})(\d{4})/);
                        push @directory, $filename;

                        move $filepath,
                            $next_filepath = $self->filepath(@directory);
                    }
                    else {
                        # move message to failed

                        my $filename = $self->message_filename($filepath);

                        # segment storage in attempt to avoid filesystem
                        # directory size error

                        my @directory = ('failed');

                        push @directory, ($filename =~ /\W?(\d{4})(\d{4})/);
                        push @directory, $filename;

                        move $filepath,
                            $next_filepath = $self->filepath(@directory);
                    }

                    # log outcome (real quick)

                    if ($next_filepath) {
                        # ridiculously simple stupid logging

                        my @audit = ();

                        push @audit, "Date: "    . Class::Date::now->string;
                        push @audit, "To: "      . $message->to;
                        push @audit, "From: "    . $message->from;
                        push @audit, "Subject: " . $message->subject;
                        push @audit, "File: "    . $next_filepath;
                        push @audit, "Status: "  . ref $message->result;

                        if (ref($message->result) =~ /failure/i) {
                            if (ref($message->result) =~ /multi/i) {
                                push @audit, "Errors: " . join ", ",
                                    map { $_->message }
                                    $message->result->failure;
                            }
                            else {
                                push @audit, "Errors: " .
                                    $message->result->message;
                            }
                        }

                        write_file $self->filepath('delivery.log'),
                            {binmode => ':utf8', append  => 1},
                            join "\n", "", @audit;
                    }
                }

                last if $worker->stop_polling;
            }

            exit(0);
        }
        elsif ($pid) {
            push @{ $self->workers }, $pid;
        }
        else {
            die # die you!
        }
    }

    my $pid = fork;

    if ($pid == 0) {
        # delegate and process queued messages

        while (1) {
            foreach my $filepath (@{ $self->message_filelist }) {
                # find suitable worker bee (currently at-random)

                my $random_worker =
                    $self->workers->[rand(@{ $self->workers })];

                my $filename = $self->message_filename($filepath);

                if ($filename) {
                    move $filepath,
                        $self->filepath('worker', $random_worker, $filename);
                }
            }

            last if $self->stop_polling;
        }

        exit(0);
    }

    elsif (! $pid) {
        confess "Couldn't fork the manager's delegator, $!" ;
    }

    # always cleanup behind yourself !!!
    $SIG{INT} = sub { $self->cleanup; exit };

    waitpid $_, 0 for (@{$self->workers}, $pid); # blocking

    $self->cleanup; # cleanup server environment

    exit # return $self;
}

1;

__END__