File Coverage

blib/lib/IO/Storm/Bolt.pm

Criterion	Covered	Total	%
statement	48	76	63.1
branch	10	22	45.4
condition	2	4	50.0
subroutine	9	14	64.2
pod	6	6	100.0
total	75	122	61.4

# ABSTRACT: The base class for all IO::Storm Bolts.

package IO::Storm::Bolt;
$IO::Storm::Bolt::VERSION = '0.15';
# Imports
use strict;
use warnings;
use v5.10;
use Try::Tiny;

# Setup Moo for object-oriented niceties
use Moo;
use namespace::clean;


extends 'IO::Storm::Component';

# A boolean indicating whether or not the bolt should automatically
# anchor emits to the incoming tuple ID. Tuple anchoring is how Storm
# provides reliability, you can read more about tuple anchoring in Storm's
# docs:
# https://storm.incubator.apache.org/documentation/Guaranteeing-message-processing.html#what-is-storms-reliability-api
has 'auto_anchor' => (
    is      => 'rw',
    default => 1
);

# A boolean indicating whether or not the bolt should automatically
# acknowledge tuples after ``process()`` is called.
has 'auto_ack' => (
    is      => 'rw',
    default => 1
);

# A boolean indicating whether or not the bolt should automatically fail
# tuples when an exception occurs when the ``process()`` method is called.
has 'auto_fail' => (
    is      => 'rw',
    default => 1
);

# Using a list so Bolt and subclasses can have more than one current_tup
has '_current_tups' => (
    is       => 'rw',
    default  => sub { [] },
    init_arg => undef
);


sub initialize {
    my ( $self, $storm_conf, $context ) = @_;
}


sub process {
    my ( $self, $tuple ) = @_;
}


sub emit ($$;$) {
    my ( $self, $tuple, $args ) = @_;

    $args = $args // {};
    my $msg = { command => 'emit', tuple => $tuple };

    my $anchors = [];
    if ( $self->auto_anchor ) {
        $anchors = $self->_current_tups // [];
    }
    unless ( defined( $args->{anchors} ) ) {
    	$args->{anchors} = $anchors;
    }

    my $a;
    for $a ( @{ $args->{anchors} } ) {
        if ( ref($a) eq "IO::Storm::Tuple" ) {
            $a = $a->id;
        }
        push( @$anchors, $a );
    }

    if ( defined( $args->{stream} ) ) {
        $msg->{stream} = $args->{stream};
    }

    if ( defined( $args->{direct_task} ) ) {
        $msg->{task} = $args->{direct_task};
    }

    $msg->{anchors} = $anchors;

    $self->send_message($msg);

    if ( defined $msg->{task} ) {
        return $msg->{task};
    }
    else {
        return $self->read_task_ids();
    }
}


sub ack {
    my ( $self, $tuple ) = @_;
    my $tup_id;
    if ( ref($tuple) eq "IO::Storm::Tuple" ) {
        $tup_id = $tuple->id;
    }
    else {
        $tup_id = $tuple;
    }
    $self->send_message( { command => 'ack', id => $tup_id } );
}


sub fail {
    my ( $self, $tuple ) = @_;
    my $tup_id;
    if ( ref($tuple) eq "IO::Storm::Tuple" ) {
        $tup_id = $tuple->id;
    }
    else {
        $tup_id = $tuple;
    }

    $self->send_message( { command => 'fail', id => $tup_id } );
}


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

    my ( $storm_conf, $context ) = $self->read_handshake();
    $self->_setup_component( $storm_conf, $context );
    $self->initialize( $storm_conf, $context );

    try {
        while (1) {
            $tup = $self->read_tuple();
            $self->_current_tups( [$tup] );
            $self->process($tup);
            if ( $self->auto_ack ) {
                $self->ack($tup);
            }

            # reset so that we don't accidentally fail the wrong tuples
            # if a successive call to read_tuple fails
            $self->_current_tups( [] );
        }
    }
    catch {
        my $error = $_;
        if ( scalar( @{ $self->_current_tups } ) == 1 ) {
            $tup = $self->_current_tups->[0];
            if ( $self->auto_fail ) {
                $self->fail($tup);
            }
        }
        $self->log("Bolt encountered exception: $_");
        die("Encounter exception in Bolt: $_");
    };
}

1;

__END__