File Coverage

blib/lib/IO/Storm/Spout.pm

Criterion	Covered	Total	%
statement	21	43	48.8
branch	2	16	12.5
condition			n/a
subroutine	6	11	54.5
pod	6	6	100.0
total	35	76	46.0

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

package IO::Storm::Spout;
$IO::Storm::Spout::VERSION = '0.17';
# Imports
use strict;
use warnings;
use v5.10;

use Moo;
use namespace::clean;

extends 'IO::Storm::Component';


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


sub ack {
    my ( $self, $id ) = @_;
}


sub fail {
    my ( $self, $id ) = @_;
}


sub next_tuple {
    my ($self) = @_;
}


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

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

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

    $self->send_message($msg);

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


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

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

    while (1) {
        my $msg = $self->read_command();
        if ( $msg->{command} eq 'next' ) {
            $self->next_tuple;
        }
        elsif ( $msg->{command} eq 'ack' ) {
            $self->ack( $msg->{id} );
        }
        elsif ( $msg->{command} eq 'fail' ) {
            $self->fail( $msg->{id} );
        }
        $self->sync();
    }
}

1;

__END__