<commit_before>package GRNOC::Simp::TSDS::Worker;

use strict;
use warnings;

use AnyEvent;
use Data::Dumper;
use List::MoreUtils qw(any);
use Moo;

use GRNOC::RabbitMQ::Client;
use GRNOC::RabbitMQ::Dispatcher;
use GRNOC::RabbitMQ::Method;
use GRNOC::Simp::TSDS;
use GRNOC::Simp::TSDS::Pusher;

=head2 public attributes
=over 12

=item logger
=item worker_name
=item rabbitmq
=item tsds_config
=item measurement_type
=item hosts
=item interval
=item composite
=item filter_name
=item filter_value
=item exclude_patterns
=item required_values

=back
=cut

has worker_name => (
    is       => 'ro',
    required => 1
);

has logger => (
    is       => 'rwp',
    required => 1
);

has rabbitmq => (
    is       => 'rwp',
    required => 1
);

has tsds_config => (
    is       => 'rwp',
    required => 1
);

has hosts => (
    is       => 'rwp',
    required => 1
);

has measurement_type => (
    is       => 'rwp',
    required => 1
);

has interval => (
    is       => 'rwp',
    required => 1
);

has composite => (
    is       => 'rwp',
    required => 1
);

has filter_name  => (is => 'rwp');
has filter_value => (is => 'rwp');

has exclude_patterns => (
    is      => 'rwp',
    default => sub { [] }
);

has required_values => (
    is      => 'rwp',
    default => sub { [] }
);

=head2 private attributes
=over 12

=item simp_client
=item tsds_pusher
=item poll_w
=item push_w
=item msg_list
=item cv
=item stop_me

=back
=cut

has simp_client => (is => 'rwp');
has tsds_pusher => (is => 'rwp');
has poll_w      => (is => 'rwp');
has push_w      => (is => 'rwp');
has cv          => (is => 'rwp');

has msg_list => (
    is      => 'rwp',
    default => sub { [] }
);

has stop_me => (
    is      => 'rwp',
    default => 0
);


=head2 start()
    This method is called by the Simp.TSDS master to start up the worker
    We loop on _run() where the event loop is entered and exited.
    The loop will only stop once the process is terminated.
    Once _run() is called, we only re-enter the loop when RMQ disconnects.
    This allows us to reconnect by restarting the worker processes.
=cut
sub start {
    
    my $self = shift;

    # Change the process name
    $0 = "simp_tsds(".$self->worker_name.")";

    # Create and set the logger
    $self->_set_logger(Log::Log4perl->get_logger('GRNOC.Simp.TSDS.Worker'));

    # This startup loop will catch when a worker exits the event loop without termination
    # In turn, we can handle connection errors here and/or restart the worker's main processes
    while (1) {

        $self->logger->info($self->worker_name." Starting...");
        $self->_run();

        if ($self->stop_me) {
            $self->logger->error($self->worker_name." Event loop ended, terminating");
            exit(0);
        }
        $self->logger->error($self->worker_name." Not connected to RabbitMQ, retrying after 1 interval (".$self->interval."s)");
        sleep $self->interval;
    }
        
}


=head2 run
    This kicks off the worker processes and enters the event loop.
    We only enter the event loop when RabbitMQ is connected.
    We lock the process on $self->cv->recv until $self->cv->send() is called.
    Unlocking only occurs when the process is terminated or RMQ disconnects.
    At that point, the process exits or _run() is called again by the start() loop.
