view release on metacpan or  search on metacpan

lib/ETL/Yertl/Transform.pm  view on Meta::CPAN

#pod
#pod Transformations can be simple subroutines or full classes (inheriting
#pod from this class).
#pod
#pod =head2 Transform Object
#pod
#pod Create ad-hoc transform objects by passing in a C<transform_doc>
#pod callback. The callback receives two arguments: The transform object, and
#pod the document to transform.  The callback should return the transformed
#pod document (whether or not it is the same document modified in-place).
#pod
#pod =head2 Transform Class
#pod
#pod Create transform classes by inheriting from C<ETL::Yertl::Transform>.
#pod Subclasses can override the C<transform_doc> method to transform
#pod documents. This method receives the same arguments, returns the same
#pod values, sets C<$_>, and behaves exactly like the C<transform_doc>
#pod callback.
#pod
#pod =head2 Overloaded Operators
#pod
#pod Transforms can be chained together using the pipe (C<|>) operator. The
#pod result of the expression is the transform on the right side, for
#pod continued chaining.
#pod
#pod     my $xform1 = ETL::Yertl::Transform->new(
#pod         transform_doc => sub { ... },
#pod     );
#pod     my $xform2 = ETL::Yertl::Transform->new(
#pod         transform_doc => sub { ... },
#pod     );
#pod     my $xform3 = $xform1 | $xform2 | ETL::Yertl::Transform->new(
#pod         transform_doc => sub { ... },
#pod     );
#pod
#pod Transforms can receive sources using the C<< << >> operator with
#pod a L<ETL::Yertl::FormatStream> object. The result of the expression is
#pod the transform object, for continued chaining.
#pod
#pod     my $input = ETL::Yertl::FormatStream->new_for_stdin;
#pod     my $xform = ETL::Yertl::Transform->new(
#pod         transform_doc => sub { ... },
#pod     ) << $input;
#pod
#pod Transforms can receive destinations using the C<< >> >> operator with
#pod a L<ETL::Yertl::FormatStream> object. The result of the expression is
#pod the transform object, for continued chaining.
#pod
#pod     my $output = ETL::Yertl::FormatStream->new_for_stdout;
#pod     my $xform = ETL::Yertl::Transform->new(
#pod         transform_doc => sub { ... },
#pod     ) >> $output;
#pod
#pod =head1 SEE ALSO
#pod
#pod L<ETL::Yertl>, L<ETL::Yertl::FormatStream>
#pod
#pod =cut

use ETL::Yertl;
use Scalar::Util qw( weaken );
use Carp qw( croak );

use base 'IO::Async::Notifier';

# override pipe, <<, and >> to set on_read_doc and on_write_doc
# handlers appropriately
use overload
    '>>' => \&_set_output,
    '<<' => \&_set_input,
    '|' => \&_pipe,
    'fallback' => 1,
    ;

sub _set_output {
    my ( $self, $output ) = @_;
    # TODO: Allow Path::Tiny objects as output
    # TODO: Allow arrayrefs as output
    $self->configure( destination => $output );
    return $self;
}

sub _set_input {
    my ( $self, $input ) = @_;
    # TODO: Allow Path::Tiny objects as input
    # TODO: Allow arrayrefs as input
    $self->configure( source => $input );
    return $self;
}

sub _pipe {
    my ( $self, $other, $swap ) = @_;
    if ( $swap ) {
        ( $self, $other ) = ( $other, $self );
    }
    # TODO: Allow CODE ref as pipe to make up transform
    $other->configure( source => $self );
    return $other;
}

#pod =method new
#pod
#pod     my $xform = ETL::Yertl::Transform->new( %args );
#pod
#pod Create a new transform object. C<%args> is a hash with the following keys:
#pod
#pod =over
#pod
#pod =item source
#pod
#pod The source for documents. Can be a L<ETL::Yertl::FormatStream> or
#pod a L<ETL::Yertl::Transform> object. You do not need to specify this right
#pod away, but it is required for the transform to do useful work.
#pod
#pod =item destination
#pod
#pod (optional) A L<ETL::Yertl::FormatStream> object to write the documents
#pod to.  This can be an intermediate destination or the ultimate
#pod destination.  The last transform in a stream should have a destination.
#pod
#pod =item transform_doc
#pod
#pod A subref to transform the documents read from the source. The subref
#pod will receive two arguments: The transform object and the document to
#pod transform. It should return the transformed document.  The document to
#pod transform is also set as C<$_> for simpler transforms.
#pod
#pod =back
#pod
#pod =method configure
#pod
#pod     $xform->configure( %args );
#pod
#pod Configure this object. Takes the same arguments as the constructor,
#pod L</new>. This method allows updating any of the transform attributes
#pod later, so that transforms can be given new sources/destinations.
#pod
#pod =cut

sub configure {
    my ( $self, %args ) = @_;

    if ( $args{source} ) {
        # Register ourselves with the source
        my $source = $self->{source} = delete $args{source};
        weaken $self;
        $source->configure(
            on_doc => sub {
                my ( $source, $doc, $eof ) = @_;
                local $_ = $doc;
                my @docs = $self->invoke_event( transform_doc => $doc );
                # ; say "Writing docs from return: " . join ", ", @docs;
                # ; use Data::Dumper;
                # ; say STDERR Dumper( \@docs );
                $self->write( $_ ) for grep { $_ } @docs;
                return;
            },
            # XXX This probably needs to be done better:
            # * Users can't add their own handler to this event at all,
            #   making it more difficult to add Yertl streams to larger
            #   programs
            # * This requires on_read_eof to be called after all
            #   transforms are complete, which prevents cooperative
            #   multitasking by using `$self->loop->later` to defer
            #   execution of the transform_doc method/callback
            on_read_eof => sub {
                if ( my $dest = $self->{destination} ) {
                    if ( $dest->{write_handle} != \*STDOUT ) {
                        # Gracefully close the destination and then let
                        # anyone using us as a source know we're finished
                        $dest->configure( on_closed => sub {
                            $self->maybe_invoke_event( 'on_read_eof' );
                        } );
                        $dest->close_when_empty;
                        return;
                    }
                }
                # We emit our own on_read_eof event so downstream things
                # can clean up
                $self->maybe_invoke_event( 'on_read_eof' );
            },
        );
    }
    elsif ( !$self->{source} ) {
        # If we remove this requirement, we can configure objects and
        # add sources to them later, which could enable a bunch of fun
        # things like a pipe-based stream.
        # Without a source, this thing does nothing useful anyway, so
        # it'll be pretty obvious that something is broken.
        # croak "Expected a source";
    }

    if ( $args{destination} ) {
        $self->{destination} = delete $args{destination};
    }

    for my $event ( qw( transform_doc on_doc on_read_eof ) ) {
        $self->{ $event } = delete $args{ $event } if exists $args{ $event };
    }
    croak "Expected either a transform_doc callback or to be able to ->transform_doc"
        unless $self->can_event( 'transform_doc' );
    $self->{on_doc} ||= sub { }; # Default on_doc does nothing

    $self->SUPER::configure( %args );
}