=cut
sub _run {
    my ($self) = @_;

    # Connect and set the RabbitMQ Client
    # The worker will be locked on this step until Rabbitmq has connected
    $self->_setup_client();

    # We don't want to perform any other operations until RabbitMQ is connected
    # Once it is, we create the pusher and enter the event loop.<commit_msg>SCTASK0056065 Terminating workers will log to INFO level instead of ERROR<commit_after>package GRNOC::Simp::TSDS::Worker;

use strict;
use warnings;

use AnyEvent;
use Data::Dumper;
use List::MoreUtils qw(any);
use Moo;

use GRNOC::RabbitMQ::Client;
use GRNOC::RabbitMQ::Dispatcher;
use GRNOC::RabbitMQ::Method;
use GRNOC::Simp::TSDS;
use GRNOC::Simp::TSDS::Pusher;

=head2 public attributes
=over 12

=item logger
=item worker_name
=item rabbitmq
=item tsds_config
=item measurement_type
=item hosts
=item interval
=item composite
=item filter_name
=item filter_value
=item exclude_patterns
=item required_values

=back
=cut

has worker_name => (
    is       => 'ro',
    required => 1
);

has logger => (
    is       => 'rwp',
    required => 1
);

has rabbitmq => (
    is       => 'rwp',
    required => 1
);

has tsds_config => (
    is       => 'rwp',
    required => 1
);

has hosts => (
    is       => 'rwp',
    required => 1
);

has measurement_type => (
    is       => 'rwp',
    required => 1
);

has interval => (
    is       => 'rwp',
    required => 1
);

has composite => (
    is       => 'rwp',
    required => 1
);

has filter_name  => (is => 'rwp');
has filter_value => (is => 'rwp');

has exclude_patterns => (
    is      => 'rwp',
    default => sub { [] }
);

has required_values => (
    is      => 'rwp',
    default => sub { [] }
);

=head2 private attributes
=over 12

=item simp_client
=item tsds_pusher
=item poll_w
=item push_w
=item msg_list
=item cv
=item stop_me

=back
=cut

has simp_client => (is => 'rwp');
has tsds_pusher => (is => 'rwp');
has poll_w      => (is => 'rwp');
has push_w      => (is => 'rwp');
has cv          => (is => 'rwp');

has msg_list => (
    is      => 'rwp',
    default => sub { [] }
);

has stop_me => (
    is      => 'rwp',
    default => 0
);


=head2 start()
    This method is called by the Simp.TSDS master to start up the worker
    We loop on _run() where the event loop is entered and exited.
    The loop will only stop once the process is terminated.
    Once _run() is called, we only re-enter the loop when RMQ disconnects.
    This allows us to reconnect by restarting the worker processes.
=cut
sub start {
    
    my $self = shift;

    # Change the process name
    $0 = "simp_tsds(".$self->worker_name.")";

    # Create and set the logger
    $self->_set_logger(Log::Log4perl->get_logger('GRNOC.Simp.TSDS.Worker'));

    # This startup loop will catch when a worker exits the event loop without termination
    # In turn, we can handle connection errors here and/or restart the worker's main processes
    while (1) {

        $self->logger->info($self->worker_name." Starting...");
        $self->_run();

        if ($self->stop_me) {
            $self->logger->info($self->worker_name." Event loop ended, terminating");
            exit(0);
        }
        $self->logger->error($self->worker_name." Not connected to RabbitMQ, retrying after 1 interval (".$self->interval."s)");
        sleep $self->interval;
    }
        
}


=head2 run
    This kicks off the worker processes and enters the event loop.
    We only enter the event loop when RabbitMQ is connected.
    We lock the process on $self->cv->recv until $self->cv->send() is called.
    Unlocking only occurs when the process is terminated or RMQ disconnects.
    At that point, the process exits or _run() is called again by the start() loop.
=cut
sub _run {
    my ($self) = @_;

    # Connect and set the RabbitMQ Client
    # The worker will be locked on this step until Rabbitmq has connected
    $self->_setup_client();

    # We don't want to perform any other operations until RabbitMQ is connected
    # Once it is, we create the pusher and enter the event loop.
{'fraction_non_alphanumeric': 0.12900608519269777, 'fraction_numerical': 0.006896551724137931, 'mean_word_length': 2.806484817292846, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 145, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2242033', 'n_tokens_mistral': 2498, 'n_tokens_neox': 2387, 'n_words': 837}